fix(iota-core): select the fullnode submission driver per request from the current epoch store (#12520)

crates/iota-core/src/authority.rs

@@ -3383,16 +3383,27 @@ impl AuthorityState {
     /// It doesn't properly reconfigure the node, hence should be only used for
     /// testing.
     pub async fn reconfigure_for_testing(&self) {
+        self.reconfigure_for_testing_impl(None).await;
+    }
+
+    /// Like [`Self::reconfigure_for_testing`], but the next epoch uses the
+    /// given protocol config.
+    pub async fn reconfigure_for_testing_with_protocol_config(
+        &self,
+        protocol_config: ProtocolConfig,
+    ) {
+        self.reconfigure_for_testing_impl(Some(protocol_config))
+            .await;
+    }
+
+    async fn reconfigure_for_testing_impl(&self, protocol_config: Option<ProtocolConfig>) {
         let mut execution_lock = self.execution_lock_for_reconfiguration().await;
         let epoch_store = self.epoch_store_for_testing().clone();
-        let protocol_config = epoch_store.protocol_config().clone();
-        // The current protocol config used in the epoch store may have been overridden
-        // and diverged from the protocol config definitions. That override may
-        // have now been dropped when the initial guard was dropped. We reapply
-        // the override before creating the new epoch store, to make sure that
-        // the new epoch store has the same protocol config as the current one.
-        // Since this is for testing only, we mostly like to keep the protocol config
-        // the same across epochs.
+        // Default to the epoch store's config, whose override guard may have
+        // been dropped. Read it under the lock so config and epoch store are
+        // one snapshot.
+        let protocol_config =
+            protocol_config.unwrap_or_else(|| epoch_store.protocol_config().clone());
         let _guard =
             ProtocolConfig::apply_overrides_for_testing(move |_, _| protocol_config.clone());
         let new_epoch_store = epoch_store.new_at_next_epoch_for_testing(

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

@@ -10,7 +10,7 @@ mod transaction_submitter;
 
 use std::{
     net::SocketAddr,
-    sync::Arc,
+    sync::{Arc, Weak},
     time::{Duration, Instant},
 };
 
@@ -54,7 +54,7 @@ pub trait AuthorityAggregatorUpdatable<A>: Send + Sync + 'static {
     fn update_authority_aggregator(&self, new_authorities: Arc<AuthorityAggregator<A>>);
 }
 
-use iota_config::node::NodeConfig;
+use iota_config::validator_client_monitor_config::ValidatorClientMonitorConfig;
 
 /// Options for submitting a transaction.
 #[derive(Clone, Default, Debug)]
@@ -93,6 +93,9 @@ pub struct TransactionDriver<A> {
     submitter: TransactionSubmitter,
     certifier: EffectsCertifier,
     client_monitor: Arc<ValidatorClientMonitor>,
+    /// Whether the P-COOL flow is enabled in the current epoch; latency-ping
+    /// and health-check rounds are skipped while false.
+    pcool_flow_enabled: Arc<dyn Fn() -> bool + Send + Sync>,
 }
 
 impl<A> TransactionDriver<A>
@@ -103,17 +106,15 @@ where
         authority_aggregator: Arc<AuthorityAggregator<A>>,
         reconfig_observer: Arc<dyn ReconfigObserver<A> + Sync + Send>,
         metrics: Arc<TransactionDriverMetrics>,
-        node_config: Option<&NodeConfig>,
+        validator_client_monitor_config: Option<ValidatorClientMonitorConfig>,
         client_metrics: Arc<ValidatorClientMetrics>,
+        pcool_flow_enabled: Arc<dyn Fn() -> bool + Send + Sync>,
     ) -> Arc<Self> {
         let shared_swap = Arc::new(ArcSwap::new(authority_aggregator));
 
-        // Extract validator client monitor config from NodeConfig or use default
-        let monitor_config = node_config
-            .and_then(|nc| nc.validator_client_monitor_config.clone())
-            .unwrap_or_default();
+        let monitor_config = validator_client_monitor_config.unwrap_or_default();
         let client_monitor = Arc::new(ValidatorClientMonitor::new(monitor_config, client_metrics));
-        client_monitor.spawn_health_checks(&shared_swap);
+        client_monitor.spawn_health_checks(&shared_swap, pcool_flow_enabled.clone());
 
         let driver = Arc::new(Self {
             authority_aggregator: shared_swap,
@@ -122,11 +123,11 @@ where
             submitter: TransactionSubmitter::new(metrics.clone()),
             certifier: EffectsCertifier::new(metrics),
             client_monitor,
+            pcool_flow_enabled,
         });
 
-        let driver_clone = driver.clone();
-
-        spawn_logged_monitored_task!(Self::run_latency_checks(driver_clone));
+        let driver_weak = Arc::downgrade(&driver);
+        spawn_logged_monitored_task!(Self::run_latency_checks(driver_weak));
 
         driver.enable_reconfig(reconfig_observer);
         driver
@@ -405,7 +406,7 @@ where
 
     // Runs a background task to send ping transactions to all validators to perform
     // latency checks.
-    async fn run_latency_checks(self: Arc<Self>) {
+    async fn run_latency_checks(driver: Weak<Self>) {
         const INTERVAL_BETWEEN_RUNS: Duration = Duration::from_secs(15);
         const MAX_JITTER: Duration = Duration::from_secs(10);
         const PING_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
@@ -416,9 +417,19 @@ where
         loop {
             interval.tick().await;
 
+            // Weak, so this detached task cannot keep the driver alive.
+            let Some(driver) = driver.upgrade() else {
+                break;
+            };
+
+            // Validators reject pings while the P-COOL flow is disabled.
+            if !(driver.pcool_flow_enabled)() {
+                continue;
+            }
+
             let mut tasks = JoinSet::new();
 
-            Self::ping(self.clone(), &mut tasks, MAX_JITTER, PING_REQUEST_TIMEOUT);
+            Self::ping(driver, &mut tasks, MAX_JITTER, PING_REQUEST_TIMEOUT);
 
             while let Some(result) = tasks.join_next().await {
                 if let Err(e) = result {