[MFP] add the block list for tx submission for transaction driver (#24161)

## Description 

Adding support for the a transaction submission block list.

For the configuration we can not have both an allow list and a block
list - those are mutually exclusive configurations.

## Test plan 

CI

---

## Release notes

Check each box that your changes affect. If none of the boxes relate to
your changes, release notes aren't required.

For each box you select, include information after the relevant heading
that describes the impact of your changes that a user might notice and
any actions they must take to implement updates.

- [ ] Protocol: 
- [ ] Nodes (Validators and Full nodes): 
- [ ] gRPC:
- [ ] JSON-RPC: 
- [ ] GraphQL: 
- [ ] CLI: 
- [ ] Rust SDK:

Author: akichidis

crates/sui-config/src/node.rs

@@ -235,6 +235,11 @@ pub struct TransactionDriverConfig {
     #[serde(default, skip_serializing_if = "Vec::is_empty")]
     pub allowed_submission_validators: Vec<String>,
 
+    /// The list of validators that are blocked from submitting block transactions to (via the transaction driver).
+    /// Each entry is a validator display name.
+    #[serde(default, skip_serializing_if = "Vec::is_empty")]
+    pub blocked_submission_validators: Vec<String>,
+
     /// Enable early transaction validation before submission to consensus.
     /// This checks for non-retriable errors (like old object versions) and rejects
     /// transactions early to provide fast feedback to clients.
@@ -247,6 +252,7 @@ impl Default for TransactionDriverConfig {
     fn default() -> Self {
         Self {
             allowed_submission_validators: vec![],
+            blocked_submission_validators: vec![],
             enable_early_validation: true,
         }
     }

crates/sui-core/src/transaction_driver/effects_certifier.rs

@@ -99,8 +99,13 @@ impl EffectsCertifier {
             }
         };
 
-        let mut retrier =
-            RequestRetrier::new(authority_aggregator, client_monitor, tx_type, vec![]);
+        let mut retrier = RequestRetrier::new(
+            authority_aggregator,
+            client_monitor,
+            tx_type,
+            vec![],
+            vec![],
+        );
         let ping_type = get_ping_type(&tx_digest, tx_type);
 
         // Setting this to None at first because if the full effects are already provided,

crates/sui-core/src/transaction_driver/mod.rs

@@ -55,6 +55,10 @@ pub struct SubmitTransactionOptions {
     /// When submitting a transaction, only the validators in the allowed validator list can be used to submit the transaction to.
     /// When the allowed validator list is empty, any validator can be used.
     pub allowed_validators: Vec<String>,
+
+    /// When submitting a transaction, the validators in the blocked validator list cannot be used to submit the transaction to.
+    /// When the blocked validator list is empty, no restrictions are applied.
+    pub blocked_validators: Vec<String>,
 }
 
 #[derive(Clone, Debug)]

crates/sui-core/src/transaction_driver/request_retrier.rs

@@ -29,6 +29,9 @@ const SELECT_LATENCY_DELTA: f64 = 0.02;
 /// When an `allowed_validators` is provided, only the validators in the list will be used to submit the transaction to.
 /// When the allowed validator list is empty, any validator can be used an then the validators are selected based on their scores.
 ///
+/// When a `blocked_validators` is provided, the validators in the list cannot be used to submit the transaction to.
+/// When the blocked validator list is empty, no restrictions are applied.
+///
 /// This component helps to manager this retry pattern.
 pub(crate) struct RequestRetrier<A: Clone> {
     ranked_clients: VecDeque<(AuthorityName, Arc<SafeClient<A>>)>,
@@ -42,6 +45,7 @@ impl<A: Clone> RequestRetrier<A> {
         client_monitor: &Arc<ValidatorClientMonitor<A>>,
         tx_type: TxType,
         allowed_validators: Vec<String>,
+        blocked_validators: Vec<String>,
     ) -> Self {
         let ranked_validators = client_monitor.select_shuffled_preferred_validators(
             &auth_agg.committee,
@@ -50,11 +54,14 @@ impl<A: Clone> RequestRetrier<A> {
         );
         let ranked_clients = ranked_validators
             .into_iter()
-            .filter(|name| {
-                let display_name = auth_agg.get_display_name(name);
-                allowed_validators.is_empty() || allowed_validators.contains(&display_name)
+            .map(|name| (name, auth_agg.get_display_name(&name)))
+            .filter(|(_name, display_name)| {
+                allowed_validators.is_empty() || allowed_validators.contains(display_name)
+            })
+            .filter(|(_name, display_name)| {
+                blocked_validators.is_empty() || !blocked_validators.contains(display_name)
             })
-            .filter_map(|name| {
+            .filter_map(|(name, _display_name)| {
                 // There is not guarantee that the `name` are in the `auth_agg.authority_clients` if those are coming from the list
                 // of `allowed_validators`, as the provided `auth_agg` might have been updated with a new committee that doesn't contain the validator in question.
                 auth_agg
@@ -167,8 +174,13 @@ mod tests {
     async fn test_next_target() {
         let auth_agg = Arc::new(get_authority_aggregator(4));
         let client_monitor = Arc::new(ValidatorClientMonitor::new_for_test(auth_agg.clone()));
-        let mut retrier =
-            RequestRetrier::new(&auth_agg, &client_monitor, TxType::SingleWriter, vec![]);
+        let mut retrier = RequestRetrier::new(
+            &auth_agg,
+            &client_monitor,
+            TxType::SingleWriter,
+            vec![],
+            vec![],
+        );
 
         for name in auth_agg.committee.names() {
             retrier.next_target().unwrap();
@@ -215,6 +227,7 @@ mod tests {
                 &client_monitor,
                 TxType::SingleWriter,
                 allowed_validators,
+                vec![],
             );
 
             // Should only have 1 remaining client (the known validator)
@@ -234,13 +247,48 @@ mod tests {
                 &client_monitor,
                 TxType::SingleWriter,
                 allowed_validators,
+                vec![],
             );
 
             // Should have no remaining clients since none of the allowed validators exist
             assert_eq!(retrier.ranked_clients.len(), 0);
         }
     }
 
+    #[tokio::test]
+    async fn test_blocked_validators() {
+        let auth_agg = Arc::new(get_authority_aggregator(4));
+        let client_monitor = Arc::new(ValidatorClientMonitor::new_for_test(auth_agg.clone()));
+
+        // Create a list of validators that should be blocked and never picked up by the retrier.
+        let blocked_validators = auth_agg
+            .committee
+            .names()
+            .take(3)
+            .copied()
+            .collect::<Vec<_>>();
+        let blocked_display_names = blocked_validators
+            .iter()
+            .map(|name| auth_agg.get_display_name(name))
+            .collect::<Vec<_>>();
+
+        // Only the last validator will be picked up.
+        let allowed_validator = auth_agg.committee.names().nth(3).unwrap();
+
+        let mut retrier = RequestRetrier::new(
+            &auth_agg,
+            &client_monitor,
+            TxType::SingleWriter,
+            vec![],
+            blocked_display_names,
+        );
+
+        // The last validator will be picked up.
+        assert_eq!(retrier.next_target().unwrap().0, *allowed_validator);
+        // No more validators will be picked up.
+        assert!(retrier.next_target().is_err());
+    }
+
     #[tokio::test]
     async fn test_add_error() {
         let auth_agg = Arc::new(get_authority_aggregator(4));
@@ -249,8 +297,13 @@ mod tests {
         // Add retriable errors.
         {
             let client_monitor = Arc::new(ValidatorClientMonitor::new_for_test(auth_agg.clone()));
-            let mut retrier =
-                RequestRetrier::new(&auth_agg, &client_monitor, TxType::SingleWriter, vec![]);
+            let mut retrier = RequestRetrier::new(
+                &auth_agg,
+                &client_monitor,
+                TxType::SingleWriter,
+                vec![],
+                vec![],
+            );
 
             // 25% stake.
             retrier
@@ -286,8 +339,13 @@ mod tests {
         // Add mix of retriable and non-retriable errors.
         {
             let client_monitor = Arc::new(ValidatorClientMonitor::new_for_test(auth_agg.clone()));
-            let mut retrier =
-                RequestRetrier::new(&auth_agg, &client_monitor, TxType::SingleWriter, vec![]);
+            let mut retrier = RequestRetrier::new(
+                &auth_agg,
+                &client_monitor,
+                TxType::SingleWriter,
+                vec![],
+                vec![],
+            );
 
             // 25% stake retriable error.
             retrier

crates/sui-core/src/transaction_driver/transaction_submitter.rs

@@ -71,6 +71,7 @@ impl TransactionSubmitter {
             client_monitor,
             tx_type,
             options.allowed_validators.clone(),
+            options.blocked_validators.clone(),
         );
 
         let ping_label = if request.ping_type.is_some() {

crates/sui-core/src/transaction_orchestrator.rs

@@ -79,6 +79,7 @@ pub struct TransactionOrchestrator<A: Clone> {
     transaction_driver: Option<Arc<TransactionDriver<A>>>,
     td_percentage: u8,
     td_allowed_submission_list: Vec<String>,
+    td_blocked_submission_list: Vec<String>,
     enable_early_validation: bool,
 }
 
@@ -174,6 +175,19 @@ where
             .map(|config| config.allowed_submission_validators.clone())
             .unwrap_or_default();
 
+        let td_blocked_submission_list = node_config
+            .transaction_driver_config
+            .as_ref()
+            .map(|config| config.blocked_submission_validators.clone())
+            .unwrap_or_default();
+
+        if !td_allowed_submission_list.is_empty() && !td_blocked_submission_list.is_empty() {
+            panic!(
+                "Both allowed and blocked submission lists are set, this is not allowed, {:?} {:?}",
+                td_allowed_submission_list, td_blocked_submission_list
+            );
+        }
+
         let enable_early_validation = node_config
             .transaction_driver_config
             .as_ref()
@@ -190,6 +204,7 @@ where
             transaction_driver,
             td_percentage,
             td_allowed_submission_list,
+            td_blocked_submission_list,
             enable_early_validation,
         }
     }
@@ -671,6 +686,7 @@ where
                 SubmitTransactionOptions {
                     forwarded_client_addr: client_addr,
                     allowed_validators: self.td_allowed_submission_list.clone(),
+                    blocked_validators: self.td_blocked_submission_list.clone(),
                 },
                 timeout_duration,
             )