diff --git a/crates/iota-core/src/authority.rs b/crates/iota-core/src/authority.rs index 0dcbff4a43a7..0b613e8df949 100644 --- a/crates/iota-core/src/authority.rs +++ b/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) { 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( diff --git a/crates/iota-core/src/transaction_driver/mod.rs b/crates/iota-core/src/transaction_driver/mod.rs index d93828a496d0..9a0cc5021edb 100644 --- a/crates/iota-core/src/transaction_driver/mod.rs +++ b/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: Send + Sync + 'static { fn update_authority_aggregator(&self, new_authorities: Arc>); } -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 { submitter: TransactionSubmitter, certifier: EffectsCertifier, client_monitor: Arc, + /// 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 bool + Send + Sync>, } impl TransactionDriver @@ -103,17 +106,15 @@ where authority_aggregator: Arc>, reconfig_observer: Arc + Sync + Send>, metrics: Arc, - node_config: Option<&NodeConfig>, + validator_client_monitor_config: Option, client_metrics: Arc, + pcool_flow_enabled: Arc bool + Send + Sync>, ) -> Arc { 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) { + async fn run_latency_checks(driver: Weak) { 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 { diff --git a/crates/iota-core/src/transaction_orchestrator.rs b/crates/iota-core/src/transaction_orchestrator.rs index 1f3340e6febf..4c2bc9154d20 100644 --- a/crates/iota-core/src/transaction_orchestrator.rs +++ b/crates/iota-core/src/transaction_orchestrator.rs @@ -2,9 +2,10 @@ // Modifications Copyright (c) 2024 IOTA Stiftung // SPDX-License-Identifier: Apache-2.0 -// Transaction Orchestrator is a Node component that utilizes Quorum Driver (or -// optionally TransactionDriver) to submit transactions to validators for -// finality, and proactively executes finalized transactions locally. +// Transaction Orchestrator is a Node component that utilizes Quorum Driver or +// TransactionDriver (selected per request by the P-COOL protocol flag) to +// submit transactions to validators for finality, and proactively executes +// finalized transactions locally. use std::{ collections::{BTreeMap, HashMap, hash_map::Entry}, @@ -84,9 +85,8 @@ const LOCAL_EXECUTION_TIMEOUT: Duration = Duration::from_secs(10); const WAIT_FOR_FINALITY_TIMEOUT: Duration = Duration::from_secs(30); -/// The submission flow used to drive transactions to finality. Exactly one -/// flow is active, selected by the P-COOL protocol flag at construction -/// time. +/// The submission flow for a transaction, selected per request by the +/// epoch's P-COOL flag. enum Driver { /// Certificate-based flow (P-COOL disabled). Quorum(Arc>), @@ -98,9 +98,18 @@ enum Driver { /// and TransactionDriver for submitting transactions to validators for /// finality. It adds inflight deduplication, waiting for local execution, /// recovery, and epoch change handling. +/// +/// The epoch's P-COOL flag selects the flow serving a request. The +/// TransactionDriver always exists, so it tracks epochs before the flag +/// enables it. The QuorumDriver exists only when the node booted with +/// P-COOL disabled and ran WAL recovery then; after a rollback a node +/// booted under P-COOL must be restarted. pub struct TransactionOrchestrator { - driver: Driver, + quorum_driver: Option>>, + transaction_driver: Arc>, validator_state: Arc, + /// Handle to the pending-tx-log cleanup loop; present only with the + /// quorum driver. _local_executor_handle: Option>, pending_tx_log: Arc, /// Digests currently being driven to finality by the TransactionDriver; @@ -124,34 +133,20 @@ impl TransactionOrchestrator { prometheus_registry: &Registry, node_config: Option<&NodeConfig>, ) -> Self { - // Check protocol config to determine if P-COOL flow is enabled - let epoch_store = validator_state.load_epoch_store_one_call_per_task(); - let use_transaction_driver = epoch_store.protocol_config().enable_pcool_flow(); - - // Create TransactionDriver reconfig observer only if P-COOL is enabled - let td_reconfig_observer = if use_transaction_driver { - Some(TdOnsiteReconfigObserver::new( - reconfig_channel.resubscribe(), - validator_state.get_object_cache_reader().clone(), - validator_state.clone_committee_store(), - validators.safe_client_metrics_base.clone(), - )) - } else { - None - }; + let td_reconfig_observer = TdOnsiteReconfigObserver::new( + reconfig_channel.resubscribe(), + validator_state.get_object_cache_reader().clone(), + validator_state.clone_committee_store(), + validators.safe_client_metrics_base.clone(), + ); - // Create QuorumDriver reconfig observer only if P-COOL is NOT enabled - let qd_reconfig_observer = if !use_transaction_driver { - Some(OnsiteReconfigObserver::new( - reconfig_channel.resubscribe(), - validator_state.get_object_cache_reader().clone(), - validator_state.clone_committee_store(), - validators.safe_client_metrics_base.clone(), - validators.metrics.deref().clone(), - )) - } else { - None - }; + let qd_reconfig_observer = OnsiteReconfigObserver::new( + reconfig_channel.resubscribe(), + validator_state.get_object_cache_reader().clone(), + validator_state.clone_committee_store(), + validators.safe_client_metrics_base.clone(), + validators.metrics.deref().clone(), + ); TransactionOrchestrator::new( validators, @@ -176,11 +171,10 @@ where validator_state: Arc, parent_path: &Path, prometheus_registry: &Registry, - reconfig_observer: Option, - td_reconfig_observer: Option, + reconfig_observer: OnsiteReconfigObserver, + td_reconfig_observer: TdOnsiteReconfigObserver, node_config: Option<&NodeConfig>, ) -> Self { - // Check protocol config to determine if P-COOL flow is enabled let epoch_store = validator_state.load_epoch_store_one_call_per_task(); let use_transaction_driver = epoch_store.protocol_config().enable_pcool_flow(); @@ -190,47 +184,58 @@ where parent_path.join("fullnode_pending_transactions"), )); - let (driver, _local_executor_handle) = if !use_transaction_driver { - let qd_metrics = Arc::new(QuorumDriverMetrics::new(prometheus_registry)); - let reconfig_observer = Arc::new( - reconfig_observer - .expect("QuorumDriver reconfig observer required when P-COOL is disabled"), - ); - let handler = Arc::new( - QuorumDriverHandlerBuilder::new(validators, qd_metrics) + // Registered for both flows even when the quorum driver is not + // built, so metric presence does not depend on the boot mode. + let quorum_driver_metrics = Arc::new(QuorumDriverMetrics::new(prometheus_registry)); + let transaction_driver_metrics = + Arc::new(TransactionDriverMetrics::new(prometheus_registry)); + let client_metrics = Arc::new(ValidatorClientMetrics::new(prometheus_registry)); + + let (quorum_driver, _local_executor_handle) = if use_transaction_driver { + (None, None) + } else { + let quorum_driver = Arc::new( + QuorumDriverHandlerBuilder::new(validators.clone(), quorum_driver_metrics) .with_notifier(notifier.clone()) - .with_reconfig_observer(reconfig_observer) + .with_reconfig_observer(Arc::new(reconfig_observer)) .start(), ); - let effects_receiver = handler.subscribe_to_effects(); + // The cleanup loop must exist before WAL recovery runs, so a + // recovered transaction cannot complete before its receiver + // exists. + let effects_receiver = quorum_driver.subscribe_to_effects(); let pending_tx_log_clone = pending_tx_log.clone(); let local_executor_handle = spawn_monitored_task!(async move { Self::loop_pending_transaction_log(effects_receiver, pending_tx_log_clone).await; }); - // Pending-transaction recovery is QuorumDriver-only; the - // TransactionDriver goes directly to consensus and tracks no - // pending certificates. - Self::schedule_txes_in_log(pending_tx_log.clone(), handler.clone()); - (Driver::Quorum(handler), Some(local_executor_handle)) - } else { - let td_metrics = Arc::new(TransactionDriverMetrics::new(prometheus_registry)); - let client_metrics = Arc::new(ValidatorClientMetrics::new(prometheus_registry)); - let observer = td_reconfig_observer - .expect("TransactionDriver reconfig observer required when P-COOL is enabled"); - ( - Driver::Transaction(TransactionDriver::new( - validators, - Arc::new(observer), - td_metrics, - node_config, - client_metrics, - )), - None, - ) + Self::schedule_txes_in_log(pending_tx_log.clone(), quorum_driver.clone()); + (Some(quorum_driver), Some(local_executor_handle)) }; + // `Weak` so detached driver tasks cannot pin the authority state. + let pcool_flow_enabled: Arc bool + Send + Sync> = { + let validator_state = Arc::downgrade(&validator_state); + Arc::new(move || { + validator_state.upgrade().is_some_and(|state| { + state + .load_epoch_store_one_call_per_task() + .protocol_config() + .enable_pcool_flow() + }) + }) + }; + let transaction_driver = TransactionDriver::new( + validators, + Arc::new(td_reconfig_observer), + transaction_driver_metrics, + node_config.and_then(|config| config.validator_client_monitor_config.clone()), + client_metrics, + pcool_flow_enabled, + ); + Self { - driver, + quorum_driver, + transaction_driver, validator_state, _local_executor_handle, pending_tx_log, @@ -245,6 +250,34 @@ impl TransactionOrchestrator where A: AuthorityAPI + Send + Sync + 'static + Clone, { + /// Returns the flow selected by `epoch_store`'s P-COOL flag. Call with + /// the request's own epoch store snapshot so the flag and the submission + /// see the same epoch. Errors when the quorum driver is selected on a + /// node that booted under P-COOL: such a node never ran WAL recovery and + /// must be restarted. + fn select_driver( + &self, + epoch_store: &AuthorityPerEpochStore, + ) -> Result, QuorumDriverError> { + if epoch_store.protocol_config().enable_pcool_flow() { + return Ok(Driver::Transaction(self.transaction_driver.clone())); + } + self.quorum_driver + .clone() + .map(Driver::Quorum) + .ok_or_else(|| { + error!( + "This fullnode started while P-COOL was enabled and must be restarted to \ + serve the certificate-based flow" + ); + QuorumDriverError::QuorumDriverInternal(IotaError::UnsupportedFeature { + error: "this fullnode started while P-COOL was enabled and must be \ + restarted to serve the certificate-based flow" + .to_string(), + }) + }) + } + #[instrument(name = "tx_orchestrator_execute_transaction_block", level = "trace", skip_all, fields( tx_digest = ?request.transaction.digest(), @@ -305,57 +338,56 @@ where request_type, ExecuteTransactionRequestType::WaitForLocalExecution ); - let (mut response, seq) = match (&self.driver, wait_for_local_execution) { - (Driver::Transaction(td), true) => { - let td = td.clone(); - let in_flight_transactions = self.in_flight_transactions.clone(); - let validator_state = self.validator_state.clone(); - let metrics = self.metrics.clone(); - // Detached so a client disconnect (this future dropped) does - // not cancel a submission that may already be in consensus; - // the task drives the transaction to finality on its own. - join_submission_task(spawn_monitored_task!(Self::submit_with_checkpoint_race( - td, - in_flight_transactions, - validator_state, - metrics, - request, - client_addr, - tx_digest, - ))) - .await? - } - (Driver::Transaction(td), false) => { - let td = td.clone(); - let in_flight_transactions = self.in_flight_transactions.clone(); - let validator_state = self.validator_state.clone(); - // Detached for the same reason as above. - let result = join_submission_task(spawn_monitored_task!( - Self::submit_with_transaction_driver( + let (mut response, seq) = + match (self.select_driver(&epoch_store)?, wait_for_local_execution) { + (Driver::Transaction(td), true) => { + let in_flight_transactions = self.in_flight_transactions.clone(); + let validator_state = self.validator_state.clone(); + let metrics = self.metrics.clone(); + // Detached so a client disconnect (this future dropped) does + // not cancel a submission that may already be in consensus; + // the task drives the transaction to finality on its own. + join_submission_task(spawn_monitored_task!(Self::submit_with_checkpoint_race( td, in_flight_transactions, validator_state, + metrics, request, client_addr, - false, - ) - )) - .await?; - (Some(result), None) - } - (Driver::Quorum(qd), _) => { - let qd_resp = self - .execute_transaction_impl( - qd, - &epoch_store, - request, - transaction.clone(), - client_addr, - ) + tx_digest, + ))) + .await? + } + (Driver::Transaction(td), false) => { + let in_flight_transactions = self.in_flight_transactions.clone(); + let validator_state = self.validator_state.clone(); + // Detached for the same reason as above. + let result = join_submission_task(spawn_monitored_task!( + Self::submit_with_transaction_driver( + td, + in_flight_transactions, + validator_state, + request, + client_addr, + false, + ) + )) .await?; - (Some(quorum_driver_response_to_v1(qd_resp)), None) - } - }; + (Some(result), None) + } + (Driver::Quorum(qd), _) => { + let qd_resp = self + .execute_transaction_impl( + &qd, + &epoch_store, + request, + transaction.clone(), + client_addr, + ) + .await?; + (Some(quorum_driver_response_to_v1(qd_resp)), None) + } + }; // `needs_cache_rebuild` is derived from finality, not caller intent: // the QD fallback path returns `Certified` and a duplicate @@ -659,9 +691,8 @@ where .validity_check(&epoch_store.tx_validity_check_context()) .map_err(QuorumDriverError::InvalidTransaction)?; - match &self.driver { + match self.select_driver(&epoch_store)? { Driver::Transaction(td) => { - let td = td.clone(); let in_flight_transactions = self.in_flight_transactions.clone(); let validator_state = self.validator_state.clone(); // v1 does not do an internal wait; callers (e.g. the gRPC @@ -683,7 +714,7 @@ where } Driver::Quorum(qd) => { let qd_resp = self - .execute_transaction_impl(qd, &epoch_store, request, transaction, client_addr) + .execute_transaction_impl(&qd, &epoch_store, request, transaction, client_addr) .await?; Ok(quorum_driver_response_to_v1(qd_resp)) } @@ -1236,39 +1267,46 @@ where } } - /// Returns the quorum driver, or `None` under the P-COOL flow. + /// Returns the quorum driver, or `None` when the node booted under + /// P-COOL. Test-only: submissions must go through the per-request + /// driver selection. + #[cfg(any(test, feature = "test-utils"))] pub fn quorum_driver(&self) -> Option<&Arc>> { - match &self.driver { - Driver::Quorum(handler) => Some(handler), - Driver::Transaction(_) => None, - } + self.quorum_driver.as_ref() } - /// Returns the quorum driver, or `None` under the P-COOL flow. + /// Owned variant of [`Self::quorum_driver`]. + #[cfg(any(test, feature = "test-utils"))] pub fn clone_quorum_driver(&self) -> Option>> { - self.quorum_driver().cloned() + self.quorum_driver.clone() } - /// Returns the transaction driver, or `None` when the P-COOL flow is - /// disabled. - pub fn transaction_driver(&self) -> Option<&Arc>> { - match &self.driver { - Driver::Quorum(_) => None, - Driver::Transaction(td) => Some(td), - } + /// Returns the transaction driver's aggregator; it always exists and its + /// reconfig observer keeps it current. + pub fn clone_authority_aggregator(&self) -> Arc> { + self.transaction_driver.authority_aggregator().load_full() } - /// Returns the authority aggregator of the active driver. - pub fn clone_authority_aggregator(&self) -> Arc> { - match &self.driver { - Driver::Quorum(qd) => qd.authority_aggregator().load_full(), - Driver::Transaction(td) => td.authority_aggregator().load_full(), + /// Returns an effects receiver only while the quorum driver is the + /// currently selected flow and this node can serve it; the P-COOL flow + /// has no effects broadcast. + pub fn subscribe_to_effects_queue(&self) -> Option> { + let epoch_store = self.validator_state.load_epoch_store_one_call_per_task(); + if epoch_store.protocol_config().enable_pcool_flow() { + return None; } + self.quorum_driver + .as_ref() + .map(|quorum_driver| quorum_driver.subscribe_to_effects()) } - /// Returns `None` under the P-COOL flow, which has no effects broadcast. - pub fn subscribe_to_effects_queue(&self) -> Option> { - self.quorum_driver().map(|qd| qd.subscribe_to_effects()) + /// Runs driver selection for `epoch_store` and reports its outcome. + #[cfg(any(test, feature = "test-utils"))] + pub fn select_driver_for_testing( + &self, + epoch_store: &AuthorityPerEpochStore, + ) -> Result<(), QuorumDriverError> { + self.select_driver(epoch_store).map(|_| ()) } fn update_metrics( @@ -1301,11 +1339,11 @@ where pending_tx_log: Arc, quorum_driver: Arc>, ) { + if std::env::var("SKIP_LOADING_FROM_PENDING_TX_LOG").is_ok() { + info!("Skipping loading pending transactions from pending_tx_log."); + return; + } spawn_logged_monitored_task!(async move { - if std::env::var("SKIP_LOADING_FROM_PENDING_TX_LOG").is_ok() { - info!("Skipping loading pending transactions from pending_tx_log."); - return; - } let pending_txes = pending_tx_log .load_all_pending_transactions() .expect("failed to load all pending transactions"); @@ -1996,4 +2034,97 @@ mod tests { // The digest can be driven again once the entry is gone. let _guard = acquire_driving(&in_flight, tx_digest); } + + async fn build_orchestrator_with_pcool( + enable_pcool: bool, + ) -> ( + Arc, + TransactionOrchestrator, + tempfile::TempDir, + tokio::sync::broadcast::Sender, + ) { + use iota_protocol_config::{Chain, ProtocolConfig, ProtocolVersion}; + + use crate::{ + authority::test_authority_builder::TestAuthorityBuilder, + authority_aggregator::AuthorityAggregatorBuilder, + }; + + telemetry_subscribers::init_for_testing(); + let network_config = + iota_swarm_config::network_config_builder::ConfigBuilder::new_with_temp_dir().build(); + + let mut protocol_config = + ProtocolConfig::get_for_version(ProtocolVersion::MAX, Chain::Unknown); + protocol_config.set_enable_pcool_flow_for_testing(enable_pcool); + let state = TestAuthorityBuilder::new() + .with_network_config(&network_config, 0) + .with_protocol_config(protocol_config) + .build() + .await; + + let (aggregator, _clients) = + AuthorityAggregatorBuilder::from_genesis(&network_config.genesis) + .build_network_clients(); + let (reconfig_tx, reconfig_rx) = tokio::sync::broadcast::channel(16); + let tempdir = tempfile::tempdir().unwrap(); + let orchestrator = TransactionOrchestrator::new_with_auth_aggregator( + Arc::new(aggregator), + state.clone(), + reconfig_rx, + tempdir.path(), + &Registry::new(), + None, + ); + (state, orchestrator, tempdir, reconfig_tx) + } + + /// A flag-off boot builds the quorum driver and runs WAL recovery at + /// construction. + #[tokio::test(flavor = "multi_thread")] + async fn qd_recovery_eager_on_flag_off_boot() { + let (state, orchestrator, _tempdir, _reconfig_tx) = + build_orchestrator_with_pcool(false).await; + assert!(orchestrator.quorum_driver().is_some()); + assert!( + orchestrator + .select_driver_for_testing(&state.epoch_store_for_testing()) + .is_ok() + ); + } + + /// A node booted under P-COOL has no quorum driver. After a rollback it + /// must reject quorum-driver selection until restarted. + #[tokio::test(flavor = "multi_thread")] + async fn flag_on_boot_rejects_selection_after_rollback() { + use iota_protocol_config::{Chain, ProtocolConfig, ProtocolVersion}; + + let (state, orchestrator, _tempdir, _reconfig_tx) = + build_orchestrator_with_pcool(true).await; + assert!(orchestrator.quorum_driver().is_none()); + + // Selection under the flag serves the P-COOL flow. + assert!( + orchestrator + .select_driver_for_testing(&state.epoch_store_for_testing()) + .is_ok() + ); + + // Epoch 1 with P-COOL off: selection and the effects queue must both + // report the missing quorum driver. + let mut protocol_config = + ProtocolConfig::get_for_version(ProtocolVersion::MAX, Chain::Unknown); + protocol_config.set_enable_pcool_flow_for_testing(false); + state + .reconfigure_for_testing_with_protocol_config(protocol_config) + .await; + let epoch_store = state.epoch_store_for_testing(); + assert_eq!(epoch_store.epoch(), 1); + + assert!(matches!( + orchestrator.select_driver_for_testing(&epoch_store), + Err(QuorumDriverError::QuorumDriverInternal(_)) + )); + assert!(orchestrator.subscribe_to_effects_queue().is_none()); + } } diff --git a/crates/iota-core/src/validator_client_monitor/monitor.rs b/crates/iota-core/src/validator_client_monitor/monitor.rs index c9269fa853bd..d1fd1b3c219c 100644 --- a/crates/iota-core/src/validator_client_monitor/monitor.rs +++ b/crates/iota-core/src/validator_client_monitor/monitor.rs @@ -47,9 +47,12 @@ impl ValidatorClientMonitor { } } + /// `enabled` is checked per round; rounds are skipped while it reports + /// false. pub fn spawn_health_checks( self: &Arc, authority_aggregator: &Arc>>, + enabled: Arc bool + Send + Sync>, ) -> JoinHandle<()> { let period = self.config.health_check_interval; let monitor = Arc::downgrade(self); @@ -57,7 +60,7 @@ impl ValidatorClientMonitor { // weak pointers allow health check task break early once shared arc objects are // dropped tokio::spawn(async move { - Self::run_health_checks(monitor, authority_aggregator, period).await; + Self::run_health_checks(monitor, authority_aggregator, period, enabled).await; }) } @@ -117,18 +120,24 @@ impl ValidatorClientMonitor { monitor: Weak, authority_aggregator: Weak>>, period: Duration, + enabled: Arc bool + Send + Sync>, ) { let mut interval = interval(period); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { interval.tick().await; + // Upgrade first: a permanently false `enabled` must not keep + // the task alive past these exits. let Some(monitor) = monitor.upgrade() else { break; }; let Some(authority_agg) = authority_aggregator.upgrade() else { break; }; + if !enabled() { + continue; + } let authority_agg = authority_agg.load(); let mut tasks = monitor.spawn_health_checks_tasks(&*authority_agg); drop(authority_agg); diff --git a/crates/iota-e2e-tests/tests/onsite_reconfig_observer_tests.rs b/crates/iota-e2e-tests/tests/onsite_reconfig_observer_tests.rs index 8800f22e3251..f538ac5aeb0b 100644 --- a/crates/iota-e2e-tests/tests/onsite_reconfig_observer_tests.rs +++ b/crates/iota-e2e-tests/tests/onsite_reconfig_observer_tests.rs @@ -27,7 +27,7 @@ async fn test_onsite_reconfig_observer_basic() { node.transaction_orchestrator() .unwrap() .clone_quorum_driver() - .expect("quorum driver should be present when P-COOL is disabled") + .expect("quorum driver exists on a flag-off boot") }); assert_eq!(qd.current_epoch(), 0); let rx = fullnode.with(|node| node.subscribe_to_epoch_change()); @@ -52,7 +52,7 @@ async fn test_onsite_reconfig_observer_basic() { node.transaction_orchestrator() .unwrap() .clone_quorum_driver() - .expect("quorum driver should be present when P-COOL is disabled") + .expect("quorum driver exists on a flag-off boot") }); assert_eq!(qd.current_epoch(), 1); assert_eq!( diff --git a/crates/iota-e2e-tests/tests/transaction_orchestrator_tests.rs b/crates/iota-e2e-tests/tests/transaction_orchestrator_tests.rs index 0783fb21d121..0efcd7e08a53 100644 --- a/crates/iota-e2e-tests/tests/transaction_orchestrator_tests.rs +++ b/crates/iota-e2e-tests/tests/transaction_orchestrator_tests.rs @@ -26,11 +26,13 @@ use iota_test_transaction_builder::{ use iota_types::{ effects::{TransactionEffectsAPI, TransactionEffectsExt}, error::IotaError, + iota_system_state::IotaSystemStateTrait, quorum_driver_types::{ EffectsFinalityInfo, ExecuteTransactionRequestType, ExecuteTransactionRequestV1, ExecuteTransactionResponseV1, FinalizedEffects, IsTransactionExecutedLocally, QuorumDriverError, }, + supported_protocol_versions::SupportedProtocolVersions, transaction::{TransactionAPI, TransactionEnvelope}, }; use test_cluster::{TestClusterBuilder, override_pcool_flow}; @@ -61,7 +63,7 @@ async fn test_blocking_execution() -> Result<(), anyhow::Error> { let digest = *txn.digest(); orchestrator .quorum_driver() - .expect("quorum driver should be present when P-COOL is disabled") + .expect("quorum driver exists on a flag-off boot") .submit_transaction_no_ticket( ExecuteTransactionRequestV1::new(txn), Some(make_socket_addr()), @@ -201,7 +203,7 @@ async fn test_transaction_orchestrator_reconfig() { node.transaction_orchestrator() .unwrap() .quorum_driver() - .expect("quorum driver should be present when P-COOL is disabled") + .expect("quorum driver exists on a flag-off boot") .current_epoch() }); assert_eq!(epoch, 0); @@ -218,7 +220,7 @@ async fn test_transaction_orchestrator_reconfig() { node.transaction_orchestrator() .unwrap() .quorum_driver() - .expect("quorum driver should be present when P-COOL is disabled") + .expect("quorum driver exists on a flag-off boot") .current_epoch() }); if epoch == 1 { @@ -399,6 +401,93 @@ async fn test_wait_for_local_execution_across_epoch_boundary() { } } +/// The driver must follow `enable_pcool_flow` across the upgrade that flips +/// it, without a fullnode restart: boot at v31 (flag off, QuorumDriver), +/// upgrade to v32 (flag on), and the same orchestrator instance must serve +/// both sides, post-upgrade via the TransactionDriver. +/// +/// No `override_pcool_flow`: the env override would pin the flag for every +/// version. +#[sim_test] +async fn test_orchestrator_follows_pcool_flag_across_protocol_upgrade() { + telemetry_subscribers::init_for_testing(); + const START: u64 = 31; + const FINISH: u64 = 32; + + let test_cluster = TestClusterBuilder::new() + .with_protocol_version(START.into()) + .with_supported_protocol_versions(SupportedProtocolVersions::new_for_testing(START, FINISH)) + .with_epoch_duration_ms(20_000) + .build() + .await; + + let orchestrator = test_cluster + .fullnode_handle + .iota_node + .with(|node| node.transaction_orchestrator().unwrap()); + let pcool_enabled = || { + test_cluster.fullnode_handle.iota_node.with(|node| { + node.state() + .epoch_store_for_testing() + .protocol_config() + .enable_pcool_flow() + }) + }; + + // Boot state: flag off, quorum driver built and recovered. + assert!(!pcool_enabled()); + assert!(orchestrator.quorum_driver().is_some()); + + let tx = make_transfer_iota_transaction(&test_cluster.wallet, None, None).await; + let (response, _) = execute_with_orchestrator( + &orchestrator, + tx, + ExecuteTransactionRequestType::WaitForLocalExecution, + ) + .await + .expect("pre-upgrade submission must succeed on the certificate flow"); + assert!(matches!( + response.effects.finality_info, + EffectsFinalityInfo::Certified(_) + )); + + // All validators support FINISH, so the upgrade lands at the first epoch + // boundary. + let system_state = test_cluster.wait_for_protocol_version(FINISH.into()).await; + let flip_epoch = system_state.epoch(); + test_cluster.wait_for_epoch_all_nodes(flip_epoch).await; + assert!(pcool_enabled()); + + // Same orchestrator, no restart: the request must use the + // TransactionDriver, since validators now reject the certificate flow. + let tx = make_transfer_iota_transaction(&test_cluster.wallet, None, None).await; + let (response, executed_locally) = execute_with_orchestrator( + &orchestrator, + tx, + ExecuteTransactionRequestType::WaitForLocalExecution, + ) + .await + .expect("post-upgrade submission must succeed without a fullnode restart"); + assert!(executed_locally); + match response.effects.finality_info { + EffectsFinalityInfo::Checkpointed(epoch, _) => assert!(epoch >= flip_epoch), + EffectsFinalityInfo::QuorumExecuted(epoch) => assert!(epoch >= flip_epoch), + other => panic!("expected TransactionDriver finality, got {other:?}"), + } + + // The TransactionDriver must keep tracking reconfiguration. + test_cluster.wait_for_epoch(Some(flip_epoch + 1)).await; + test_cluster.wait_for_epoch_all_nodes(flip_epoch + 1).await; + let tx = make_transfer_iota_transaction(&test_cluster.wallet, None, None).await; + execute_with_orchestrator( + &orchestrator, + tx, + ExecuteTransactionRequestType::WaitForLocalExecution, + ) + .await + .expect("submission must succeed one epoch after the flip"); +} + async fn execute_with_orchestrator( orchestrator: &TransactionOrchestrator, tx: TransactionEnvelope, diff --git a/crates/iota-node/src/lib.rs b/crates/iota-node/src/lib.rs index 3e4f22864650..b56ef672b688 100644 --- a/crates/iota-node/src/lib.rs +++ b/crates/iota-node/src/lib.rs @@ -1731,11 +1731,9 @@ impl IotaNode { // self.state.db() // } - /// Clone the AuthorityAggregator currently used by this node's - /// transaction orchestrator, if the node is a fullnode. After reconfig, - /// the active driver builds a new AuthorityAggregator. The caller - /// of this function will mostly likely want to call this again - /// to get a fresh one. + /// Clone an AuthorityAggregator from the transaction orchestrator, if + /// this is a fullnode. The snapshot goes stale after an epoch change; + /// call again for a fresh one. pub fn clone_authority_aggregator( &self, ) -> Option>> { @@ -1750,6 +1748,8 @@ impl IotaNode { self.transaction_orchestrator.clone() } + /// Subscribe to the quorum driver's effects stream; errors while the + /// quorum driver is not the currently served flow on this node. pub fn subscribe_to_transaction_orchestrator_effects( &self, ) -> Result> { @@ -1759,7 +1759,12 @@ impl IotaNode { anyhow::anyhow!("Transaction Orchestrator is not enabled in this node.") })? .subscribe_to_effects_queue() - .ok_or_else(|| anyhow::anyhow!("Effects queue is not available under the P-COOL flow.")) + .ok_or_else(|| { + anyhow::anyhow!( + "Effects queue is not available: the quorum driver is not the currently \ + served flow on this node." + ) + }) } /// This function awaits the completion of checkpoint execution of the