Skip to content

fix(node): gracefully clean up iota-node validator components #6831

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

fix(node): gracefully clean up iota-node validator components #6831

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
7082ec3
Revert "fix(iota-node): unregister and re-register consensus adapter …
semenov-vladyslav May 9, 2025
6e8f409
fix: gracefully shutdown iota-node validator components
semenov-vladyslav May 9, 2025
f7d54b1
fix: review comments
muXxer May 22, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
110 changes: 0 additions & 110 deletions crates/iota-core/src/consensus_adapter.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -193,60 +193,6 @@ impl ConsensusAdapterMetrics {
pub fn new_test() -> Self {
Self::new(&Registry::default())
}

pub fn unregister(&self, registry: &Registry) {
registry
.unregister(Box::new(self.sequencing_certificate_attempt.clone()))
.expect("sequencing_certificate_attempt is in registry");
registry
.unregister(Box::new(self.sequencing_certificate_success.clone()))
.expect("sequencing_certificate_success is in registry");
registry
.unregister(Box::new(self.sequencing_certificate_failures.clone()))
.expect("sequencing_certificate_failures is in registry");
registry
.unregister(Box::new(self.sequencing_certificate_status.clone()))
.expect("sequencing_certificate_status is in registry");
registry
.unregister(Box::new(self.sequencing_certificate_inflight.clone()))
.expect("sequencing_certificate_inflight is in registry");
registry
.unregister(Box::new(self.sequencing_acknowledge_latency.clone()))
.expect("sequencing_acknowledge_latency is in registry");
registry
.unregister(Box::new(self.sequencing_certificate_latency.clone()))
.expect("sequencing_certificate_latency is in registry");
registry
.unregister(Box::new(
self.sequencing_certificate_authority_position.clone(),
))
.expect("sequencing_certificate_authority_position is in registry");
registry
.unregister(Box::new(
self.sequencing_certificate_positions_moved.clone(),
))
.expect("sequencing_certificate_positions_moved is in registry");
registry
.unregister(Box::new(
self.sequencing_certificate_preceding_disconnected.clone(),
))
.expect("sequencing_certificate_preceding_disconnected is in registry");
registry
.unregister(Box::new(self.sequencing_certificate_processed.clone()))
.expect("sequencing_certificate_processed is in registry");
registry
.unregister(Box::new(self.sequencing_in_flight_semaphore_wait.clone()))
.expect("sequencing_in_flight_semaphore_wait is in registry");
registry
.unregister(Box::new(self.sequencing_in_flight_submissions.clone()))
.expect("sequencing_in_flight_submissions is in registry");
registry
.unregister(Box::new(self.sequencing_estimated_latency.clone()))
.expect("sequencing_estimated_latency is in registry");
registry
.unregister(Box::new(self.sequencing_resubmission_interval_ms.clone()))
.expect("sequencing_resubmission_interval_ms is in registry");
}
}

pub type BlockStatusReceiver = oneshot::Receiver<BlockStatus>;
Expand Down Expand Up @@ -398,10 +344,6 @@ impl ConsensusAdapter {
}
}

pub fn unregister_consensus_adapter_metrics(&self, registry: &Registry) {
self.metrics.unregister(registry);
}

fn await_submit_delay(
&self,
committee: &Committee,
Expand Down Expand Up @@ -1302,7 +1244,6 @@ mod adapter_tests {
committee::Committee,
crypto::{AuthorityKeyPair, AuthorityPublicKeyBytes, get_key_pair_from_rng},
};
use prometheus::Registry;
use rand::{Rng, SeedableRng, rngs::StdRng};

use super::position_submit_certificate;
Expand Down Expand Up @@ -1412,55 +1353,4 @@ mod adapter_tests {
assert!(zero_found);
}
}

#[tokio::test]
#[should_panic]
async fn test_reregister_consensus_adapter_metrics() {
let registry = Registry::new();
// create metric the first time
let _metrics = ConsensusAdapterMetrics::new(&registry);
// create a new metric in the same registry without unregistering
// should panic
let _metrics = ConsensusAdapterMetrics::new(&registry);
}

#[tokio::test]
async fn test_unregister_consensus_adapter_metrics() {
let registry = Registry::new();

// create metric the first time
let metrics = ConsensusAdapterMetrics::new(&registry);
// use metric
metrics
.sequencing_certificate_attempt
.with_label_values(&["tx"])
.inc_by(1);
assert_eq!(
1,
metrics
.sequencing_certificate_attempt
.with_label_values(&["tx"])
.get()
);
// should not panic
metrics.unregister(&registry);
// metric can safely be used unregistered
metrics
.sequencing_certificate_attempt
.with_label_values(&["tx"])
.inc_by(1);

// create a new metric in the same registry
let metrics = ConsensusAdapterMetrics::new(&registry);
// it's fresh
assert_eq!(
0,
metrics
.sequencing_certificate_attempt
.with_label_values(&["tx"])
.get()
);
// and can be unregistered
metrics.unregister(&registry);
}
}
5 changes: 2 additions & 3 deletions crates/iota-core/src/consensus_manager/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -97,11 +97,10 @@ impl ConsensusManager {
node_config: &NodeConfig,
consensus_config: &ConsensusConfig,
registry_service: &RegistryService,
metrics_registry: &Registry,
consensus_client: Arc<UpdatableConsensusClient>,
) -> Self {
let metrics = Arc::new(ConsensusManagerMetrics::new(
&registry_service.default_registry(),
));
let metrics = Arc::new(ConsensusManagerMetrics::new(metrics_registry));
let mysticeti_client = Arc::new(LazyMysticetiClient::new());
let mysticeti_manager = ProtocolManager::new_mysticeti(
node_config,
Expand Down
94 changes: 94 additions & 0 deletions crates/iota-e2e-tests/tests/reconfiguration_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -671,6 +671,100 @@ async fn test_reconfig_with_committee_change_basic() {
});
}

#[sim_test]
async fn test_reconfig_with_same_validator() {
use iota_swarm_config::genesis_config::{AccountConfig, DEFAULT_GAS_AMOUNT, GenesisConfig};
use iota_types::{
crypto::{AuthorityPublicKeyBytes, KeypairTraits},
governance::MIN_VALIDATOR_JOINING_STAKE_NANOS,
};
use rand::{SeedableRng, rngs::StdRng};

// ValidatorGenesisConfig doesn't impl Clone
// generate the same config with the same rng seed
let build_node_config =
|| ValidatorGenesisConfigBuilder::new().build(&mut StdRng::seed_from_u64(0));

// the node that will re-join committee
let node_config = build_node_config();
let node_name: AuthorityPublicKeyBytes = node_config.authority_key_pair.public().into();
let node_address = (&node_config.account_key_pair.public()).into();
let mut node_handle = None;

// add coins to the node at the genesis to avoid dealing with faucet
let mut genesis_config = GenesisConfig::default();
genesis_config
.accounts
.extend(std::iter::once(AccountConfig {
address: Some(node_address),
gas_amounts: vec![
DEFAULT_GAS_AMOUNT,
MIN_VALIDATOR_JOINING_STAKE_NANOS,
DEFAULT_GAS_AMOUNT,
MIN_VALIDATOR_JOINING_STAKE_NANOS,
],
}));

// create test cluster with 4 default validators
let mut test_cluster = TestClusterBuilder::new()
.with_num_validators(4)
.set_genesis_config(genesis_config)
.build()
.await;

// whether node is in committee in a corresponding epoch
// test a few join/leave/join cases
let node_schedule = [true, true, false, false, true, false, true];
// the node initially is not in the committee
let mut was_in_committee = false;

let mut epoch = 0;
for is_in_committee in node_schedule {
if !was_in_committee && is_in_committee {
// add node to committee
execute_add_validator_transactions(&test_cluster, &build_node_config()).await;
}
if was_in_committee && !is_in_committee {
// remove node from committee
execute_remove_validator_tx(&test_cluster, node_handle.as_ref().unwrap()).await;
}

// reconfiguration
test_cluster.force_new_epoch().await;
epoch += 1;

// check that node has joined or left the committee
test_cluster.fullnode_handle.iota_node.with(|node| {
assert_eq!(
is_in_committee,
node.state()
.epoch_store_for_testing()
.committee()
.authority_exists(&node_name)
);
});

if node_handle.is_none() {
// spawn node if not already
node_handle = Some(test_cluster.spawn_new_validator(build_node_config()).await);
}

// sync nodes
test_cluster.wait_for_epoch_all_nodes(epoch).await;

// the running node acknowledges being or not being a committee member
node_handle.as_ref().unwrap().with(|node| {
assert_eq!(
is_in_committee,
node.state()
.is_validator(&node.state().epoch_store_for_testing())
);
});

was_in_committee = is_in_committee;
}
}

#[sim_test]
async fn test_reconfig_with_committee_change_stress() {
do_test_reconfig_with_committee_change_stress().await;
Expand Down
28 changes: 1 addition & 27 deletions crates/iota-metrics/src/histogram.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,7 @@ use std::{
use futures::FutureExt;
use parking_lot::Mutex;
use prometheus::{
IntCounterVec, IntGaugeVec, Registry, opts, register_int_counter_vec_with_registry,
IntCounterVec, IntGaugeVec, Registry, register_int_counter_vec_with_registry,
register_int_gauge_vec_with_registry,
};
use tokio::{
Expand Down Expand Up @@ -191,32 +191,6 @@ impl HistogramVec {
channel: self.channel.clone(),
}
}

// HistogramVec uses asynchronous model to report metrics which makes
// it difficult to unregister counters in the usual manner. Here we
// re-create counters so that their `desc()`s match the ones created by
// HistogramVec and remove them from the registry. Counters can be safely
// unregistered even if they are still in use.
pub fn unregister(name: &str, desc: &str, labels: &[&str], registry: &Registry) {
let sum_name = format!("{}_sum", name);
let count_name = format!("{}_count", name);

let sum = IntCounterVec::new(opts!(sum_name, desc), labels).unwrap();
registry
.unregister(Box::new(sum))
.unwrap_or_else(|_| panic!("{}_sum counter is in prometheus registry", name));

let count = IntCounterVec::new(opts!(count_name, desc), labels).unwrap();
registry
.unregister(Box::new(count))
.unwrap_or_else(|_| panic!("{}_count counter is in prometheus registry", name));

let labels: Vec<_> = labels.iter().cloned().chain(["pct"]).collect();
let gauge = IntGaugeVec::new(opts!(name, desc), &labels).unwrap();
registry
.unregister(Box::new(gauge))
.unwrap_or_else(|_| panic!("{} gauge is in prometheus registry", name));
}
}

impl HistogramLabelsInner {
Expand Down
Loading
Loading