diff --git a/Cargo.lock b/Cargo.lock index b0868fe48e91..fccc6d5d03b8 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7566,6 +7566,7 @@ version = "0.9.0" dependencies = [ "assert_matches", "criterion", + "hex", "ic-artifact-pool", "ic-btc-replica-types", "ic-config", @@ -7646,6 +7647,7 @@ dependencies = [ "ic-artifact-pool", "ic-canonical-state", "ic-canonical-state-tree-hash", + "ic-config", "ic-consensus-mocks", "ic-consensus-utils", "ic-crypto-test-utils-crypto-returning-ok", @@ -7657,9 +7659,11 @@ dependencies = [ "ic-metrics", "ic-registry-subnet-type", "ic-replicated-state", + "ic-test-artifact-pool", "ic-test-utilities", "ic-test-utilities-consensus", "ic-test-utilities-logger", + "ic-test-utilities-registry", "ic-test-utilities-types", "ic-types", "ic-types-test-utils", @@ -7721,6 +7725,7 @@ dependencies = [ "phantom_newtype", "prost", "slog", + "thiserror 2.0.18", ] [[package]] diff --git a/rs/consensus/BUILD.bazel b/rs/consensus/BUILD.bazel index 056fa3ccded6..8d6c7400e2c1 100644 --- a/rs/consensus/BUILD.bazel +++ b/rs/consensus/BUILD.bazel @@ -153,6 +153,7 @@ rust_test( "//rs/types/types", "//rs/types/types_test_utils", "@crate_index//:assert_matches", + "@crate_index//:hex", "@crate_index//:mockall", "@crate_index//:num-traits", "@crate_index//:prometheus", @@ -226,6 +227,7 @@ rust_test( "//rs/types/types", "//rs/types/types_test_utils", "@crate_index//:assert_matches", + "@crate_index//:hex", "@crate_index//:mockall", "@crate_index//:num-traits", "@crate_index//:prometheus", @@ -298,6 +300,7 @@ rust_test( "//rs/types/types", "//rs/types/types_test_utils", "@crate_index//:assert_matches", + "@crate_index//:hex", "@crate_index//:mockall", "@crate_index//:num-traits", "@crate_index//:prometheus", @@ -354,6 +357,7 @@ rust_test( "//rs/types/types", "//rs/types/types_test_utils", "@crate_index//:assert_matches", + "@crate_index//:hex", "@crate_index//:mockall", "@crate_index//:num-traits", "@crate_index//:prometheus", diff --git a/rs/consensus/Cargo.toml b/rs/consensus/Cargo.toml index 8ed60035a754..0c7707d222e9 100644 --- a/rs/consensus/Cargo.toml +++ b/rs/consensus/Cargo.toml @@ -40,6 +40,7 @@ strum_macros = { workspace = true } [dev-dependencies] assert_matches = { workspace = true } criterion = { workspace = true } +hex = { workspace = true } ic-artifact-pool = { path = "../artifact_pool" } ic-btc-replica-types = { path = "../bitcoin/replica_types" } ic-config = { path = "../config" } diff --git a/rs/consensus/certification/BUILD.bazel b/rs/consensus/certification/BUILD.bazel index 513fc9ca0388..574c6f36bc72 100644 --- a/rs/consensus/certification/BUILD.bazel +++ b/rs/consensus/certification/BUILD.bazel @@ -39,6 +39,7 @@ rust_test( "//rs/artifact_pool", "//rs/canonical_state", "//rs/canonical_state/tree_hash", + "//rs/config", "//rs/consensus/mocks", "//rs/consensus/utils", "//rs/crypto/test_utils/crypto_returning_ok", @@ -51,8 +52,10 @@ rust_test( "//rs/registry/subnet_type", "//rs/replicated_state", "//rs/test_utilities", + "//rs/test_utilities/artifact_pool", "//rs/test_utilities/consensus", "//rs/test_utilities/logger", + "//rs/test_utilities/registry", "//rs/test_utilities/types", "//rs/types/types", "//rs/types/types_test_utils", diff --git a/rs/consensus/certification/Cargo.toml b/rs/consensus/certification/Cargo.toml index ea337a24d4a0..6163c3b3bfff 100644 --- a/rs/consensus/certification/Cargo.toml +++ b/rs/consensus/certification/Cargo.toml @@ -25,11 +25,14 @@ slog = { workspace = true } assert_matches = { workspace = true } ic-artifact-pool = { path = "../../artifact_pool" } ic-consensus-mocks = { path = "../mocks" } +ic-config = { path = "../../config" } ic-crypto-test-utils-crypto-returning-ok = { path = "../../crypto/test_utils/crypto_returning_ok" } ic-registry-subnet-type = { path = "../../registry/subnet_type" } +ic-test-artifact-pool = { path = "../../test_utilities/artifact_pool" } ic-test-utilities = { path = "../../test_utilities" } ic-test-utilities-consensus = { path = "../../test_utilities/consensus" } ic-test-utilities-logger = { path = "../../test_utilities/logger" } +ic-test-utilities-registry = { path = "../../test_utilities/registry" } ic-test-utilities-types = { path = "../../test_utilities/types" } ic-types-test-utils = { path = "../../types/types_test_utils" } mockall = { workspace = true } diff --git a/rs/consensus/certification/src/certifier.rs b/rs/consensus/certification/src/certifier.rs index a02af098f609..a14ac83919cd 100644 --- a/rs/consensus/certification/src/certifier.rs +++ b/rs/consensus/certification/src/certifier.rs @@ -4,6 +4,7 @@ use ic_canonical_state_tree_hash::lazy_tree::materialize::materialize; use ic_consensus_utils::{ MINIMUM_CHAIN_LENGTH, active_high_threshold_nidkg_id, aggregate, bouncer_metrics::BouncerMetrics, membership::Membership, registry_version_at_height, + subnet_splitting_status_at_height, }; use ic_crypto_tree_hash::{Witness, recompute_digest}; use ic_interfaces::{ @@ -14,7 +15,7 @@ use ic_interfaces::{ }; use ic_interfaces_registry::RegistryClient; use ic_interfaces_state_manager::{StateHashMetadata, StateManager}; -use ic_logger::{ReplicaLogger, debug, error, trace}; +use ic_logger::{ReplicaLogger, debug, error, info, trace, warn}; use ic_metrics::{MetricsRegistry, buckets::decimal_buckets}; use ic_replicated_state::ReplicatedState; use ic_types::{ @@ -25,6 +26,7 @@ use ic_types::{ certification::{ Certification, CertificationContent, CertificationMessage, CertificationShare, }, + dkg::{PostSplitArgs, SubnetSplittingStatus}, }, crypto::{CryptoHash, Signed}, replica_config::ReplicaConfig, @@ -340,6 +342,18 @@ impl CertifierImpl { .shares_at_height(state_hash_metadata.height) .all(|share| share.signed.signature.signer != self.replica_config.node_id) }) + // Filter out all heights, where the subnet splitting is taking place + .filter(|state_hash_metadata| { + self.should_skip_due_to_subnet_splitting(state_hash_metadata.height) + .inspect_err(|err| { + warn!( + self.log, + "Failed to check the subnet splitting status: {err}. \ + Skipping creation of the certificate share" + ) + }) + .is_ok_and(|should_skip| !should_skip) + }) .cloned() .filter_map(|state_hash_metadata| { let content = CertificationContent::new(state_hash_metadata.hash); @@ -479,6 +493,32 @@ impl CertifierImpl { let registry_version = registry_version_at_height(self.consensus_pool_cache.as_ref(), certification.height)?; + // If a subnet splitting is taking place, we need to skip validating certifications (and + // shares). In particular because after a split, before replicas get restarted, they are + // still under the same P2P network and can gossip certifications for states of different + // subnets. + match self.should_skip_due_to_subnet_splitting(certification.height) { + Ok(true) => { + info!( + every_n_seconds => 30, + self.log, + "Skipping the validation of a certification at height {} because a \ + subnet splitting is taking place", + certification.height + ); + return None; + } + Ok(false) => {} + Err(err) => { + warn!( + self.log, + "Failed to check the subnet splitting status: {err}. \ + Skipping validation of the certificate" + ); + return None; + } + } + // check if the certification is indeed valid for the specified height. If // not, we consider the certification invalid. if let Err(e) = validate_height_witness( @@ -517,6 +557,32 @@ impl CertifierImpl { let msg = CertificationMessage::CertificationShare(share.clone()); let content = &share.signed.content; + // If a subnet splitting is taking place, we need to skip validating certifications (and + // shares). In particular because after a split, before replicas get restarted, they are + // still under the same P2P network and can gossip certifications for states of different + // subnets. + match self.should_skip_due_to_subnet_splitting(share.height) { + Ok(true) => { + info!( + every_n_seconds => 30, + self.log, + "Skipping the validation of a certification share at height {} because a \ + subnet splitting is taking place", + share.height + ); + return None; + } + Ok(false) => {} + Err(err) => { + warn!( + self.log, + "Failed to check the subnet splitting status: {err}. \ + Skipping validation of the certificate share" + ); + return None; + } + } + // If the share has an invalid content or does not belong to the // committee if let Err(e) = validate_height_witness( @@ -582,6 +648,24 @@ impl CertifierImpl { } } } + + /// Checks if we should skip the creation and/or validation of certifications/shares + /// at the given height, due to an ongoing subnet splitting. + fn should_skip_due_to_subnet_splitting(&self, height: Height) -> Result { + match subnet_splitting_status_at_height(self.consensus_pool_cache.as_ref(), height) { + None => Err(format!( + "Missing finalized summary block for height {height}" + )), + Some(SubnetSplittingStatus::NotScheduled) => Ok(false), + // Don't produce certifications in the dkg interval where the subnet splitting is + // happening as it will be skipped by consensus anyways + Some(SubnetSplittingStatus::Scheduled(..)) => Ok(true), + // Wait for the replica to be restarted with the new `subnet_id` + Some(SubnetSplittingStatus::PostSplit(PostSplitArgs { new_subnet_id })) => { + Ok(new_subnet_id != self.replica_config.subnet_id) + } + } + } } fn validate_height_witness( @@ -616,7 +700,8 @@ mod tests { use ic_canonical_state::lazy_tree_conversion::replicated_state_as_lazy_tree; use ic_canonical_state_tree_hash::hash_tree::hash_lazy_tree; use ic_canonical_state_tree_hash::lazy_tree::materialize::materialize_partial; - use ic_consensus_mocks::{Dependencies, dependencies}; + use ic_config::artifact_pool::ArtifactPoolConfig; + use ic_consensus_mocks::{Dependencies, dependencies, dependencies_with_subnet_params}; use ic_crypto_tree_hash::{Digest, Witness, sparse_labeled_tree_from_paths}; use ic_interfaces::{ certification::CertificationPool, @@ -624,9 +709,13 @@ mod tests { }; use ic_interfaces_state_manager::StateHashMetadata; use ic_registry_subnet_type::SubnetType; + use ic_test_artifact_pool::consensus_pool::TestConsensusPool; use ic_test_utilities_consensus::fake::*; use ic_test_utilities_logger::with_test_replica_logger; + use ic_test_utilities_registry::SubnetRecordBuilder; use ic_test_utilities_types::ids::{node_test_id, subnet_test_id}; + use ic_types::backwards_compatibility::BackwardsCompatible; + use ic_types::consensus::{BlockPayload, HashedBlock, Payload, dkg::SplittingArgs}; use ic_types::{ CryptoHashOfPartialState, Height, artifact::CertificationMessageId, @@ -1557,4 +1646,305 @@ mod tests { }) }) } + + // DKG interval length used for subnet-splitting tests. + const TEST_DKG_INTERVAL: u64 = 9; + + fn dependencies_for_splitting_tests( + pool_config: ArtifactPoolConfig, + nodes: u64, + ) -> Dependencies { + let committee = (0..nodes).map(node_test_id).collect::>(); + dependencies_with_subnet_params( + pool_config, + subnet_test_id(0), + vec![( + 1, + SubnetRecordBuilder::from(&committee) + .with_dkg_interval_length(TEST_DKG_INTERVAL) + .build(), + )], + ) + } + + // Advances `pool` by TEST_DKG_INTERVAL rounds so the next block is a DKG + // summary block, then inserts and finalizes that summary block after setting + // its subnet-splitting status to `status`. + // + // Returns the height of the newly finalized summary block. Heights in + // [split_height, split_height + TEST_DKG_INTERVAL] are covered by this + // summary, so `subnet_splitting_status_at_height` will return `status` for + // any of those heights. + fn advance_to_splitting_interval( + pool: &mut TestConsensusPool, + status: SubnetSplittingStatus, + ) -> Height { + pool.advance_round_normal_operation_n(TEST_DKG_INTERVAL); + + let mut proposal = pool.make_next_block(); + let block = proposal.content.as_mut(); + let mut payload = block.payload.as_ref().as_summary().clone(); + payload.dkg.subnet_splitting_status = BackwardsCompatible::new_for_test_only(Some(status)); + block.payload = Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(payload), + ); + proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone()); + + pool.advance_round_with_block(&proposal); + + proposal.height() + } + + fn not_scheduled_splitting() -> SubnetSplittingStatus { + SubnetSplittingStatus::NotScheduled + } + + fn scheduled_splitting() -> SubnetSplittingStatus { + SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: subnet_test_id(0), + destination_subnet_id: subnet_test_id(1), + }) + } + + fn done_splitting_different_subnet() -> SubnetSplittingStatus { + SubnetSplittingStatus::PostSplit(PostSplitArgs { + new_subnet_id: subnet_test_id(1), + }) + } + + fn done_splitting_same_subnet() -> SubnetSplittingStatus { + SubnetSplittingStatus::PostSplit(PostSplitArgs { + new_subnet_id: subnet_test_id(0), + }) + } + + fn assert_for_all_subnet_splitting_statuses( + pool: &mut TestConsensusPool, + mut test: impl FnMut(SubnetSplittingStatus, Height), + ) { + for status in [ + not_scheduled_splitting(), + scheduled_splitting(), + done_splitting_different_subnet(), + done_splitting_same_subnet(), + ] { + let splitting_height = advance_to_splitting_interval(pool, status); + for test_height in splitting_height.get()..=splitting_height.get() + TEST_DKG_INTERVAL { + let test_height = Height::from(test_height); + + test(status, test_height); + } + } + } + + /// Signing should be skipped for heights covered by a `Scheduled` or `Done` with different + /// subnet ID splitting interval. + /// In a `Done` interval with same subnet ID, signing should proceed as normal. + #[test] + fn test_sign_skips_during_subnet_splitting() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + mut pool, + replica_config, + registry, + crypto, + state_manager, + .. + } = dependencies_for_splitting_tests(pool_config.clone(), 4); + + let metrics_registry = MetricsRegistry::new(); + let cert_pool = CertificationPoolImpl::new( + replica_config.node_id, + pool_config, + ic_logger::replica_logger::no_op_logger(), + metrics_registry.clone(), + ); + let certifier = CertifierImpl::new( + replica_config, + registry, + crypto, + state_manager, + pool.get_cache(), + metrics_registry, + log, + ); + + assert_for_all_subnet_splitting_statuses(&mut pool, |status, test_height| { + let shares = certifier.sign( + &cert_pool, + &[StateHashMetadata { + height: test_height, + hash: CryptoHashOfPartialState::from(CryptoHash(vec![1, 2, 3])), + height_witness: Witness::new_for_testing_with_height(), + }], + ); + + match status { + SubnetSplittingStatus::Scheduled(..) => { + assert!( + shares.is_empty(), + "Expected no shares during subnet splitting, got: {shares:?}" + ); + } + SubnetSplittingStatus::PostSplit(PostSplitArgs { new_subnet_id }) + if new_subnet_id != subnet_test_id(0) => + { + assert!( + shares.is_empty(), + "Expected no shares after Done splitting with different subnet ID, got: {shares:?}" + ); + } + SubnetSplittingStatus::NotScheduled + | SubnetSplittingStatus::PostSplit(..) => { + assert!( + !shares.is_empty(), + "Expected shares when not splitting or splitting with same subnet ID" + ); + } + } + }); + }) + }) + } + + /// An incoming share at a height inside a `Scheduled` or `Done` with different subnet ID + /// splitting interval should be ignored and not validated, as it could be from the other + /// subnet. + /// In a `Done` interval with same subnet ID, shares should be validated as normal. + #[test] + fn test_validate_share_handles_invalid_during_scheduled_subnet_splitting() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + mut pool, + replica_config, + registry, + crypto, + state_manager, + .. + } = dependencies_for_splitting_tests(pool_config.clone(), 4); + + let metrics_registry = MetricsRegistry::new(); + let cert_pool = CertificationPoolImpl::new( + replica_config.node_id, + pool_config, + ic_logger::replica_logger::no_op_logger(), + metrics_registry.clone(), + ); + let certifier = CertifierImpl::new( + replica_config, + registry, + crypto, + state_manager, + pool.get_cache(), + metrics_registry, + log, + ); + + assert_for_all_subnet_splitting_statuses(&mut pool, |status, test_height| { + let content = gen_content(test_height); + let share = CertificationShare { + height: test_height, + height_witness: Witness::new_for_testing_with_height(), + signed: Signed { + content, + signature: ThresholdSignatureShare::fake(node_test_id(1)), + }, + }; + + let result = certifier.validate_share(&cert_pool, &share); + match status { + SubnetSplittingStatus::Scheduled(..) => { + assert_eq!(result, None, "Expected None during subnet splitting"); + } + SubnetSplittingStatus::PostSplit(PostSplitArgs { new_subnet_id }) + if new_subnet_id != subnet_test_id(0) => + { + assert_eq!( + result, None, + "Expected None after Done splitting with different subnet ID" + ); + } + SubnetSplittingStatus::NotScheduled + | SubnetSplittingStatus::PostSplit(..) => { + assert_eq!( + result, + Some(ChangeAction::MoveToValidated( + CertificationMessage::CertificationShare(share) + )), + "Expected MoveToValidated when not splitting or splitting with same subnet ID" + ); + } + } + }); + }) + }) + } + + /// Full certifications received during a `Scheduled` or `Done` with different subnet ID + /// splitting interval should be ignored and not validated, as they could be from the other + #[test] + fn test_validate_certification_validates_despite_scheduled_subnet_splitting() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|log| { + let Dependencies { + mut pool, + replica_config, + registry, + crypto, + state_manager, + .. + } = dependencies_for_splitting_tests(pool_config.clone(), 1); + + let certifier = CertifierImpl::new( + replica_config, + registry, + crypto, + state_manager, + pool.get_cache(), + MetricsRegistry::new(), + log, + ); + + assert_for_all_subnet_splitting_statuses(&mut pool, |status, test_height| { + let content = gen_content(test_height); + let cert = Certification { + height: test_height, + height_witness: Some(Witness::new_for_testing_with_height()), + signed: Signed { + content, + signature: ThresholdSignature::fake(), + }, + }; + + let result = certifier.validate_certification(&cert); + match status { + SubnetSplittingStatus::Scheduled(..) => { + assert_eq!(result, None, "Expected None during subnet splitting"); + } + SubnetSplittingStatus::PostSplit(PostSplitArgs { new_subnet_id }) + if new_subnet_id != subnet_test_id(0) => + { + assert_eq!( + result, None, + "Expected None after Done splitting with different subnet ID" + ); + } + SubnetSplittingStatus::NotScheduled + | SubnetSplittingStatus::PostSplit(..) => { + assert_eq!( + result, + Some(ChangeAction::MoveToValidated( + CertificationMessage::Certification(cert.clone()) + )), + "Expected MoveToValidated when not splitting or splitting with same subnet ID" + ); + } + } + }); + }) + }) + } } diff --git a/rs/consensus/cup_utils/BUILD.bazel b/rs/consensus/cup_utils/BUILD.bazel index 30503a9687b0..7e947e4a3286 100644 --- a/rs/consensus/cup_utils/BUILD.bazel +++ b/rs/consensus/cup_utils/BUILD.bazel @@ -17,6 +17,7 @@ rust_library( "//rs/registry/helpers", "//rs/types/types", "@crate_index//:slog", + "@crate_index//:thiserror", ], ) diff --git a/rs/consensus/cup_utils/Cargo.toml b/rs/consensus/cup_utils/Cargo.toml index bae8e34e72dc..c2037720ca79 100644 --- a/rs/consensus/cup_utils/Cargo.toml +++ b/rs/consensus/cup_utils/Cargo.toml @@ -16,6 +16,7 @@ ic-registry-client-helpers = { path = "../../registry/helpers" } ic-types = { path = "../../types/types" } phantom_newtype = { path = "../../phantom_newtype" } slog = { workspace = true } +thiserror = { workspace = true } [dev-dependencies] ic-crypto-test-utils-ni-dkg = { path = "../../crypto/test_utils/ni-dkg" } diff --git a/rs/consensus/cup_utils/src/lib.rs b/rs/consensus/cup_utils/src/lib.rs index 8bf354154c3d..c5afbf311b8c 100644 --- a/rs/consensus/cup_utils/src/lib.rs +++ b/rs/consensus/cup_utils/src/lib.rs @@ -6,7 +6,7 @@ use ic_consensus_idkg::{ utils::{get_idkg_chain_key_config_if_enabled, inspect_idkg_chain_key_initializations}, }; use ic_interfaces_registry::RegistryClient; -use ic_logger::{ReplicaLogger, warn}; +use ic_logger::ReplicaLogger; use ic_protobuf::registry::subnet::v1::CatchUpPackageContents; use ic_registry_client_helpers::subnet::SubnetRegistry; use ic_types::{ @@ -14,15 +14,35 @@ use ic_types::{ batch::ValidationContext, consensus::{ Block, BlockPayload, CatchUpContent, CatchUpPackage, HashedBlock, HashedRandomBeacon, - Payload, RandomBeaconContent, Rank, SummaryPayload, idkg, + Payload, RandomBeaconContent, Rank, RegistryCUP, RegistryCupType, SummaryPayload, idkg, }, crypto::{ CombinedThresholdSig, CombinedThresholdSigOf, CryptoHash, Signed, crypto_hash, threshold_sig::ni_dkg::NiDkgTag, }, + registry::RegistryClientError, signature::ThresholdSignature, }; use phantom_newtype::Id; +use thiserror::Error; + +#[derive(Error, Debug)] +pub enum RegistryCupCreationError { + #[error("Failed to retrieve subnet replica version at registry version {0}: {1:?}")] + ReplicaVersionError(RegistryVersion, RegistryClientError), + #[error("Missing subnet replica version at registry version {0}")] + ReplicaVersionMissing(RegistryVersion), + #[error("Failed constructing NiDKG summary block from CUP contents: {0}")] + DkgSummaryCreationError(String), + #[error("Failed constructing IDKG summary block from CUP contents: {0}")] + IDkgSummaryCreationError(String), + #[error("No current threshold transcript with tag {0:?} in registry CUP contents")] + ThresholdTranscriptMissing(NiDkgTag), + #[error("Failed to retrieve CUP contents from the registry at version {0}: {1:?}")] + FailedToGetCupContents(RegistryVersion, RegistryClientError), + #[error("Missing registry CUP contents at version {0}")] + CupContentsMissing(RegistryVersion), +} /// Constructs a genesis/recovery CUP from the CUP contents associated with the /// given subnet from the provided CUP contents @@ -32,73 +52,44 @@ pub fn make_registry_cup_from_cup_contents( cup_contents: CatchUpPackageContents, registry_version: RegistryVersion, logger: &ReplicaLogger, -) -> Option { - let replica_version = match registry.get_replica_version(subnet_id, registry_version) { - Ok(Some(replica_version)) => replica_version, - err => { - warn!( - logger, - "Failed to retrieve subnet replica version at registry version {:?}: {:?}", - registry_version, - err - ); - return None; - } - }; - let dkg_summary = match get_dkg_summary_from_cup_contents( +) -> Result { + let replica_version = registry + .get_replica_version(subnet_id, registry_version) + .map_err(|err| RegistryCupCreationError::ReplicaVersionError(registry_version, err))? + .ok_or(RegistryCupCreationError::ReplicaVersionMissing( + registry_version, + ))?; + + let dkg_summary = get_dkg_summary_from_cup_contents( cup_contents.clone(), subnet_id, registry, registry_version, - ) { - Ok(summary) => summary, - Err(err) => { - warn!( - logger, - "Failed constructing NiDKG summary block from CUP contents: {}.", err - ); - - return None; - } - }; + ) + .map_err(RegistryCupCreationError::DkgSummaryCreationError)?; let cup_height = Height::new(cup_contents.height); - let idkg_summary = match bootstrap_idkg_summary( + let idkg_summary = bootstrap_idkg_summary( cup_contents.clone(), subnet_id, registry_version, registry, logger, - ) { - Ok(summary) => summary, - Err(err) => { - warn!( - logger, - "Failed constructing IDKG summary block from CUP contents: {}.", err - ); - - return None; - } - }; + ) + .map_err(RegistryCupCreationError::IDkgSummaryCreationError)?; - let Some(low_threshold_transcript) = dkg_summary.current_transcript(&NiDkgTag::LowThreshold) - else { - warn!( - logger, - "No current low threshold transcript in registry CUP contents" - ); - return None; - }; + let low_threshold_transcript = dkg_summary + .current_transcript(&NiDkgTag::LowThreshold) + .ok_or(RegistryCupCreationError::ThresholdTranscriptMissing( + NiDkgTag::LowThreshold, + ))?; let low_dkg_id = low_threshold_transcript.dkg_id.clone(); - let Some(high_threshold_transcript) = dkg_summary.current_transcript(&NiDkgTag::HighThreshold) - else { - warn!( - logger, - "No current high threshold transcript in registry CUP contents" - ); - return None; - }; + let high_threshold_transcript = dkg_summary + .current_transcript(&NiDkgTag::HighThreshold) + .ok_or(RegistryCupCreationError::ThresholdTranscriptMissing( + NiDkgTag::HighThreshold, + ))?; let high_dkg_id = high_threshold_transcript.dkg_id.clone(); // In a NNS subnet recovery case the block validation context needs to reference a registry @@ -138,8 +129,9 @@ pub fn make_registry_cup_from_cup_contents( signature: CombinedThresholdSigOf::new(CombinedThresholdSig(vec![])), }, }; + let cup_type = RegistryCupType::from(&cup_contents); - Some(CatchUpPackage { + let cup = CatchUpPackage { content: CatchUpContent::new( HashedBlock::new(crypto_hash, block), HashedRandomBeacon::new(crypto_hash, random_beacon), @@ -150,7 +142,9 @@ pub fn make_registry_cup_from_cup_contents( signer: high_dkg_id, signature: CombinedThresholdSigOf::new(CombinedThresholdSig(vec![])), }, - }) + }; + + Ok(RegistryCUP { cup, cup_type }) } /// Constructs a genesis/recovery CUP from the CUP contents associated with the @@ -159,26 +153,20 @@ pub fn make_registry_cup( registry: &dyn RegistryClient, subnet_id: SubnetId, logger: &ReplicaLogger, -) -> Option { - let versioned_record = match registry.get_cup_contents(subnet_id, registry.get_latest_version()) - { - Ok(versioned_record) => versioned_record, - Err(e) => { - warn!( - logger, - "Failed to retrieve versioned record from the registry {:?}", e, - ); - return None; - } - }; +) -> Result { + let latest_registry_version = registry.get_latest_version(); + let versioned_record = registry + .get_cup_contents(subnet_id, latest_registry_version) + .map_err(|err| { + RegistryCupCreationError::FailedToGetCupContents(latest_registry_version, err) + })?; - let Some(cup_contents) = versioned_record.value else { - warn!( - logger, - "Missing registry CUP contents at version {}", versioned_record.version - ); - return None; - }; + let cup_contents = + versioned_record + .value + .ok_or(RegistryCupCreationError::CupContentsMissing( + versioned_record.version, + ))?; make_registry_cup_from_cup_contents( registry, @@ -310,26 +298,27 @@ mod tests { None } }); - let result = + let RegistryCUP { cup, cup_type } = make_registry_cup(®istry_client, subnet_test_id(0), &no_op_logger()).unwrap(); + assert_eq!(cup_type, RegistryCupType::Recovery); assert_eq!( - result.content.state_hash.get_ref(), + cup.content.state_hash.get_ref(), &CryptoHash(vec![1, 2, 3, 4, 5]) ); assert_eq!( - result.content.block.get_value().context.registry_version, + cup.content.block.get_value().context.registry_version, RegistryVersion::from(12345) ); assert_eq!( - result.content.block.get_value().context.certified_height, + cup.content.block.get_value().context.certified_height, Height::from(54321) ); assert_eq!( - result.content.version(), + cup.content.version(), &ReplicaVersion::try_from("TestID").unwrap() ); - assert_eq!(result.signature.signer.dealer_subnet, subnet_test_id(0)); + assert_eq!(cup.signature.signer.dealer_subnet, subnet_test_id(0)); } /// `RegistryClient` implementation that allows to provide a custom function diff --git a/rs/consensus/dkg/src/dkg_key_manager.rs b/rs/consensus/dkg/src/dkg_key_manager.rs index c533205776da..3c7ef54e1600 100644 --- a/rs/consensus/dkg/src/dkg_key_manager.rs +++ b/rs/consensus/dkg/src/dkg_key_manager.rs @@ -3,17 +3,19 @@ //! there is something to do. On high-level, it's responsible of spawning //! threads triggering long-running CSP operation and book-keeping of //! thread-handles. -use ic_consensus_utils::{crypto::ConsensusCrypto, pool_reader::PoolReader}; +use ic_consensus_utils::{crypto::ConsensusCrypto, pool_reader::PoolReader, subnet_splitting}; use ic_interfaces::crypto::{ErrorReproducibility, LoadTranscriptResult, NiDkgAlgorithm}; +use ic_interfaces_registry::RegistryClient; use ic_logger::{ReplicaLogger, error, info, warn}; use ic_metrics::{MetricsRegistry, buckets::decimal_buckets}; use ic_types::{ Height, - consensus::{HasHeight, dkg::DkgSummary}, + consensus::{Block, HasHeight, dkg::DkgSummary}, crypto::threshold_sig::ni_dkg::{ NiDkgId, NiDkgTag, NiDkgTargetSubnet, NiDkgTranscript, errors::load_transcript_error::DkgLoadTranscriptError, }, + replica_config::ReplicaConfig, }; use prometheus::{HistogramVec, IntCounterVec, IntGauge, IntGaugeVec}; use std::{ @@ -25,6 +27,8 @@ use std::{ time::Instant, }; +use crate::payload_builder::get_post_split_dkg_summary; + struct Metrics { pub dkg_ops_duration: HistogramVec, pub dkg_instance_id: IntGaugeVec, @@ -94,6 +98,8 @@ pub struct DkgKeyManager { >, // This is a thread handle used to keep track of asynchronous key removals. pending_key_removal: Option>, + registry: Arc, + replica_config: ReplicaConfig, } impl DkgKeyManager { @@ -103,6 +109,8 @@ impl DkgKeyManager { crypto: Arc, logger: ReplicaLogger, pool_reader: &PoolReader<'_>, + registry: Arc, + replica_config: ReplicaConfig, ) -> Self { let mut manager = Self { crypto, @@ -112,6 +120,8 @@ impl DkgKeyManager { last_cup_height: Default::default(), pending_transcript_loads: Default::default(), pending_key_removal: Default::default(), + registry, + replica_config, }; // By calling on state change during initialization, we make sure, that the key store is @@ -234,6 +244,53 @@ impl DkgKeyManager { self.load_transcripts_from_summary(&summary.dkg); self.last_dkg_summary_height = Some(summary_block.height); } + + // Always try to create the summary following a potential subnet split and load its + // transcripts. This is needed as a special case, to let the replica sign and verify CUP + // shares corresponding to that post-split summary using the new DKG transcripts. + self.load_post_split_transcripts_if_necessary(&summary_block); + } + + fn load_post_split_transcripts_if_necessary(&mut self, summary_block: &Block) { + let Some(scheduled) = subnet_splitting::is_split_scheduled(summary_block) else { + return; + }; + + let new_subnet_id = match subnet_splitting::get_post_split_subnet_assignment( + self.replica_config.node_id, + summary_block, + self.registry.as_ref(), + scheduled, + ) { + Ok(assignment) => assignment.new_subnet_id, + Err(err) => { + error!( + self.logger, + "Couldn't determine the post-split subnet assignment after the split: {err:?}" + ); + return; + } + }; + + let next_summary = match get_post_split_dkg_summary( + new_subnet_id, + self.registry.as_ref(), + summary_block, + ) { + Ok(next_summary) => next_summary, + Err(err) => { + error!( + self.logger, + "Couldn't get the next DKG summary for the new subnet {new_subnet_id:?} \ + after the split: {err:?}" + ); + return; + } + }; + + info!(every_n_seconds => 5, self.logger, "Adding post-split DKG transcripts"); + self.load_transcripts_from_summary(&next_summary); + self.last_dkg_summary_height = Some(next_summary.height); } /// Ensures that the pending transcripts are loaded BEFORE they are needed. For @@ -557,12 +614,17 @@ fn dkg_id_log_msg(id: &NiDkgId) -> String { #[cfg(test)] mod tests { use super::*; - use ic_consensus_mocks::{Dependencies, dependencies_with_subnet_params}; + use ic_consensus_mocks::{Dependencies, DependenciesBuilder, dependencies_with_subnet_params}; use ic_crypto_test_utils_crypto_returning_ok::CryptoReturningOk; use ic_metrics::MetricsRegistry; use ic_test_utilities_logger::with_test_replica_logger; use ic_test_utilities_registry::SubnetRecordBuilder; use ic_test_utilities_types::ids::{node_test_id, subnet_test_id}; + use ic_types::backwards_compatibility::BackwardsCompatible; + use ic_types::consensus::{ + BlockPayload, HashedBlock, Payload, + dkg::{SplittingArgs, SubnetSplittingStatus}, + }; #[test] fn test_transcripts_get_loaded_and_retained() { @@ -570,7 +632,12 @@ mod tests { with_test_replica_logger(|logger| { let nodes: Vec<_> = (0..1).map(node_test_id).collect(); let dkg_interval_len = 3; - let Dependencies { mut pool, .. } = dependencies_with_subnet_params( + let Dependencies { + mut pool, + registry, + replica_config, + .. + } = dependencies_with_subnet_params( pool_config, subnet_test_id(222), vec![( @@ -586,6 +653,8 @@ mod tests { csp.clone(), logger, &PoolReader::new(&pool), + registry, + replica_config, ); // Emulate the first invocation of the dkg key manager and make sure all @@ -676,4 +745,121 @@ mod tests { }); }); } + + /// Verifies that when a subnet split is in progress, the key manager loads + /// the transcripts from the post-split DKG summary in addition to the + /// current summary's transcripts. + #[test] + fn test_subnet_splitting_loads_post_split_transcripts() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + with_test_replica_logger(|logger| { + let source_subnet_id = subnet_test_id(1); + let destination_subnet_id = subnet_test_id(2); + let source_nodes = vec![node_test_id(0)]; + let destination_nodes = vec![node_test_id(1)]; + let registry_version = 1u64; + let dkg_interval_len = 3; + + // Set up two subnets: the source subnet is where node_test_id(0) lives, the + // destination subnet is where node_test_id(1) lives after the split. + // DependenciesBuilder calls insert_initial_dkg_transcript for both, so valid + // CUP contents exist in the registry for both subnets. + let Dependencies { + mut pool, + registry, + replica_config, + .. + } = DependenciesBuilder::new( + pool_config, + vec![ + ( + registry_version, + source_subnet_id, + SubnetRecordBuilder::from(&source_nodes) + .with_dkg_interval_length(dkg_interval_len) + .build(), + ), + ( + registry_version, + destination_subnet_id, + SubnetRecordBuilder::from(&destination_nodes) + .with_dkg_interval_length(dkg_interval_len) + .build(), + ), + ], + ) + .with_mocked_state_manager() + .build(); + + // Advance dkg_interval_len rounds so the finalized tip is at height + // dkg_interval_len, stopping just before the next DKG interval boundary. + pool.advance_round_normal_operation_no_cup_n(dkg_interval_len); + + // Build a DKG summary at the next interval boundary that signals a subnet split. + let mut splitting_proposal = pool.make_next_block(); + let mut splitting_block = splitting_proposal.content.as_ref().clone(); + let splitting_height = splitting_block.height; + let mut splitting_summary = splitting_block.payload.as_ref().as_summary().clone(); + splitting_summary.dkg.subnet_splitting_status = + BackwardsCompatible::new_for_test_only(Some(SubnetSplittingStatus::Scheduled( + SplittingArgs { + source_subnet_id, + destination_subnet_id, + }, + ))); + splitting_block.payload = Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(splitting_summary.clone()), + ); + splitting_proposal.content = + HashedBlock::new(ic_types::crypto::crypto_hash, splitting_block.clone()); + pool.advance_round_with_block(&splitting_proposal); + + let post_split_dkg_summary = get_post_split_dkg_summary( + destination_subnet_id, + registry.as_ref(), + &splitting_block, + ) + .expect("Couldn't get the post-split summary"); + // Safety-check: the post-split summary should have been produced by the registry, + // meaning `next_transcripts` should be empty + assert!( + post_split_dkg_summary.next_transcripts().is_empty(), + "The post-split summary should not contain next transcripts" + ); + + let csp = Arc::new(CryptoReturningOk::default()); + let mut key_manager = DkgKeyManager::new( + MetricsRegistry::new(), + csp.clone(), + logger, + &PoolReader::new(&pool), + registry, + replica_config, + ); + key_manager.sync(); + + // The key manager should not only load the transcripts from the splitting summary, + // but also the transcripts from the post split one, which is expected to be one + // interval later. + assert_eq!( + key_manager.last_dkg_summary_height, + Some(splitting_height + (dkg_interval_len + 1).into()) + ); + splitting_summary + .dkg + .into_transcripts() + .iter() + .chain(post_split_dkg_summary.into_transcripts().iter()) + .for_each(|transcript| { + let id = &transcript.dkg_id; + assert!( + csp.loaded_transcripts.read().unwrap().contains(id), + "Transcript {} should have been loaded", + dkg_id_log_msg(id) + ); + }); + }); + }); + } } diff --git a/rs/consensus/dkg/src/lib.rs b/rs/consensus/dkg/src/lib.rs index bdf33dfc803a..d83b5ab55705 100644 --- a/rs/consensus/dkg/src/lib.rs +++ b/rs/consensus/dkg/src/lib.rs @@ -458,6 +458,7 @@ mod tests { errors::create_transcript_error::DkgCreateTranscriptError, }, }, + replica_config::ReplicaConfig, time::UNIX_EPOCH, }; use payload_validator::validate_payload; @@ -507,6 +508,7 @@ mod tests { dkg_pool, registry, state_manager, + replica_config, .. } = dependencies_with_subnet_params( pool_config, @@ -529,8 +531,13 @@ mod tests { // Now we instantiate the DKG component for node Id = 1, who is a dealer. let replica_1 = node_test_id(1); - let dkg_key_manager = - new_dkg_key_manager(crypto.clone(), logger.clone(), &PoolReader::new(&pool)); + let dkg_key_manager = new_dkg_key_manager( + crypto.clone(), + logger.clone(), + &PoolReader::new(&pool), + registry.clone(), + replica_config.clone(), + ); let dkg = DkgImpl::new( replica_1, subnet_id, @@ -599,8 +606,13 @@ mod tests { // Create another dealer and add his dealings into the unvalidated pool of // replica 1. let replica_2 = node_test_id(2); - let dkg_key_manager_2 = - new_dkg_key_manager(crypto.clone(), logger.clone(), &PoolReader::new(&pool)); + let dkg_key_manager_2 = new_dkg_key_manager( + crypto.clone(), + logger.clone(), + &PoolReader::new(&pool), + registry.clone(), + replica_config.clone(), + ); let dkg_2 = DkgImpl::new( replica_2, subnet_id, @@ -687,8 +699,13 @@ mod tests { let mut dkg_pool = DkgPoolImpl::new(MetricsRegistry::new(), logger.clone(), Height::from(0)); // Let's check that replica 3, who's not a dealer, does not produce dealings. - let dkg_key_manager = - new_dkg_key_manager(crypto.clone(), logger.clone(), &PoolReader::new(&pool)); + let dkg_key_manager = new_dkg_key_manager( + crypto.clone(), + logger.clone(), + &PoolReader::new(&pool), + registry.clone(), + replica_config.clone(), + ); let dkg = DkgImpl::new( node_test_id(3), replica_config.subnet_id, @@ -703,8 +720,13 @@ mod tests { assert!(dkg.on_state_change(&dkg_pool).is_empty()); // Now we instantiate the DKG component for node Id = 1, who is a dealer. - let dkg_key_manager = - new_dkg_key_manager(crypto.clone(), logger.clone(), &PoolReader::new(&pool)); + let dkg_key_manager = new_dkg_key_manager( + crypto.clone(), + logger.clone(), + &PoolReader::new(&pool), + registry.clone(), + replica_config.clone(), + ); let dkg = DkgImpl::new( node_test_id(1), replica_config.subnet_id, @@ -777,6 +799,7 @@ mod tests { mut pool, crypto, registry, + replica_config, state_manager, dkg_pool, .. @@ -801,8 +824,13 @@ mod tests { ); // Now we instantiate the DKG component for node Id = 1, who is a dealer. - let dkg_key_manager = - new_dkg_key_manager(crypto.clone(), logger.clone(), &PoolReader::new(&pool)); + let dkg_key_manager = new_dkg_key_manager( + crypto.clone(), + logger.clone(), + &PoolReader::new(&pool), + registry.clone(), + replica_config, + ); let dkg = DkgImpl::new( node_test_id(1), subnet_id, @@ -1094,6 +1122,8 @@ mod tests { crypto.clone(), logger.clone(), &PoolReader::new(&consensus_pool_1), + registry_1.clone(), + replica_config_1.clone(), ); let dkg_1 = DkgImpl::new( node_id_1, @@ -1111,6 +1141,8 @@ mod tests { crypto.clone(), logger.clone(), &PoolReader::new(&consensus_pool_2), + registry_2.clone(), + replica_config_2.clone(), ); let dkg_2 = DkgImpl::new( node_id_2, @@ -1599,6 +1631,8 @@ mod tests { crypto_1.clone(), logger.clone(), &PoolReader::new(&pool_1), + dependencies_1.registry.clone(), + dependencies_1.replica_config.clone(), ); let dkg_1 = DkgImpl::new( node_test_id(1), @@ -1619,7 +1653,13 @@ mod tests { state_manager_2, crypto_2.clone(), pool_2.get_cache(), - new_dkg_key_manager(crypto_2, logger.clone(), &PoolReader::new(&pool_2)), + new_dkg_key_manager( + crypto_2, + logger.clone(), + &PoolReader::new(&pool_2), + dependencies_2.registry.clone(), + dependencies_2.replica_config.clone(), + ), MetricsRegistry::new(), logger.clone(), ); @@ -1948,6 +1988,7 @@ mod tests { } _ => panic!("expected data block"), }; + assert!( validate_payload( subnet_test_id(0), @@ -2126,6 +2167,8 @@ mod tests { deps.crypto.clone(), logger.clone(), &PoolReader::new(&deps.pool), + deps.registry.clone(), + deps.replica_config.clone(), ); let receiver_dkg = DkgImpl::new( node_test_id(2), @@ -2849,12 +2892,16 @@ mod tests { crypto: Arc, logger: ReplicaLogger, pool_reader: &PoolReader<'_>, + registry: Arc, + replica_config: ReplicaConfig, ) -> Arc> { Arc::new(Mutex::new(DkgKeyManager::new( MetricsRegistry::new(), crypto, logger, pool_reader, + registry, + replica_config, ))) } diff --git a/rs/consensus/dkg/src/payload_builder.rs b/rs/consensus/dkg/src/payload_builder.rs index cd4f4fa8cc95..5de2cf197294 100644 --- a/rs/consensus/dkg/src/payload_builder.rs +++ b/rs/consensus/dkg/src/payload_builder.rs @@ -6,7 +6,7 @@ use crate::{ }, utils::{self, tags_iter, vetkd_key_ids_for_subnet}, }; -use ic_consensus_utils::{crypto::ConsensusCrypto, pool_reader::PoolReader}; +use ic_consensus_utils::{crypto::ConsensusCrypto, pool_reader::PoolReader, subnet_splitting}; use ic_interfaces::{ crypto::{ErrorReproducibility, NiDkgAlgorithm}, dkg::DkgPool, @@ -28,7 +28,7 @@ use ic_types::{ Block, dkg::{ DkgDataPayload, DkgPayload, DkgPayloadCreationError, DkgSummary, Message, - RemoteTranscriptResult, + PostSplitArgs, RemoteTranscriptResult, SplittingArgs, SubnetSplittingStatus, }, get_faults_tolerated, }, @@ -526,6 +526,24 @@ pub(super) fn create_summary_payload( subnet_id, )?; + let subnet_splitting_status = match subnet_splitting::get_status( + registry_client, + subnet_id, + registry_version, + validation_context.registry_version, + ) + .map_err(|err| DkgPayloadCreationError::SubnetSplittingStatusError(err.to_string()))? + { + subnet_splitting::Status::NotScheduled => SubnetSplittingStatus::NotScheduled, + subnet_splitting::Status::Scheduled { + destination_subnet_id, + scheduled_at: _, + } => SubnetSplittingStatus::Scheduled(SplittingArgs { + destination_subnet_id, + source_subnet_id: subnet_id, + }), + }; + // New configs are created using the new stable registry version proposed by this // block, which determines receivers of the dealings. let local_configs = get_configs_for_local_transcripts( @@ -550,6 +568,7 @@ pub(super) fn create_summary_payload( next_interval_length, height, remote_dkg_attempts, + subnet_splitting_status, )) } @@ -576,6 +595,22 @@ pub fn get_dkg_summary_from_cup_contents( subnet_id: SubnetId, registry: &dyn RegistryClient, registry_version: RegistryVersion, +) -> Result { + get_dkg_summary_from_cup_contents_with_subnet_splitting( + cup_contents, + subnet_id, + registry, + registry_version, + SubnetSplittingStatus::default(), + ) +} + +fn get_dkg_summary_from_cup_contents_with_subnet_splitting( + cup_contents: CatchUpPackageContents, + subnet_id: SubnetId, + registry: &dyn RegistryClient, + registry_version: RegistryVersion, + subnet_splitting_status: SubnetSplittingStatus, ) -> Result { // If we're in a NNS subnet recovery case with failover nodes, we extract the registry of the // NNS we're recovering. @@ -677,6 +712,7 @@ pub fn get_dkg_summary_from_cup_contents( next_interval_length, height, BTreeMap::new(), // remote_dkg_attempts + subnet_splitting_status, )) } @@ -870,6 +906,40 @@ pub(crate) fn create_remote_dkg_config( }) } +/// Creates a DKG summary for the summary block right after the subnet has been split. +pub fn get_post_split_dkg_summary( + new_subnet_id: SubnetId, + registry: &dyn RegistryClient, + last_summary_block: &Block, +) -> Result { + let last_summary = &last_summary_block.payload.as_ref().as_summary().dkg; + debug_assert!(matches!( + last_summary.subnet_splitting_status(), + SubnetSplittingStatus::Scheduled(..) + )); + let registry_version = last_summary_block.context.registry_version; + + let mut cup_contents = registry + .get_cup_contents(new_subnet_id, registry_version) + .map_err(|err| { + format!("Failed to get the cup contents at registry version {registry_version}: {err}") + })? + .value + .ok_or_else(|| format!("Empty cup contents at registry version {registry_version}"))?; + + // During subnet splitting we skip one DKG interval + cup_contents.height = last_summary.get_next_start_height().get(); + + get_dkg_summary_from_cup_contents_with_subnet_splitting( + cup_contents, + new_subnet_id, + registry, + registry_version, + SubnetSplittingStatus::PostSplit(PostSplitArgs { new_subnet_id }), + ) + .map_err(|err| format!("Failed to create post-split dkg summary from contents: {err}")) +} + #[cfg(test)] mod tests { use crate::tests::test_vet_key_config; diff --git a/rs/consensus/dkg/src/payload_validator.rs b/rs/consensus/dkg/src/payload_validator.rs index aaba630b5695..616850428932 100644 --- a/rs/consensus/dkg/src/payload_validator.rs +++ b/rs/consensus/dkg/src/payload_validator.rs @@ -283,6 +283,7 @@ mod tests { }, crypto::threshold_sig::ni_dkg::{NiDkgId, NiDkgTag, NiDkgTargetSubnet}, messages::CallbackId, + replica_config::ReplicaConfig, time::UNIX_EPOCH, }; use std::{ @@ -357,6 +358,9 @@ mod tests { // This will be a summary block, since we are at dkg_interval_length height let block = Block::from(pool.make_next_block()); let summary = block.payload.as_ref(); + let last_summary_block = PoolReader::new(&pool) + .dkg_summary_block(&parent_block) + .unwrap(); let last_summary_block = PoolReader::new(&pool) .dkg_summary_block(&parent_block) @@ -792,6 +796,8 @@ mod tests { crypto.clone(), no_op_logger(), &PoolReader::new(&pool), + registry.clone(), + ReplicaConfig { node_id, subnet_id }, ); let key_manager = Arc::new(Mutex::new(key_manager)); let dkg_impl = DkgImpl::new( diff --git a/rs/consensus/dkg/src/remote.rs b/rs/consensus/dkg/src/remote.rs index ebe6c41b1b02..84f87d42b8f2 100644 --- a/rs/consensus/dkg/src/remote.rs +++ b/rs/consensus/dkg/src/remote.rs @@ -298,6 +298,7 @@ mod tests { Height::from(10), height, remote_dkg_attempts, + Default::default(), ) } diff --git a/rs/consensus/idkg/src/payload_builder.rs b/rs/consensus/idkg/src/payload_builder.rs index 1194f2a67190..da3ed038bf8e 100644 --- a/rs/consensus/idkg/src/payload_builder.rs +++ b/rs/consensus/idkg/src/payload_builder.rs @@ -782,6 +782,7 @@ mod tests { Height::from(100), height, BTreeMap::new(), + Default::default(), ), idkg: Some(idkg_summary), }) diff --git a/rs/consensus/mocks/src/lib.rs b/rs/consensus/mocks/src/lib.rs index 3895d75b7e71..c0cc35cffcdc 100644 --- a/rs/consensus/mocks/src/lib.rs +++ b/rs/consensus/mocks/src/lib.rs @@ -17,7 +17,10 @@ use ic_registry_proto_data_provider::ProtoRegistryDataProvider; use ic_test_artifact_pool::consensus_pool::TestConsensusPool; use ic_test_utilities::state_manager::RefMockStateManager; use ic_test_utilities_consensus::IDkgStatsNoOp; -use ic_test_utilities_registry::{SubnetRecordBuilder, setup_registry_non_final}; +use ic_test_utilities_registry::{ + SubnetRecordBuilder, add_single_subnet_record, add_subnet_list_record, + insert_initial_dkg_transcript, +}; use ic_test_utilities_time::FastForwardTimeSource; use ic_test_utilities_types::ids::{node_test_id, subnet_test_id}; use ic_types::{ @@ -28,7 +31,10 @@ use ic_types::{ }; use mockall::predicate::*; use mockall::*; -use std::sync::{Arc, RwLock}; +use std::{ + collections::BTreeSet, + sync::{Arc, RwLock}, +}; mock! { pub PayloadBuilder {} @@ -105,6 +111,169 @@ pub struct Dependencies { pub canister_http_pool: Arc>, } +pub struct DependenciesBuilder { + pool_config: ArtifactPoolConfig, + records: Vec<(u64, SubnetId, SubnetRecord)>, + replica_config: ReplicaConfig, + mocked_state_manager: bool, + #[allow(clippy::type_complexity)] + additional_registry_mutations: Vec)>>, +} + +impl DependenciesBuilder { + pub fn new( + pool_config: ArtifactPoolConfig, + records: Vec<(u64, SubnetId, SubnetRecord)>, + ) -> Self { + Self { + pool_config, + replica_config: ReplicaConfig { + node_id: node_test_id(0), + subnet_id: records[0].1, + }, + records, + mocked_state_manager: false, + additional_registry_mutations: Vec::new(), + } + } + + pub fn with_replica_config(mut self, replica_config: ReplicaConfig) -> Self { + self.replica_config = replica_config; + + self + } + + pub fn with_mocked_state_manager(mut self) -> Self { + self.mocked_state_manager = true; + + self + } + + pub fn add_additional_registry_mutation( + mut self, + mutation: impl Fn(&Arc) + 'static, + ) -> Self { + self.additional_registry_mutations.push(Box::new(mutation)); + + self + } + + pub fn build(self) -> Dependencies { + let time_source = FastForwardTimeSource::new(); + let initial_registry_version = RegistryVersion::from(self.records[0].clone().0); + let registry_data_provider = Arc::new(ProtoRegistryDataProvider::new()); + assert!( + !self.records.is_empty(), + "Cannot setup a registry without records." + ); + let mut subnet_ids: BTreeSet = BTreeSet::default(); + let mut last_version = None; + + for (version, subnet_id, record) in self.records { + if let Some(last_version) = last_version + && last_version != version + { + add_subnet_list_record( + ®istry_data_provider, + last_version, + Vec::from_iter(subnet_ids.clone()), + ); + } + + if subnet_ids.insert(subnet_id) { + insert_initial_dkg_transcript(version, subnet_id, &record, ®istry_data_provider); + } + + add_single_subnet_record(®istry_data_provider, version, subnet_id, record); + + last_version = Some(version); + } + + if let Some(last_version) = last_version { + add_subnet_list_record( + ®istry_data_provider, + last_version, + Vec::from_iter(subnet_ids), + ); + } + + for registry_mutation in self.additional_registry_mutations { + registry_mutation(®istry_data_provider); + } + + let registry = Arc::new(FakeRegistryClient::new( + Arc::clone(®istry_data_provider) as Arc<_> + )); + + registry_data_provider + .add( + ROOT_SUBNET_ID_KEY, + initial_registry_version, + Some(ic_types::subnet_id_into_protobuf(subnet_test_id(0))), + ) + .unwrap(); + registry.update_to_latest_version(); + let crypto = Arc::new(CryptoReturningOk::default()); + let state_manager = Arc::new(RefMockStateManager::default()); + let log = ic_logger::replica_logger::no_op_logger(); + let dkg_pool = Arc::new(RwLock::new(DkgPoolImpl::new( + ic_metrics::MetricsRegistry::new(), + log.clone(), + Height::from(0), + ))); + let idkg_pool = Arc::new(RwLock::new(IDkgPoolImpl::new( + self.replica_config.node_id, + self.pool_config.clone(), + log.clone(), + ic_metrics::MetricsRegistry::new(), + Box::new(IDkgStatsNoOp {}), + ))); + let canister_http_pool = Arc::new(RwLock::new(CanisterHttpPoolImpl::new( + ic_metrics::MetricsRegistry::new(), + log, + ))); + let pool = TestConsensusPool::new( + self.replica_config.node_id, + self.replica_config.subnet_id, + self.pool_config, + time_source.clone(), + registry.clone(), + crypto.clone(), + state_manager.clone(), + Some(dkg_pool.clone()), + ); + let membership = Arc::new(Membership::new( + pool.get_cache(), + registry.clone(), + self.replica_config.subnet_id, + )); + + if self.mocked_state_manager { + state_manager + .get_mut() + .expect_get_state_at() + .return_const(Ok(ic_interfaces_state_manager::Labeled::new( + Height::new(0), + Arc::new(ic_test_utilities_state::get_initial_state(0, 0)), + ))); + } + + Dependencies { + crypto, + registry, + registry_data_provider, + membership, + time_source, + pool, + replica_config: self.replica_config, + state_manager, + dkg_pool, + idkg_pool, + canister_http_pool, + } + } +} + /// Creates most common consensus components used for testing. All components /// share the same mocked registry with the provided records, so they refer to /// the identical registry content at any time. The MockStateManager instance @@ -114,68 +283,14 @@ pub fn dependencies_with_subnet_records_with_raw_state_manager( subnet_id: SubnetId, records: Vec<(u64, SubnetRecord)>, ) -> Dependencies { - let time_source = FastForwardTimeSource::new(); - let registry_version = RegistryVersion::from(records[0].clone().0); - let (registry_data_provider, registry) = setup_registry_non_final(subnet_id, records); - registry_data_provider - .add( - ROOT_SUBNET_ID_KEY, - registry_version, - Some(ic_types::subnet_id_into_protobuf(subnet_test_id(0))), - ) - .unwrap(); - registry.update_to_latest_version(); - let replica_config = ReplicaConfig { - subnet_id, - node_id: node_test_id(0), - }; - let crypto = Arc::new(CryptoReturningOk::default()); - let state_manager = Arc::new(RefMockStateManager::default()); - let log = ic_logger::replica_logger::no_op_logger(); - let dkg_pool = Arc::new(RwLock::new(DkgPoolImpl::new( - ic_metrics::MetricsRegistry::new(), - log.clone(), - Height::from(0), - ))); - let idkg_pool = Arc::new(RwLock::new(IDkgPoolImpl::new( - replica_config.node_id, - pool_config.clone(), - log.clone(), - ic_metrics::MetricsRegistry::new(), - Box::new(IDkgStatsNoOp {}), - ))); - let canister_http_pool = Arc::new(RwLock::new(CanisterHttpPoolImpl::new( - ic_metrics::MetricsRegistry::new(), - log, - ))); - let pool = TestConsensusPool::new( - replica_config.node_id, - subnet_id, + DependenciesBuilder::new( pool_config, - time_source.clone(), - registry.clone(), - crypto.clone(), - state_manager.clone(), - Some(dkg_pool.clone()), - ); - let membership = Arc::new(Membership::new( - pool.get_cache(), - registry.clone(), - subnet_id, - )); - Dependencies { - crypto, - registry, - registry_data_provider, - membership, - time_source, - pool, - replica_config, - state_manager, - dkg_pool, - idkg_pool, - canister_http_pool, - } + records + .into_iter() + .map(|(version, record)| (version, subnet_id, record)) + .collect(), + ) + .build() } /// Creates most common consensus components used for testing. All components @@ -187,42 +302,15 @@ pub fn dependencies_with_subnet_params( subnet_id: SubnetId, records: Vec<(u64, SubnetRecord)>, ) -> Dependencies { - let Dependencies { - time_source, - registry_data_provider, - registry, - membership, - crypto, - pool, - replica_config, - state_manager, - dkg_pool, - idkg_pool, - canister_http_pool, - .. - } = dependencies_with_subnet_records_with_raw_state_manager(pool_config, subnet_id, records); - - state_manager - .get_mut() - .expect_get_state_at() - .return_const(Ok(ic_interfaces_state_manager::Labeled::new( - Height::new(0), - Arc::new(ic_test_utilities_state::get_initial_state(0, 0)), - ))); - - Dependencies { - crypto, - registry, - registry_data_provider, - membership, - time_source, - pool, - replica_config, - state_manager, - dkg_pool, - idkg_pool, - canister_http_pool, - } + DependenciesBuilder::new( + pool_config, + records + .into_iter() + .map(|(version, record)| (version, subnet_id, record)) + .collect(), + ) + .with_mocked_state_manager() + .build() } /// Creates most common consensus components used for testing. All components @@ -231,9 +319,14 @@ pub fn dependencies_with_subnet_params( /// their default values. pub fn dependencies(pool_config: ArtifactPoolConfig, nodes: u64) -> Dependencies { let committee = (0..nodes).map(node_test_id).collect::>(); - dependencies_with_subnet_params( + DependenciesBuilder::new( pool_config, - subnet_test_id(0), - vec![(1, SubnetRecordBuilder::from(&committee).build())], + vec![( + 1, + subnet_test_id(0), + SubnetRecordBuilder::from(&committee).build(), + )], ) + .with_mocked_state_manager() + .build() } diff --git a/rs/consensus/src/consensus.rs b/rs/consensus/src/consensus.rs index 7563f0cee5b5..6c4338c66f6c 100644 --- a/rs/consensus/src/consensus.rs +++ b/rs/consensus/src/consensus.rs @@ -253,6 +253,7 @@ impl ConsensusImpl { crypto.clone(), state_manager.clone(), message_routing.clone(), + Arc::clone(®istry_client), logger.clone(), ), block_maker: BlockMaker::new( @@ -287,6 +288,8 @@ impl ConsensusImpl { membership, message_routing.clone(), crypto.clone(), + registry_client.clone(), + replica_config.clone(), logger.clone(), ), purger: Purger::new( @@ -693,8 +696,8 @@ mod tests { let metrics_registry = MetricsRegistry::new(); let consensus_impl = ConsensusImpl::new( - replica_config, - registry, + replica_config.clone(), + registry.clone(), pool.get_cache(), crypto.clone(), Arc::new(FakeIngressSelector::new()), @@ -710,6 +713,8 @@ mod tests { crypto, no_op_logger(), &PoolReader::new(&pool), + registry, + replica_config, ))), Arc::new(FakeMessageRouting::new()), state_manager, diff --git a/rs/consensus/src/consensus/batch_delivery.rs b/rs/consensus/src/consensus/batch_delivery.rs index 6773ccc1bba6..6fec5d4050ff 100644 --- a/rs/consensus/src/consensus/batch_delivery.rs +++ b/rs/consensus/src/consensus/batch_delivery.rs @@ -9,7 +9,7 @@ use crate::consensus::{ use ic_consensus_chain_key::ChainKeyPayloadBuilderImpl; use ic_consensus_dkg::get_vetkey_public_keys; use ic_consensus_idkg::utils::get_idkg_subnet_public_keys_and_pre_signatures; -use ic_consensus_utils::{membership::Membership, pool_reader::PoolReader}; +use ic_consensus_utils::{membership::Membership, pool_reader::PoolReader, subnet_splitting}; use ic_error_types::RejectCode; use ic_https_outcalls_consensus::payload_builder::CanisterHttpPayloadBuilderImpl; use ic_interfaces::{ @@ -24,20 +24,18 @@ use ic_protobuf::{ registry::{crypto::v1::PublicKey as PublicKeyProto, subnet::v1::InitialNiDkgTranscriptRecord}, }; use ic_types::{ - Height, PrincipalId, SubnetId, + Height, NodeId, PrincipalId, SubnetId, batch::{ Batch, BatchContent, BatchMessages, BatchSummary, BlockmakerMetrics, CanisterHttpSpent, ChainKeyData, ConsensusResponse, }, - consensus::{ - Block, BlockPayload, HasVersion, - dkg::RemoteTranscriptResult, - idkg::{self}, - }, - crypto::randomness_from_crypto_hashable, - crypto::threshold_sig::{ - ThresholdSigPublicKey, - ni_dkg::{NiDkgId, NiDkgTag, NiDkgTranscript}, + consensus::{Block, BlockPayload, HasVersion, dkg::RemoteTranscriptResult, idkg}, + crypto::{ + randomness_from_crypto_hashable, + threshold_sig::{ + ThresholdSigPublicKey, + ni_dkg::{NiDkgId, NiDkgTag, NiDkgTranscript}, + }, }, messages::{CallbackId, Payload, RejectContext}, }; @@ -64,6 +62,7 @@ pub fn deliver_batches( pool, registry_client, subnet_id, + /*maybe_node_id=*/ None, log, max_batch_height_to_deliver, /*result_processor=*/ None, @@ -80,6 +79,7 @@ pub(crate) fn deliver_batches_with_result_processor( pool: &PoolReader<'_>, registry_client: &dyn RegistryClient, subnet_id: SubnetId, + maybe_node_id: Option, log: &ReplicaLogger, // This argument should only be used by the ic-replay tool. If it is set to `None`, we will // deliver all batches until the finalized height. If it is set to `Some(h)`, we will @@ -166,11 +166,10 @@ pub(crate) fn deliver_batches_with_result_processor( log, ) { Some(Status::Halting | Status::Halted) => { - debug!( - every_n_seconds => 5, + info!( + every_n_seconds => 30, log, - "Batch of height {} is not delivered because replica is halted", - height, + "Batch of height {height} is not delivered because replica is halted" ); return Ok(last_delivered_batch_height); } @@ -224,13 +223,50 @@ pub(crate) fn deliver_batches_with_result_processor( let persist_batch = Some(height) == max_batch_height_to_deliver; let requires_full_state_hash = block.payload.is_summary() || persist_batch; let batch_content = match block.payload.as_ref() { - BlockPayload::Summary(_summary_payload) => BatchContent::Data { - batch_messages: BatchMessages::default(), - chain_key_data, - consensus_responses, - canister_http_spent, - requires_full_state_hash, - }, + BlockPayload::Summary(_summary_payload) => { + if let Some(scheduled) = subnet_splitting::is_split_scheduled(&block) { + let node_id = + maybe_node_id.expect("Subnet splitting not yet enabled in ic-replay"); + let subnet_splitting::PostSplitAssignment { + new_subnet_id, + other_subnet_id, + } = match subnet_splitting::get_post_split_subnet_assignment( + node_id, + &block, + registry_client, + scheduled, + ) { + Ok(assignment) => assignment, + Err(err) => { + warn!( + every_n_seconds => 30, + log, + "Error getting new subnet assignment: {}", + err + ); + break; + } + }; + + info!( + log, + "Delivering splitting block. New subnet assignment: {}", new_subnet_id + ); + + BatchContent::Splitting { + new_subnet_id, + other_subnet_id, + } + } else { + BatchContent::Data { + batch_messages: BatchMessages::default(), + chain_key_data, + consensus_responses, + canister_http_spent, + requires_full_state_hash, + } + } + } BlockPayload::Data(data_payload) => { batch_stats.add_from_payload(&data_payload.batch); BatchContent::Data { @@ -238,9 +274,8 @@ pub(crate) fn deliver_batches_with_result_processor( .batch .clone() .into_messages() - .map_err(|err| { - error!(log, "batch payload deserialization failed: {:?}", err); - err + .inspect_err(|err| { + error!(log, "batch payload deserialization failed: {err:?}"); }) .unwrap_or_default(), chain_key_data, @@ -255,8 +290,7 @@ pub(crate) fn deliver_batches_with_result_processor( warn!( every_n_seconds => 5, log, - "No batch delivery at height {}: no random beacon found.", - height + "No batch delivery at height {height}: no random beacon found." ); return Ok(last_delivered_batch_height); }; @@ -270,9 +304,7 @@ pub(crate) fn deliver_batches_with_result_processor( warn!( every_n_seconds => 5, log, - "No batch delivery at height {}: membership error: {:?}", - height, - e + "No batch delivery at height {height}: membership error: {e:?}" ); return Ok(last_delivered_batch_height); } @@ -304,7 +336,7 @@ pub(crate) fn deliver_batches_with_result_processor( f(&result, block_stats, batch_stats); } if let Err(err) = result { - warn!(every_n_seconds => 5, log, "Batch delivery failed: {:?}", err); + warn!(every_n_seconds => 5, log, "Batch delivery failed: {err:?}"); return Err(err); } last_delivered_batch_height = height; @@ -577,16 +609,20 @@ mod tests { //! Finalizer unit tests use super::*; use crate::consensus::batch_delivery::generate_responses_to_remote_dkgs; + use ic_consensus_mocks::{Dependencies, DependenciesBuilder}; use ic_crypto_test_utils_ni_dkg::dummy_transcript_for_tests; use ic_logger::replica_logger::no_op_logger; use ic_management_canister_types_private::{SetupInitialDKGResponse, VetKdCurve, VetKdKeyId}; + use ic_test_utilities::message_routing::FakeMessageRouting; + use ic_test_utilities_registry::SubnetRecordBuilder; use ic_test_utilities_types::ids::subnet_test_id; use ic_types::{ PrincipalId, RegistryVersion, SubnetId, + backwards_compatibility::BackwardsCompatible, batch::{BatchPayload, ValidationContext}, consensus::{ - DataPayload, Payload as ConsensusPayload, Rank, - dkg::{DkgDataPayload, RemoteTranscriptResult}, + DataPayload, HashedBlock, Payload as ConsensusPayload, Rank, + dkg::{DkgDataPayload, RemoteTranscriptResult, SplittingArgs, SubnetSplittingStatus}, }, crypto::{ CryptoHash, CryptoHashOf, @@ -595,10 +631,15 @@ mod tests { }, }, messages::{CallbackId, Payload}, + replica_config::ReplicaConfig, time::UNIX_EPOCH, }; + use ic_types_test_utils::ids::{NODE_1, NODE_2, NODE_3, NODE_4, SUBNET_1, SUBNET_2}; use std::str::FromStr; + const SOURCE_SUBNET_ID: SubnetId = SUBNET_1; + const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2; + const TARGET_ID: NiDkgTargetId = NiDkgTargetId::new([8; 32]); const EXPECTED_FRESH_SUBNET_ID_STR: &str = @@ -781,4 +822,112 @@ mod tests { SubnetId::from(PrincipalId::from_str(EXPECTED_FRESH_SUBNET_ID_STR).unwrap()) ); } + + fn deliver_splitting_batch(node_id: NodeId) -> (SubnetId, SubnetId) { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + const SPLITTING_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2); + const INTERVAL_LENGTH: u64 = 9; + let summary_height = Height::from(INTERVAL_LENGTH + 1); + + let Dependencies { + mut pool, + membership, + registry, + .. + } = DependenciesBuilder::new( + pool_config, + vec![ + ( + 1, + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_3]) + .with_dkg_interval_length(INTERVAL_LENGTH) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + DESTINATION_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_2, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH) + .build(), + ), + ], + ) + .with_replica_config(ReplicaConfig { + node_id: NODE_1, + subnet_id: SOURCE_SUBNET_ID, + }) + .with_mocked_state_manager() + .build(); + + pool.advance_round_normal_operation_n(INTERVAL_LENGTH); + + let mut proposal = pool.make_next_block(); + let block = proposal.content.as_mut(); + block.context.registry_version = SPLITTING_REGISTRY_VERSION; + let mut payload = block.payload.as_ref().as_summary().clone(); + payload.dkg.subnet_splitting_status = BackwardsCompatible::new_for_test_only(Some( + SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: SOURCE_SUBNET_ID, + destination_subnet_id: DESTINATION_SUBNET_ID, + }), + )); + block.payload = ConsensusPayload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(payload), + ); + proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone()); + pool.insert_validated(proposal.clone()); + pool.notarize(&proposal); + pool.finalize(&proposal); + pool.insert_random_tape(summary_height); + + let message_routing = FakeMessageRouting::new(); + *message_routing.next_batch_height.write().unwrap() = summary_height; + + let result = deliver_batches_with_result_processor( + &message_routing, + &membership, + &PoolReader::new(&pool), + registry.as_ref(), + SOURCE_SUBNET_ID, + Some(node_id), + &no_op_logger(), + None, + None, + ); + + assert_eq!(result, Ok(summary_height)); + let batches = message_routing.batches.read().unwrap(); + assert_eq!(batches.len(), 1); + match &batches[0].content { + BatchContent::Splitting { + new_subnet_id, + other_subnet_id, + } => (*new_subnet_id, *other_subnet_id), + other => panic!("Expected BatchContent::Splitting, got: {other:?}"), + } + }) + } + + #[test] + fn test_deliver_splitting_batch_node_on_source_subnet() { + let (new_subnet_id, other_subnet_id) = deliver_splitting_batch(NODE_1); + assert_eq!(new_subnet_id, SOURCE_SUBNET_ID); + assert_eq!(other_subnet_id, DESTINATION_SUBNET_ID); + } + + #[test] + fn test_deliver_splitting_batch_node_on_destination_subnet() { + let (new_subnet_id, other_subnet_id) = deliver_splitting_batch(NODE_4); + assert_eq!(new_subnet_id, DESTINATION_SUBNET_ID); + assert_eq!(other_subnet_id, SOURCE_SUBNET_ID); + } } diff --git a/rs/consensus/src/consensus/block_maker.rs b/rs/consensus/src/consensus/block_maker.rs index 3edf4519422c..8ae468638a51 100644 --- a/rs/consensus/src/consensus/block_maker.rs +++ b/rs/consensus/src/consensus/block_maker.rs @@ -381,16 +381,28 @@ impl BlockMaker { pool, &self.log, )? { - // Don't propose any block if the replica is halted. Status::Halted => { + info!( + every_n_seconds => 30, + self.log, + "Not proposing any block at height {height} \ + because the replica is halted" + ); return None; } - // Use empty payload and empty DKG dealings if the replica is halting. - Status::Halting => ( - BatchPayload::default(), - DkgDataPayload::new_empty(dkg.start_height), - /*idkg_data=*/ None, - ), + Status::Halting => { + info!( + every_n_seconds => 30, + self.log, + "Proposing an empty block at height {height} \ + because the replica is halting" + ); + ( + BatchPayload::default(), + DkgDataPayload::new_empty(dkg.start_height), + /*idkg_data=*/ None, + ) + } Status::Running => { let batch_payload = self.build_batch_payload( pool, diff --git a/rs/consensus/src/consensus/catchup_package_maker.rs b/rs/consensus/src/consensus/catchup_package_maker.rs index a98e40af549b..836486aa7404 100644 --- a/rs/consensus/src/consensus/catchup_package_maker.rs +++ b/rs/consensus/src/consensus/catchup_package_maker.rs @@ -14,35 +14,60 @@ //! block is considered finalized. use crate::consensus::status; +use ic_consensus_dkg::payload_builder::get_post_split_dkg_summary; use ic_consensus_utils::{ active_high_threshold_nidkg_id, crypto::ConsensusCrypto, get_oldest_state_registry_version, membership::Membership, pool_reader::PoolReader, }; use ic_interfaces::messaging::MessageRouting; +use ic_interfaces_registry::RegistryClient; use ic_interfaces_state_manager::{ PermanentStateHashError::*, StateHashError, StateManager, TransientStateHashError::*, }; -use ic_logger::{ReplicaLogger, debug, error, trace}; +use ic_logger::{ReplicaLogger, debug, error, info, trace, warn}; +use ic_registry_client_helpers::node::NodeRegistry; use ic_replicated_state::ReplicatedState; use ic_types::{ + Height, NodeId, SubnetId, + batch::ValidationContext, consensus::{ - Block, CatchUpContent, CatchUpPackage, CatchUpPackageShare, CatchUpShareContent, - HasCommittee, HasHeight, HashedBlock, HashedRandomBeacon, + Block, BlockPayload, CatchUpContent, CatchUpPackage, CatchUpPackageShare, + CatchUpShareContent, HasCommittee, HasHeight, HashedBlock, HashedRandomBeacon, Payload, + RandomBeacon, RandomBeaconContent, Rank, SummaryPayload, + dkg::{SplittingArgs, SubnetSplittingStatus}, + }, + crypto::{ + CombinedThresholdSig, CombinedThresholdSigOf, CryptoHash, CryptoHashOf, Signed, + crypto_hash, + threshold_sig::ni_dkg::{NiDkgId, NiDkgTag, NiDkgTranscript}, }, replica_config::ReplicaConfig, + signature::ThresholdSignature, }; use std::sync::Arc; -/// CatchUpPackage maker is responsible for creating beacon shares +/// [`CatchUpPackage`] maker is responsible for creating beacon shares pub(crate) struct CatchUpPackageMaker { replica_config: ReplicaConfig, membership: Arc, crypto: Arc, state_manager: Arc>, message_routing: Arc, + registry: Arc, log: ReplicaLogger, } +/// Type of [`CatchUpPackage`]. +#[derive(Copy, Clone, Eq, PartialEq, Debug)] +pub(crate) enum CatchUpPackageType { + Normal, + /// After deliverying a splitting block to the DSM, we immediately create a CUP at the start of + /// the next dkg interval and we create a new summary block and a dummy random beacon on the fly. + PostSplit { + new_subnet_id: SubnetId, + }, +} + impl CatchUpPackageMaker { /// Instantiate a new CatchUpPackage maker and save a copy of the config. pub fn new( @@ -51,6 +76,7 @@ impl CatchUpPackageMaker { crypto: Arc, state_manager: Arc>, message_routing: Arc, + registry: Arc, log: ReplicaLogger, ) -> Self { Self { @@ -59,6 +85,7 @@ impl CatchUpPackageMaker { crypto, state_manager, message_routing, + registry, log, } } @@ -144,38 +171,23 @@ impl CatchUpPackageMaker { } /// Consider the provided block for the creation of a catch up package. - fn consider_block( + pub(crate) fn consider_block( &self, pool: &PoolReader<'_>, start_block: Block, ) -> Option { - let height = start_block.height(); - - // Skip if this node is not in the committee to make CUP shares - let my_node_id = self.replica_config.node_id; - if self.membership.node_belongs_to_threshold_committee( - my_node_id, - height, - CatchUpPackage::committee(), - ) != Ok(true) - { - return None; - } - - // Skip if this node has already made a share - if pool - .get_catch_up_package_shares(height) - .any(|share| share.signature.signer == my_node_id) - { - return None; - } - - // Skip if random beacon does not exist for the height - let random_beacon = pool.get_random_beacon(height)?; + let summary_height = start_block.height(); + let cup_type = get_catch_up_package_type( + self.registry.as_ref(), + self.replica_config.node_id, + &start_block, + ) + .inspect_err(|err| warn!(self.log, "Failed to get the catch up package type: {err}")) + .ok()?; let halting = || { status::should_halt( - height, + summary_height, Some(&start_block), self.membership.registry_client.as_ref(), self.membership.subnet_id, @@ -206,95 +218,365 @@ impl CatchUpPackageMaker { // It is not a problem to make this exception, because when we are halting, all blocks have // empty payloads, and thus do not need to access states and payloads at the validation // context's certified height. - if pool.get_finalized_tip().context.certified_height < height && !halting() { + if pool.get_finalized_tip().context.certified_height < summary_height && !halting() { return None; } - match self.state_manager.get_state_hash_at(height) { + let state_hash = match self.state_manager.get_state_hash_at(summary_height) { + Ok(state_hash) => state_hash, Err(StateHashError::Transient(StateNotCommittedYet(_))) => { // TODO: Setup a delay before retry debug!( self.log, - "Cannot make CUP at height {} because state is not committed yet. Will retry", - height + "Cannot make CUP at height {} because \ + state is not committed yet. Will retry", + summary_height ); - None + return None; } Err(StateHashError::Transient(HashNotComputedYet(_))) => { debug!( self.log, - "Cannot make CUP at height {} because state hash is not computed yet. Will retry", - height + "Cannot make CUP at height {} because \ + state hash is not computed yet. Will retry", + summary_height ); - None + return None; } Err(StateHashError::Permanent(StateRemoved(_))) => { // This should never happen as we don't want to remove the state // for CUP before the hash is fetched. panic!( - "State at height {height} had disappeared before we had a chance to make a CUP. This should not happen.", + "State at height {summary_height} had disappeared before \ + we had a chance to make a CUP. \ + This should not happen.", ); } Err(StateHashError::Permanent(StateNotFullyCertified(_))) => { - panic!("Height {height} is not a fully certified height. This should not happen.",); + panic!( + "Height {summary_height} is not a fully certified height. \ + This should not happen.", + ); } - Ok(state_hash) => { - // Should succeed as we already got the hash above - let state = self - .state_manager - .get_state_at(height) - .map_err(|err| { - error!( - self.log, - "Cannot make CUP at height {}: `get_state_hash_at` \ - succeeded but `get_state_at` failed with {}. Will retry", - height, - err, - ) - }) - .ok()?; - let registry_version = get_oldest_state_registry_version(state.get_ref()); - let content = CatchUpContent::new( - HashedBlock::new(ic_types::crypto::crypto_hash, start_block), - HashedRandomBeacon::new(ic_types::crypto::crypto_hash, random_beacon), - state_hash, - registry_version, + }; + + // Should succeed as we already got the hash above + let state = self + .state_manager + .get_state_at(summary_height) + .map_err(|err| { + error!( + self.log, + "Cannot make CUP at height {summary_height}: `get_state_hash_at` \ + succeeded but `get_state_at` failed with {err}. Will retry", + ) + }) + .ok()?; + let oldest_registry_version_in_use_by_replicated_state = + get_oldest_state_registry_version(state.get_ref()); + + let cup_block = self + .get_cup_block(start_block, cup_type) + .inspect_err(|err| warn!(self.log, "Can't get a block for a CUP: {err}")) + .ok()?; + + let random_beacon = self + .get_cup_random_beacon(pool, &cup_block, cup_type) + .inspect_err(|err| warn!(self.log, "Can't get a random beacon for a CUP: {err}")) + .ok()?; + + let high_dkg_id = self + .get_high_dkg_id(pool, &cup_block, cup_type) + .inspect_err(|err| warn!(self.log, "Can't get a high dkg id for a CUP: {err}")) + .ok()?; + + if !self + .node_belongs_to_threshold_committee(&cup_block, cup_type) + .inspect_err(|err| warn!(self.log, "Can't check if node belongs to committee: {err}")) + .unwrap_or_default() + { + return None; + } + + // Skip if this node has already made a share + if pool + .get_catch_up_package_shares(cup_block.height()) + .any(|share| share.signature.signer == self.replica_config.node_id) + { + return None; + } + + let content = CatchUpContent::new( + HashedBlock::new(ic_types::crypto::crypto_hash, cup_block), + HashedRandomBeacon::new(ic_types::crypto::crypto_hash, random_beacon), + state_hash, + oldest_registry_version_in_use_by_replicated_state, + ); + + let share_content = CatchUpShareContent::from(&content); + let share_height = share_content.height(); + match self + .crypto + .sign(&content, self.replica_config.node_id, high_dkg_id) + { + Ok(signature) => { + info!( + self.log, + "Proposing a CatchUpPackageShare (type: {cup_type:?}) at height {share_height}" ); - let share_content = CatchUpShareContent::from(&content); - if let Some(dkg_id) = active_high_threshold_nidkg_id(pool.as_cache(), height) { - match self.crypto.sign(&content, my_node_id, dkg_id) { - Ok(signature) => { - // Caution: The log string below is checked in replica_determinism_test. - // Changing the string might break the test. - debug!( - self.log, - "Proposing a CatchUpPackageShare at height {}", height - ); - Some(CatchUpPackageShare { - content: share_content, - signature, - }) - } - Err(err) => { - error!(self.log, "Couldn't create a signature: {:?}", err); - None - } - } - } else { - error!(self.log, "Couldn't find transcript at height {}", height); - None + Some(CatchUpPackageShare { + content: share_content, + signature, + }) + } + Err(err) => { + error!( + self.log, + "Couldn't create a signature at height {share_height}: {err}" + ); + None + } + } + } + + fn get_cup_block( + &self, + summary_block: Block, + cup_type: CatchUpPackageType, + ) -> Result { + match cup_type { + CatchUpPackageType::Normal => Ok(summary_block), + CatchUpPackageType::PostSplit { new_subnet_id } => create_post_split_summary_block( + &summary_block, + new_subnet_id, + self.registry.as_ref(), + ) + .map_err(|err| format!("Failed to create a post split block: {err}")), + } + } + + fn get_cup_random_beacon( + &self, + pool: &PoolReader<'_>, + cup_block: &Block, + cup_type: CatchUpPackageType, + ) -> Result { + match cup_type { + CatchUpPackageType::Normal => pool + .get_random_beacon(cup_block.height()) + .ok_or_else(|| format!("No random beacon found at height {}", cup_block.height())), + // During subnet splitting we create a dummy, unsigned random beacon, because at the + // height at which we are building a CUP, we won't have a random beacon. + CatchUpPackageType::PostSplit { .. } => create_post_split_random_beacon(cup_block), + } + } + + fn get_high_dkg_id( + &self, + pool: &PoolReader<'_>, + cup_block: &Block, + cup_type: CatchUpPackageType, + ) -> Result { + // TODO: can we always take the transcript from the block? + match cup_type { + CatchUpPackageType::Normal => { + active_high_threshold_nidkg_id(pool.as_cache(), cup_block.height).ok_or_else(|| { + format!("Couldn't find transcript at height {}", cup_block.height) + }) + } + CatchUpPackageType::PostSplit { .. } => { + match get_current_transcript_from_summary_block(cup_block, &NiDkgTag::HighThreshold) + { + Some(transcript) => Ok(transcript.dkg_id.clone()), + None => Err(format!( + "Couldn't find post-split transcript at height {}", + cup_block.height + )), + } + } + } + } + + fn node_belongs_to_threshold_committee( + &self, + cup_block: &Block, + cup_type: CatchUpPackageType, + ) -> Result { + // TODO: can we always take the transcript from the block? + match cup_type { + CatchUpPackageType::Normal => self + .membership + .node_belongs_to_threshold_committee( + self.replica_config.node_id, + cup_block.height, + CatchUpPackage::committee(), + ) + .map_err(|err| { + format!("Failed to check if node belongs to threshold committee {err:?}") + }), + CatchUpPackageType::PostSplit { .. } => { + match get_current_transcript_from_summary_block(cup_block, &NiDkgTag::HighThreshold) + { + Some(transcript) => Ok(transcript + .committee + .get() + .contains(&self.replica_config.node_id)), + None => Err(format!( + "Couldn't find post-split transcript at height {}", + cup_block.height + )), } } } } } +pub(crate) fn get_catch_up_package_type( + registry: &dyn RegistryClient, + node_id: NodeId, + summary_block: &Block, +) -> Result { + match summary_block + .payload + .as_ref() + .as_summary() + .dkg + .subnet_splitting_status() + { + SubnetSplittingStatus::Scheduled(SplittingArgs { + destination_subnet_id, + source_subnet_id, + }) => { + let new_subnet_id = get_new_subnet_id( + registry, + summary_block, + node_id, + source_subnet_id, + destination_subnet_id, + ) + .map_err(|err| format!("Failed to get the new subnet assignment: {err}"))?; + + Ok(CatchUpPackageType::PostSplit { new_subnet_id }) + } + SubnetSplittingStatus::NotScheduled | SubnetSplittingStatus::PostSplit(..) => { + Ok(CatchUpPackageType::Normal) + } + } +} + +fn get_current_transcript_from_summary_block<'a>( + summary_block: &'a Block, + tag: &NiDkgTag, +) -> Option<&'a NiDkgTranscript> { + summary_block + .payload + .as_ref() + .as_summary() + .dkg + .current_transcript(tag) +} + +pub(crate) fn create_post_split_summary_block( + splitting_summary_block: &Block, + subnet_id: SubnetId, + registry: &dyn RegistryClient, +) -> Result { + let post_split_dkg_summary = + get_post_split_dkg_summary(subnet_id, registry, splitting_summary_block) + .map_err(|err| format!("Failed to get post-split DKG summary: {err}"))?; + + let post_split_height = post_split_dkg_summary.height; + Ok(Block { + version: splitting_summary_block.version.clone(), + // Fake parent + parent: CryptoHashOf::from(CryptoHash(Vec::new())), + payload: Payload::new( + crypto_hash, + BlockPayload::Summary(SummaryPayload { + dkg: post_split_dkg_summary, + // Copy over the IDKG summary from the splitting block + idkg: splitting_summary_block + .payload + .as_ref() + .as_summary() + .idkg + .clone(), + }), + ), + height: post_split_height, + rank: Rank(0), + context: ValidationContext { + registry_version: splitting_summary_block.context.registry_version, + certified_height: post_split_height, + // time needs to be strictly increasing + time: splitting_summary_block.context.time + std::time::Duration::from_millis(1), + }, + }) +} + +// During subnet splitting we create a dummy, unsigned random beacon, because at the +// height at which we are building a CUP, we won't have a random beacon. +pub(crate) fn create_post_split_random_beacon(cup_block: &Block) -> Result { + let transcript = get_current_transcript_from_summary_block(cup_block, &NiDkgTag::LowThreshold) + .ok_or_else(|| { + format!( + "Couldn't find post-split transcript at height {}", + cup_block.height(), + ) + })?; + + Ok(Signed { + content: RandomBeaconContent { + version: cup_block.version.clone(), + height: cup_block.height(), + parent: CryptoHashOf::from(CryptoHash(Vec::new())), + }, + signature: ThresholdSignature { + signer: transcript.dkg_id.clone(), + signature: CombinedThresholdSigOf::new(CombinedThresholdSig(vec![])), + }, + }) +} + +fn get_new_subnet_id( + registry: &dyn RegistryClient, + summary_block: &Block, + node_id: NodeId, + source_subnet_id: SubnetId, + destination_subnet_id: SubnetId, +) -> Result { + let registry_version = summary_block.context.registry_version; + let new_subnet_id = registry + .get_subnet_id_from_node_id(node_id, registry_version) + .map_err(|err| { + format!( + "Failed to get the new subnet id at \ + registry version {registry_version}: {err}" + ) + })? + .ok_or_else(|| { + format!( + "Node is not assigned to any subnet at \ + registry version {registry_version}" + ) + })?; + + if ![source_subnet_id, destination_subnet_id].contains(&new_subnet_id) { + return Err(format!( + "According to the registry version {registry_version} \ + the node belongs to neither source subnet nor the destination subnet" + )); + } + + Ok(new_subnet_id) +} + #[cfg(test)] mod tests { //! CatchUpPackageMaker unit tests use super::*; use ic_consensus_mocks::{ - Dependencies, dependencies_with_subnet_params, + Dependencies, DependenciesBuilder, dependencies_with_subnet_params, dependencies_with_subnet_records_with_raw_state_manager, }; use ic_logger::replica_logger::no_op_logger; @@ -312,13 +594,20 @@ mod tests { fake_signature_request_context_with_registry_version, }, }; - use ic_test_utilities_registry::SubnetRecordBuilder; + use ic_test_utilities_logger::with_test_replica_logger; + use ic_test_utilities_registry::{SubnetRecordBuilder, insert_initial_dkg_transcript}; use ic_test_utilities_types::ids::{node_test_id, subnet_test_id}; use ic_types::{ - CryptoHashOfState, Height, RegistryVersion, - consensus::{BlockPayload, BlockProposal, Payload, SummaryPayload, idkg::PreSigId}, + CryptoHashOfState, Height, NodeId, RegistryVersion, + backwards_compatibility::BackwardsCompatible, + consensus::{ + BlockPayload, BlockProposal, ConsensusMessageHashable, HasVersion, Payload, + SummaryPayload, idkg::PreSigId, + }, crypto::CryptoHash, }; + use ic_types_test_utils::ids::{NODE_1, NODE_2, NODE_3, NODE_4}; + use ic_types_test_utils::ids::{SUBNET_1, SUBNET_2}; use rstest::rstest; use std::sync::{Arc, RwLock}; @@ -379,6 +668,7 @@ mod tests { deps.crypto.clone(), deps.state_manager.clone(), message_routing, + deps.registry.clone(), no_op_logger(), ); @@ -605,6 +895,7 @@ mod tests { membership, replica_config, crypto, + registry, state_manager, .. } = dependencies_with_subnet_records_with_raw_state_manager( @@ -644,6 +935,7 @@ mod tests { crypto, state_manager.clone(), message_routing, + registry, no_op_logger(), ); @@ -699,6 +991,7 @@ mod tests { mut pool, membership, replica_config, + registry, crypto, state_manager, .. @@ -740,6 +1033,7 @@ mod tests { crypto, state_manager, message_routing, + registry, no_op_logger(), ); @@ -760,6 +1054,7 @@ mod tests { replica_config, crypto, state_manager, + registry, .. } = dependencies_with_subnet_params( pool_config, @@ -784,6 +1079,7 @@ mod tests { crypto, state_manager.clone(), message_routing, + registry, no_op_logger(), ); @@ -829,4 +1125,265 @@ mod tests { cup_maker.on_state_change(&PoolReader::new(&pool)); }) } + + // In this test the subnet initially has 4 nodes, and after the split `NODE_1, NODE_2` will stay + // in the original subnet, and `NODE_3, NODE_4` will be moved to a new one. + #[rstest] + #[case::source_subnet_node( + NODE_1, + "d5a517cd0906e1d36b43edf4103ef9b0dfb0e6892a87712ce5ed6602bfa5c97e" + )] + #[case::source_subnet_node( + NODE_2, + "d5a517cd0906e1d36b43edf4103ef9b0dfb0e6892a87712ce5ed6602bfa5c97e" + )] + #[case::destination_subnet_node( + NODE_3, + "e8614bf48bba176a546186f90e7cfc02ec573e4b87296e9d73a70547ca168416" + )] + #[case::destination_subnet_node( + NODE_4, + "e8614bf48bba176a546186f90e7cfc02ec573e4b87296e9d73a70547ca168416" + )] + #[trace] + fn create_post_split_cup_share_test( + #[case] node_id: NodeId, + // We don't necessarily care what the hash is, but we want to ensure that different + // nodes produce different blocks (and hence different hashes), depending on which subnet + // they are going to land on + #[case] expected_block_hash_in_cup: &str, + #[values(Height::new(0), Height::new(1000))] context_certified_height: Height, + ) { + with_test_replica_logger(|log| { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + const SOURCE_SUBNET_ID: SubnetId = SUBNET_1; + const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2; + const INITIAL_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(1); + const SPLITTING_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2); + const INTERVAL_LENGTH: Height = Height::new(9); + let fake_state_hash = CryptoHashOfState::from(CryptoHash(vec![1, 2, 3])); + + let Dependencies { + mut pool, + membership, + registry, + crypto, + state_manager, + replica_config, + .. + } = DependenciesBuilder::new( + pool_config, + vec![ + ( + INITIAL_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + DESTINATION_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ], + ) + .add_additional_registry_mutation(|registry_data_provider| { + insert_initial_dkg_transcript( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + &SubnetRecordBuilder::from(&[NODE_1, NODE_2]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + registry_data_provider, + ) + }) + .with_replica_config(ReplicaConfig { + node_id, + subnet_id: SOURCE_SUBNET_ID, + }) + .with_mocked_state_manager() + .build(); + + state_manager + .get_mut() + .expect_get_state_hash_at() + .return_const(Ok(fake_state_hash.clone())); + + let message_routing = FakeMessageRouting::new(); + *message_routing.next_batch_height.write().unwrap() = Height::from(2); + let message_routing = Arc::new(message_routing); + + let cup_maker = CatchUpPackageMaker::new( + replica_config, + membership, + crypto, + state_manager, + message_routing, + registry, + log, + ); + + pool.advance_round_normal_operation_n(INTERVAL_LENGTH.get()); + + let subnet_splitting_status = SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: SOURCE_SUBNET_ID, + destination_subnet_id: DESTINATION_SUBNET_ID, + }); + let mut proposal = pool.make_next_block(); + let block = proposal.content.as_mut(); + block.context.certified_height = context_certified_height; + block.context.registry_version = SPLITTING_REGISTRY_VERSION; + let mut payload = block.payload.as_ref().as_summary().clone(); + payload.dkg.subnet_splitting_status = + BackwardsCompatible::new_for_test_only(Some(subnet_splitting_status)); + block.payload = Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(payload), + ); + proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone()); + pool.insert_validated(proposal.clone()); + pool.notarize(&proposal); + pool.finalize(&proposal); + + let share = cup_maker + .consider_block(&PoolReader::new(&pool), proposal.content.as_ref().clone()) + .expect("Should succeed with valid inputs"); + + assert!(share.check_integrity()); + assert_eq!(share.content.version, *proposal.content.version()); + assert_eq!( + hex::encode(&share.content.block.get().0), + expected_block_hash_in_cup + ); + assert_eq!( + share.content.random_beacon.get_value().content.height, + proposal.content.height() + INTERVAL_LENGTH + Height::new(1), + ); + assert_eq!( + share.content.random_beacon.get_value().content.version, + *proposal.content.version(), + ); + assert_eq!(share.content.state_hash, fake_state_hash); + assert_eq!( + share + .content + .oldest_registry_version_in_use_by_replicated_state, + None + ); + assert_eq!(share.signature.signer, node_id); + }) + }) + } + + #[test] + fn create_post_split_summary_block_copies_idkg_summary() { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + const SOURCE_SUBNET_ID: SubnetId = SUBNET_1; + const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2; + const INITIAL_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(1); + const SPLITTING_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2); + const INTERVAL_LENGTH: Height = Height::new(9); + + let Dependencies { + mut pool, registry, .. + } = DependenciesBuilder::new( + pool_config, + vec![ + ( + INITIAL_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + DESTINATION_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ], + ) + .add_additional_registry_mutation(|registry_data_provider| { + insert_initial_dkg_transcript( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + &SubnetRecordBuilder::from(&[NODE_1, NODE_2]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + registry_data_provider, + ) + }) + .with_replica_config(ReplicaConfig { + node_id: NODE_1, + subnet_id: SOURCE_SUBNET_ID, + }) + .with_mocked_state_manager() + .build(); + + pool.advance_round_normal_operation_n(INTERVAL_LENGTH.get()); + + let subnet_splitting_status = + ic_types::consensus::dkg::SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: SOURCE_SUBNET_ID, + destination_subnet_id: DESTINATION_SUBNET_ID, + }); + + let mut proposal = pool.make_next_block(); + let block = proposal.content.as_mut(); + block.context.registry_version = SPLITTING_REGISTRY_VERSION; + let mut payload = block.payload.as_ref().as_summary().clone(); + payload.dkg.subnet_splitting_status = + BackwardsCompatible::new_for_test_only(Some(subnet_splitting_status)); + let idkg = empty_idkg_payload(SOURCE_SUBNET_ID); + payload.idkg = Some(idkg.clone()); + block.payload = Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(payload), + ); + proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone()); + pool.insert_validated(proposal.clone()); + pool.notarize(&proposal); + pool.finalize(&proposal); + + let splitting_block = proposal.content.as_ref(); + let post_split_block = create_post_split_summary_block( + splitting_block, + SOURCE_SUBNET_ID, + registry.as_ref(), + ) + .expect("create_post_split_summary_block should succeed"); + + let post_split_idkg = post_split_block + .payload + .as_ref() + .as_summary() + .idkg + .as_ref() + .expect("Post-split summary block should have an IDKG summary"); + + assert_eq!( + *post_split_idkg, idkg, + "IDKG summary in post-split block should match the splitting block's IDKG summary" + ); + }) + } } diff --git a/rs/consensus/src/consensus/finalizer.rs b/rs/consensus/src/consensus/finalizer.rs index f69cc144b7e3..4f6e3d0e701a 100644 --- a/rs/consensus/src/consensus/finalizer.rs +++ b/rs/consensus/src/consensus/finalizer.rs @@ -101,6 +101,7 @@ impl Finalizer { pool, &*self.registry_client, self.replica_config.subnet_id, + Some(self.replica_config.node_id), &self.log, None, Some(&|result, block_stats, batch_stats| { diff --git a/rs/consensus/src/consensus/priority.rs b/rs/consensus/src/consensus/priority.rs index c467ecd31afa..1f15e0b7f7f9 100644 --- a/rs/consensus/src/consensus/priority.rs +++ b/rs/consensus/src/consensus/priority.rs @@ -18,6 +18,7 @@ pub fn new_bouncer( let finalized_height = pool_reader.get_finalized_height(); let notarized_height = pool_reader.get_notarized_height(); let beacon_height = pool_reader.get_random_beacon_height(); + let next_summary_height = pool_reader.get_next_summary_height(); Box::new(move |id: &'_ ConsensusMessageId| { compute_bouncer( @@ -27,6 +28,7 @@ pub fn new_bouncer( finalized_height, notarized_height, beacon_height, + next_summary_height, id, ) }) @@ -44,6 +46,7 @@ fn compute_bouncer( finalized_height: Height, notarized_height: Height, beacon_height: Height, + next_summary_height: Height, id: &ConsensusMessageId, ) -> BouncerValue { let height = id.height; @@ -53,8 +56,10 @@ fn compute_bouncer( } // Stash non-CUP artifacts, as long as they're too far ahead of the next pending CUP height. // This prevents nodes that have fallen behind from exceeding their validated pool bounds. - if !matches!(id.hash, ConsensusMessageHash::CatchUpPackage(_)) - && height > next_cup_height + Height::new(ACCEPTABLE_NOTARIZATION_CUP_GAP) + if !matches!( + id.hash, + ConsensusMessageHash::CatchUpPackage(_) | ConsensusMessageHash::CatchUpPackageShare(_) + ) && height > next_cup_height + Height::new(ACCEPTABLE_NOTARIZATION_CUP_GAP) { return MaybeWantsLater; } @@ -107,7 +112,7 @@ fn compute_bouncer( ConsensusMessageHash::CatchUpPackageShare(_) => { if height <= cup_height { Unwanted - } else if height <= finalized_height { + } else if height <= next_summary_height { Wants } else { MaybeWantsLater diff --git a/rs/consensus/src/consensus/share_aggregator.rs b/rs/consensus/src/consensus/share_aggregator.rs index 3c26940ed4a5..0d92eb88c4f8 100644 --- a/rs/consensus/src/consensus/share_aggregator.rs +++ b/rs/consensus/src/consensus/share_aggregator.rs @@ -2,30 +2,40 @@ //! of shares into full objects. That is, it constructs Random Beacon objects //! from random beacon shares, Notarizations from notarization shares and //! Finalizations from finalization shares. -use crate::consensus::random_tape_maker::RANDOM_TAPE_CHECK_MAX_HEIGHT_RANGE; +use crate::consensus::{ + catchup_package_maker::CatchUpPackageType, + random_tape_maker::RANDOM_TAPE_CHECK_MAX_HEIGHT_RANGE, +}; use ic_consensus_utils::{ active_high_threshold_nidkg_id, active_low_threshold_nidkg_id, aggregate, crypto::ConsensusCrypto, membership::Membership, pool_reader::PoolReader, registry_version_at_height, }; use ic_interfaces::messaging::MessageRouting; -use ic_logger::ReplicaLogger; +use ic_interfaces_registry::RegistryClient; +use ic_logger::{ReplicaLogger, debug, info, warn}; use ic_types::{ Height, consensus::{ - CatchUpContent, ConsensusMessage, ConsensusMessageHashable, FinalizationContent, HasHeight, - RandomTapeContent, + Block, CatchUpContent, CatchUpPackage, ConsensusMessage, ConsensusMessageHashable, + FinalizationContent, HasCommittee, HasHeight, RandomTapeContent, + dkg::{PostSplitArgs, SubnetSplittingStatus}, }, - crypto::Signed, + crypto::threshold_sig::ni_dkg::NiDkgTag, + replica_config::ReplicaConfig, }; use std::{cmp::min, sync::Arc}; +use super::catchup_package_maker; + /// The ShareAggregator is responsible for aggregating shares of random beacons, /// notarizations, and finalizations into full objects pub(crate) struct ShareAggregator { membership: Arc, crypto: Arc, message_routing: Arc, + registry: Arc, + replica_config: ReplicaConfig, log: ReplicaLogger, } @@ -34,12 +44,16 @@ impl ShareAggregator { membership: Arc, message_routing: Arc, crypto: Arc, + registry: Arc, + replica_config: ReplicaConfig, log: ReplicaLogger, ) -> ShareAggregator { ShareAggregator { membership, crypto, message_routing, + registry, + replica_config, log, } } @@ -53,6 +67,7 @@ impl ShareAggregator { messages.append(&mut self.aggregate_notarization_shares(pool)); messages.append(&mut self.aggregate_finalization_shares(pool)); messages.append(&mut self.aggregate_catch_up_package_shares(pool)); + messages } @@ -134,32 +149,33 @@ impl ShareAggregator { let mut start_block = pool.get_highest_finalized_summary_block(); let current_cup_height = pool.get_catch_up_height(); - while start_block.height() > current_cup_height { - let height = start_block.height(); - let shares = pool.get_catch_up_package_shares(height).map(|share| { - let block = pool - .get_block(&share.content.block, height) - .unwrap_or_else(|| panic!("Block not found for {share:?}")); - Signed { - content: CatchUpContent::from_share_content(share.content, block.into_inner()), - signature: share.signature, + loop { + let start_block_height = start_block.height(); + if start_block_height <= current_cup_height { + break; + } + match self.aggregate_catch_up_package_shares_for_summary_block(pool, start_block) { + Ok(Some(cup)) => { + return vec![ConsensusMessage::CatchUpPackage(cup)]; + } + Ok(None) => { + debug!( + self.log, + "Not enough shares to be able to create a full CUP at height{}", + start_block_height + ); + } + Err(err) => { + warn!( + self.log, + "Encountered an error while aggregating CUP shares at height {}: {err}", + start_block_height + ); } - }); - let state_reader = pool.as_cache(); - let dkg_id = active_high_threshold_nidkg_id(state_reader, height); - let result = aggregate( - &self.log, - self.membership.as_ref(), - self.crypto.as_aggregate(), - Box::new(|_| dkg_id.clone()), - shares, - ); - if !result.is_empty() { - return to_messages(result); } let Some(block_from_last_interval) = - pool.get_finalized_block(start_block.height.decrement()) + pool.get_finalized_block(start_block_height.decrement()) else { break; }; @@ -177,6 +193,98 @@ impl ShareAggregator { } Vec::new() } + + fn aggregate_catch_up_package_shares_for_summary_block( + &self, + pool: &PoolReader<'_>, + summary_block: Block, + ) -> Result, String> { + let (threshold, dkg_id, block) = match catchup_package_maker::get_catch_up_package_type( + self.registry.as_ref(), + self.replica_config.node_id, + &summary_block, + ) + .map_err(|err| format!("Failed to determine the cup type: {err}"))? + { + CatchUpPackageType::Normal => { + let threshold = self + .membership + .get_committee_threshold(summary_block.height(), CatchUpPackage::committee()) + .map_err(|err| format!("Failed to get the committee threshold: {err:?}"))?; + + let dkg_id = + active_high_threshold_nidkg_id(pool.as_cache(), summary_block.height()) + .ok_or_else(|| String::from("Couldn't get the high dkg id"))?; + + (threshold, dkg_id, summary_block) + } + CatchUpPackageType::PostSplit { new_subnet_id } => { + let post_split_summary_block = + catchup_package_maker::create_post_split_summary_block( + &summary_block, + new_subnet_id, + self.registry.as_ref(), + ) + .map_err(|err| format!("Failed to create a post-split summary block: {err}"))?; + + let transcript = post_split_summary_block + .payload + .as_ref() + .as_summary() + .dkg + .current_transcript(&NiDkgTag::HighThreshold) + .ok_or_else(|| { + String::from("Couldn't find the transcript in the post-split summary block") + })?; + + let threshold = transcript.threshold.get().get() as usize; + let dkg_id = transcript.dkg_id.clone(); + + (threshold, dkg_id, post_split_summary_block) + } + }; + + let shares = pool + .get_catch_up_package_shares(block.height()) + .collect::>(); + + // The validation logic of CUP shares ensures that all of them have the same content for a + // given height, and it matches the content of the summary block. + if shares.len() < threshold { + return Ok(None); + } + let share_content = shares.first().unwrap().content.clone(); + + let subnet_splitting_status = block + .payload + .as_ref() + .as_summary() + .dkg + .subnet_splitting_status(); + let cup_content = CatchUpContent::from_share_content(share_content, block); + let signatures = shares.iter().map(|share| &share.signature).collect(); + + let cup = self + .crypto + .aggregate(signatures, dkg_id) + .map_err(|err| format!("Failed to aggregate shares: {err}")) + .map(|signature| CatchUpPackage { + content: cup_content, + signature, + })?; + + if let SubnetSplittingStatus::PostSplit(PostSplitArgs { new_subnet_id }) = + subnet_splitting_status + { + info!( + self.log, + "Aggregated a Post-Split CUP for subnet {new_subnet_id} at height {}", + cup.height() + ); + } + + Ok(Some(cup)) + } } fn to_messages(artifacts: Vec) -> Vec { @@ -185,23 +293,33 @@ fn to_messages(artifacts: Vec) -> Vec panic!("Expecting CatchUpPackageShare but got {x:?}\n"), }; + assert!(cup.check_integrity()); assert_eq!(CatchUpShareContent::from(&cup.content), share0.content); cup }) } + + #[rstest] + #[trace] + #[case::no_shares(&[], false)] + #[case::not_enough_shares(&[NODE_1], false)] + #[case::not_enough_shares(&[NODE_1, NODE_2], false)] + #[case::enough_shares(&[NODE_1, NODE_2, NODE_3], true)] + #[case::enough_shares(&[NODE_1, NODE_2, NODE_3, NODE_4], true)] + fn aggregate_post_split_cup_shares_test( + #[case] signers: &[NodeId], + #[case] expected_cup: bool, + ) { + with_test_replica_logger(|log| { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + const SOURCE_SUBNET_ID: SubnetId = SUBNET_1; + const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2; + const INITIAL_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(1); + const SPLITTING_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2); + const INTERVAL_LENGTH: Height = Height::new(9); + let fake_state_hash = CryptoHashOfState::from(CryptoHash(vec![1, 2, 3])); + + let Dependencies { + mut pool, + membership, + registry, + crypto, + state_manager, + replica_config, + .. + } = DependenciesBuilder::new( + pool_config, + vec![ + ( + INITIAL_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4, NODE_5]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + DESTINATION_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_5]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ], + ) + .add_additional_registry_mutation(|registry_data_provider| { + insert_initial_dkg_transcript( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + &SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + registry_data_provider, + ) + }) + .with_replica_config(ReplicaConfig { + node_id: NODE_1, + subnet_id: SOURCE_SUBNET_ID, + }) + .with_mocked_state_manager() + .build(); + + state_manager + .get_mut() + .expect_get_state_hash_at() + .return_const(Ok(fake_state_hash.clone())); + + let message_routing = FakeMessageRouting::new(); + *message_routing.next_batch_height.write().unwrap() = Height::from(2); + let message_routing = Arc::new(message_routing); + + pool.advance_round_normal_operation_n(INTERVAL_LENGTH.get()); + + let subnet_splitting_status = SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: SOURCE_SUBNET_ID, + destination_subnet_id: DESTINATION_SUBNET_ID, + }); + let mut proposal = pool.make_next_block(); + let block = proposal.content.as_mut(); + block.context.certified_height = block.height; + block.context.registry_version = SPLITTING_REGISTRY_VERSION; + let mut payload = block.payload.as_ref().as_summary().clone(); + payload.dkg.subnet_splitting_status = + BackwardsCompatible::new_for_test_only(Some(subnet_splitting_status)); + block.payload = Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(payload), + ); + proposal.content = HashedBlock::new(ic_types::crypto::crypto_hash, block.clone()); + pool.insert_validated(proposal.clone()); + pool.notarize(&proposal); + pool.finalize(&proposal); + + let mut insert_cup_share = |node_id: NodeId| { + let cup_maker = CatchUpPackageMaker::new( + ReplicaConfig { + node_id, + subnet_id: SOURCE_SUBNET_ID, + }, + membership.clone(), + crypto.clone(), + state_manager.clone(), + message_routing.clone(), + registry.clone(), + log.clone(), + ); + + let share = cup_maker + .consider_block(&PoolReader::new(&pool), proposal.content.as_ref().clone()) + .expect("Should succeed with valid inputs"); + pool.insert_validated(share.clone()); + share + }; + + let shares = signers + .iter() + .map(|node_id| insert_cup_share(*node_id)) + .collect::>(); + + let aggregator = ShareAggregator::new( + membership, + message_routing, + crypto, + registry, + replica_config, + log, + ); + + let messages = aggregator.on_state_change(&PoolReader::new(&pool)); + + if expected_cup { + let [ConsensusMessage::CatchUpPackage(cup)] = messages.as_slice() else { + panic!("Should have aggregated a single CUP: {messages:?}"); + }; + + assert!(cup.check_integrity()); + for share in shares { + assert_eq!(CatchUpShareContent::from(&cup.content), share.content); + } + } else { + assert_eq!(messages, vec![], "Shouldn't have aggregated any artifacts"); + } + }) + }) + } } diff --git a/rs/consensus/src/consensus/status.rs b/rs/consensus/src/consensus/status.rs index 08f47ebb0cc3..2c6d7fdef3bd 100644 --- a/rs/consensus/src/consensus/status.rs +++ b/rs/consensus/src/consensus/status.rs @@ -111,6 +111,12 @@ pub(crate) fn should_halt( let should_halt_by_subnet_record = registry_client .get_halt_at_cup_height(subnet_id, registry_version) + .inspect_err(|err| { + warn!( + logger, + "Failed querying the registry at version {registry_version}: {err}" + ) + }) .ok() .flatten() .warn_if_none( diff --git a/rs/consensus/src/consensus/validator.rs b/rs/consensus/src/consensus/validator.rs index 6b61fdda0605..2370a79d7ae2 100644 --- a/rs/consensus/src/consensus/validator.rs +++ b/rs/consensus/src/consensus/validator.rs @@ -2,15 +2,16 @@ //! artifacts. #![allow(clippy::result_large_err)] use crate::consensus::{ - ConsensusMessageId, check_protocol_version, + ConsensusMessageId, + catchup_package_maker::{self, CatchUpPackageType}, + check_protocol_version, metrics::ValidatorMetrics, status::{self, Status}, }; use ic_consensus_dkg as dkg; use ic_consensus_idkg::{self as idkg}; use ic_consensus_utils::{ - MINIMUM_CHAIN_LENGTH, RoundRobin, active_high_threshold_nidkg_id, - active_low_threshold_nidkg_id, + MINIMUM_CHAIN_LENGTH, RoundRobin, active_low_threshold_nidkg_id, crypto::ConsensusCrypto, get_oldest_state_registry_version, membership::{Membership, MembershipError}, @@ -30,6 +31,7 @@ use ic_interfaces_registry::RegistryClient; use ic_interfaces_state_manager::{StateHashError, StateManager}; use ic_logger::{ReplicaLogger, trace, warn}; use ic_metrics::MetricsRegistry; +use ic_registry_client_helpers::node::NodeRegistry; use ic_replicated_state::ReplicatedState; use ic_types::{ Height, NodeId, RegistryVersion, SubnetId, @@ -42,7 +44,10 @@ use ic_types::{ RandomTape, RandomTapeShare, Rank, dkg::{DkgPayloadValidationFailure, InvalidDkgPayloadReason}, }, - crypto::{CryptoError, CryptoHashOf, Signed, threshold_sig::ni_dkg::NiDkgId}, + crypto::{ + CryptoError, CryptoHashOf, Signed, + threshold_sig::ni_dkg::{NiDkgId, NiDkgTag}, + }, registry::RegistryClientError, replica_config::ReplicaConfig, signature::{BasicSigned, MultiSignature, MultiSignatureShare, ThresholdSignatureShare}, @@ -92,6 +97,8 @@ enum ValidationFailure { CatchUpHeightNegligible, MissingPastPayloads, SubnetSplittingStatusError(subnet_splitting::StatusError), + CatchUpPackageTypeError(String), + SubnetSplittingError(String), } /// Possible reasons for invalid artifacts. @@ -99,6 +106,7 @@ enum ValidationFailure { // The fields are only read by the `Debug` implementation. // The `dead_code` lint ignores `Debug` impls, see: https://github.com/rust-lang/rust/issues/88900. #[allow(dead_code)] +#[allow(clippy::large_enum_variant)] enum InvalidArtifactReason { CryptoError(CryptoError), MismatchedRank(Rank, Option), @@ -121,6 +129,8 @@ enum InvalidArtifactReason { RepeatedSigner, ReplicaVersionMismatch, NotABlockmaker, + InvalidHeightInSplittingCatchUpPackageShare, + InvalidSubnetIdInSplittingCatchUpPackage, RegistryVersionNotFrozenDuringSubnetSplitting { context_registry_version: RegistryVersion, subnet_split_scheduled_at: RegistryVersion, @@ -294,12 +304,22 @@ impl SignatureVerify for Signed, + _pool: &PoolReader<'_>, _cfg: &ReplicaConfig, ) -> ValidationResult { let height = self.height(); - let dkg_id = active_high_threshold_nidkg_id(pool.as_cache(), height) - .ok_or_else(|| ValidationFailure::DkgSummaryNotFound(self.height()))?; + let dkg_id = self + .content + .block + .as_ref() + .payload + .as_ref() + .as_summary() + .dkg + .current_transcript(&NiDkgTag::HighThreshold) + .ok_or_else(|| ValidationFailure::DkgSummaryNotFound(self.height()))? + .dkg_id + .clone(); verify_threshold_committee( membership, self.signature.signer, @@ -317,19 +337,42 @@ impl SignatureVerify for CatchUpPackage { membership: &Membership, crypto: &dyn ConsensusCrypto, _pool: &PoolReader<'_>, - _cfg: &ReplicaConfig, + cfg: &ReplicaConfig, ) -> ValidationResult { + let cup_registry_version = self.content.registry_version(); + + let registry_subnet_id = match membership + .registry_client + .get_subnet_id_from_node_id(cfg.node_id, cup_registry_version) + { + Ok(Some(subnet_id)) => subnet_id, + Ok(None) => { + return Err(ValidationError::ValidationFailed( + ValidationFailure::SubnetSplittingError(format!( + "Node {} is not assigned to any subnet at registry version {}", + cfg.node_id, cup_registry_version + )), + )); + } + Err(e) => { + return Err(ValidationError::ValidationFailed( + ValidationFailure::RegistryClientError(e), + )); + } + }; + + if registry_subnet_id != membership.subnet_id { + return Err(ValidationError::InvalidArtifact( + InvalidArtifactReason::InvalidSubnetIdInSplittingCatchUpPackage, + )); + } + crypto .verify_combined_threshold_sig_by_public_key( &self.signature.signature, &self.content, membership.subnet_id, - // Using any registry version here is fine because we assume that the - // public key of the subnet will not change. The alternative of trying - // to use the registry version obtained from the pool is not an option - // here because we may not be able to get a proper value if we do not - // have the relevant portion of the chain. - membership.registry_client.get_latest_version(), + cup_registry_version, ) .map_err(ValidatorError::from) } @@ -1733,10 +1776,60 @@ impl Validator { pool_reader: &PoolReader<'_>, share_content: &CatchUpShareContent, ) -> Result { - let height = share_content.height(); - let block = pool_reader - .get_finalized_block(height) - .ok_or(ValidationFailure::FinalizedBlockNotFound(height))?; + let share_height = share_content.height(); + + let dkg_summary_block = pool_reader.get_highest_finalized_summary_block(); + let dkg_summary = &dkg_summary_block.payload.as_ref().as_summary().dkg; + + let (block, beacon, state_height) = match catchup_package_maker::get_catch_up_package_type( + self.registry_client.as_ref(), + self.replica_config.node_id, + &dkg_summary_block, + ) + .map_err(|err| { + ValidationFailure::CatchUpPackageTypeError(format!( + "Failed to determine the cup type: {err}" + )) + })? { + CatchUpPackageType::PostSplit { new_subnet_id } + if dkg_summary.get_next_start_height() == share_height => + { + let post_split_block = catchup_package_maker::create_post_split_summary_block( + &dkg_summary_block, + new_subnet_id, + self.registry_client.as_ref(), + ) + .map_err(ValidationFailure::SubnetSplittingError)?; + + let post_split_random_beacon = + catchup_package_maker::create_post_split_random_beacon(&post_split_block) + .map_err(ValidationFailure::SubnetSplittingError)?; + + let state_height = dkg_summary.height; + + (post_split_block, post_split_random_beacon, state_height) + } + // We don't produce CUPs for the height at which a subnet splitting is happening. + CatchUpPackageType::PostSplit { .. } if dkg_summary.height == share_height => { + return Err( + InvalidArtifactReason::InvalidHeightInSplittingCatchUpPackageShare.into(), + ); + } + CatchUpPackageType::PostSplit { .. } | CatchUpPackageType::Normal => { + let block = pool_reader + .get_finalized_block(share_height) + .ok_or(ValidationFailure::FinalizedBlockNotFound(share_height))?; + + let beacon = pool_reader + .get_random_beacon(share_height) + .ok_or(ValidationFailure::RandomBeaconNotFound(share_height))?; + + let state_height = share_height; + + (block, beacon, state_height) + } + }; + if ic_types::crypto::crypto_hash(&block) != share_content.block { return Err(InvalidArtifactReason::MismatchedBlockInCatchUpPackageShare.into()); } @@ -1745,25 +1838,22 @@ impl Validator { return Err(InvalidArtifactReason::DataPayloadBlockInCatchUpPackageShare.into()); } - let beacon = pool_reader - .get_random_beacon(height) - .ok_or(ValidationFailure::RandomBeaconNotFound(height))?; if &beacon != share_content.random_beacon.get_value() { return Err(InvalidArtifactReason::MismatchedRandomBeaconInCatchUpPackageShare.into()); } - let hash = self + let state_hash = self .state_manager - .get_state_hash_at(height) + .get_state_hash_at(state_height) .map_err(ValidationFailure::StateHashError)?; - if hash != share_content.state_hash { + if state_hash != share_content.state_hash { return Err(InvalidArtifactReason::MismatchedStateHashInCatchUpPackageShare.into()); } // Should succeed as we already got the hash above let state = self .state_manager - .get_state_at(height) + .get_state_at(state_height) .map_err(ValidationFailure::StateManagerError)?; let registry_version = get_oldest_state_registry_version(state.get_ref()); if registry_version != share_content.oldest_registry_version_in_use_by_replicated_state { @@ -4931,4 +5021,227 @@ pub mod test { }) }); } + + mod subnet_splitting { + use super::*; + + use crate::consensus::catchup_package_maker::CatchUpPackageMaker; + use ic_consensus_mocks::DependenciesBuilder; + use ic_test_utilities::message_routing::FakeMessageRouting; + use ic_test_utilities_logger::with_test_replica_logger; + use ic_test_utilities_registry::insert_initial_dkg_transcript; + use ic_types::backwards_compatibility::BackwardsCompatible; + use ic_types::consensus::dkg::{SplittingArgs, SubnetSplittingStatus}; + use ic_types_test_utils::ids::{NODE_3, NODE_4, SUBNET_1, SUBNET_2}; + + use std::str::FromStr; + + enum MalformShare { + StateHash, + RandomBeacon, + RegistryVersion, + Height, + } + #[rstest] + #[case(NODE_1, None, Ok(()))] + #[case(NODE_2, None, Ok(()))] + // after the split, nodes NODE_3 and NODE_4 will be on a different subnet than the validator + // (NODE_1) + #[case::wrong_subnet(NODE_3, None, Err("MismatchedBlockInCatchUpPackageShare"))] + #[case::wrong_subnet(NODE_4, None, Err("MismatchedBlockInCatchUpPackageShare"))] + #[case::wrong_state_hash( + NODE_1, + Some(MalformShare::StateHash), + Err("MismatchedStateHashInCatchUpPackageShare") + )] + #[case::wrong_random_beacon( + NODE_1, + Some(MalformShare::RandomBeacon), + Err("MismatchedRandomBeaconInCatchUpPackageShare") + )] + #[case::wrong_registry_version( + NODE_1, + Some(MalformShare::RegistryVersion), + Err("MismatchedOldestRegistryVersionInCatchUpPackageShare") + )] + #[case::wrong_height( + NODE_1, + Some(MalformShare::Height), + Err("InvalidHeightInSplittingCatchUpPackageShare") + )] + fn validate_post_split_cup_share_test( + #[case] cup_share_node_id: NodeId, + #[case] malform_share: Option, + #[case] expected_validation_result: Result<(), &str>, + ) { + with_test_replica_logger(|log| { + ic_test_utilities::artifact_pool_config::with_test_pool_config(|pool_config| { + const SOURCE_SUBNET_ID: SubnetId = SUBNET_1; + const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2; + const INITIAL_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(1); + const SPLITTING_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2); + const INTERVAL_LENGTH: Height = Height::new(9); + let fake_state_hash = CryptoHashOfState::from(CryptoHash(vec![1, 2, 3])); + + let ValidatorAndDependencies { + mut pool, + membership, + registry_client: registry, + crypto, + validator, + state_manager, + .. + } = ValidatorAndDependencies::new( + DependenciesBuilder::new( + pool_config, + vec![ + ( + INITIAL_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2, NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_1, NODE_2]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ( + SPLITTING_REGISTRY_VERSION.get(), + DESTINATION_SUBNET_ID, + SubnetRecordBuilder::from(&[NODE_3, NODE_4]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + ), + ], + ) + .add_additional_registry_mutation(|registry_data_provider| { + insert_initial_dkg_transcript( + SPLITTING_REGISTRY_VERSION.get(), + SOURCE_SUBNET_ID, + &SubnetRecordBuilder::from(&[NODE_1, NODE_2]) + .with_dkg_interval_length(INTERVAL_LENGTH.get()) + .build(), + registry_data_provider, + ) + }) + .with_replica_config(ReplicaConfig { + node_id: NODE_1, + subnet_id: SOURCE_SUBNET_ID, + }) + .with_mocked_state_manager() + .build(), + ); + + state_manager + .get_mut() + .expect_get_state_hash_at() + .return_const(Ok(fake_state_hash.clone())); + + let message_routing = FakeMessageRouting::new(); + *message_routing.next_batch_height.write().unwrap() = Height::from(2); + let message_routing = Arc::new(message_routing); + + let cup_maker = CatchUpPackageMaker::new( + ReplicaConfig { + node_id: cup_share_node_id, + subnet_id: SOURCE_SUBNET_ID, + }, + membership, + crypto, + state_manager, + message_routing, + registry, + log, + ); + + pool.advance_round_normal_operation_n(INTERVAL_LENGTH.get()); + + let subnet_splitting_status = SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: SOURCE_SUBNET_ID, + destination_subnet_id: DESTINATION_SUBNET_ID, + }); + let mut proposal = pool.make_next_block(); + let block = proposal.content.as_mut(); + block.context.certified_height = block.height; + block.context.registry_version = SPLITTING_REGISTRY_VERSION; + let mut payload = block.payload.as_ref().as_summary().clone(); + payload.dkg.subnet_splitting_status = + BackwardsCompatible::new_for_test_only(Some(subnet_splitting_status)); + block.payload = Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(payload), + ); + proposal.content = + HashedBlock::new(ic_types::crypto::crypto_hash, block.clone()); + pool.insert_validated(proposal.clone()); + pool.notarize(&proposal); + pool.finalize(&proposal); + + let mut share = cup_maker + .consider_block(&PoolReader::new(&pool), proposal.content.as_ref().clone()) + .expect("Should succeed with valid inputs"); + + match malform_share { + Some(MalformShare::StateHash) => { + share.content.state_hash = + CryptoHashOfState::from(CryptoHash(vec![3, 1, 4])); + } + Some(MalformShare::RandomBeacon) => { + let mut invalid_beacon = share.content.random_beacon.into_inner(); + invalid_beacon.content.version = + ReplicaVersion::from_str("invalid_replica_version").unwrap(); + + share.content.random_beacon = HashedRandomBeacon::new( + ic_types::crypto::crypto_hash, + invalid_beacon, + ); + } + Some(MalformShare::RegistryVersion) => { + share + .content + .oldest_registry_version_in_use_by_replicated_state = + Some(INITIAL_REGISTRY_VERSION); + } + Some(MalformShare::Height) => { + let mut beacon = share.content.random_beacon.into_inner(); + beacon.content.height = proposal.height(); + + share.content.random_beacon = + HashedRandomBeacon::new(ic_types::crypto::crypto_hash, beacon); + } + None => {} + } + + pool.insert_unvalidated(share.clone()); + + let pool_reader = PoolReader::new(&pool); + let change_set = validator.validate_catch_up_package_shares(&pool_reader); + + match expected_validation_result { + Ok(()) => { + assert_eq!( + change_set, + vec![ChangeAction::MoveToValidated( + ConsensusMessage::CatchUpPackageShare(share) + )] + ); + } + Err(err) => { + assert_eq!( + change_set, + vec![ChangeAction::HandleInvalid( + ConsensusMessage::CatchUpPackageShare(share), + String::from(err), + )] + ); + } + } + }) + }) + } + } } diff --git a/rs/consensus/tests/framework/runner.rs b/rs/consensus/tests/framework/runner.rs index 759ed2c3f9a8..17cc6aca7a7d 100644 --- a/rs/consensus/tests/framework/runner.rs +++ b/rs/consensus/tests/framework/runner.rs @@ -140,6 +140,8 @@ impl<'a> ConsensusRunner<'a> { consensus_crypto.clone(), replica_logger.clone(), pool_reader, + deps.registry_client.clone(), + deps.replica_config.clone(), ))); let malicious_flags = MaliciousFlags::default(); let consensus = ic_consensus::consensus::ConsensusImpl::new( diff --git a/rs/consensus/tests/payload.rs b/rs/consensus/tests/payload.rs index cea5a9749104..995d0be0a5ee 100644 --- a/rs/consensus/tests/payload.rs +++ b/rs/consensus/tests/payload.rs @@ -159,6 +159,8 @@ fn consensus_produces_expected_batches() { Arc::clone(&fake_crypto) as Arc<_>, no_op_logger(), &PoolReader::new(&*consensus_pool.read().unwrap()), + registry_client.clone(), + replica_config.clone(), ))); let consensus = ic_consensus::consensus::ConsensusImpl::new( diff --git a/rs/consensus/utils/BUILD.bazel b/rs/consensus/utils/BUILD.bazel index fa7437d08ae1..63ec9e8f929b 100644 --- a/rs/consensus/utils/BUILD.bazel +++ b/rs/consensus/utils/BUILD.bazel @@ -68,5 +68,6 @@ rust_test( "@crate_index//:rayon", "@crate_index//:rstest", "@crate_index//:slog", + "@crate_index//:thiserror", ], ) diff --git a/rs/consensus/utils/src/lib.rs b/rs/consensus/utils/src/lib.rs index eae36d7cae12..c927405c0142 100644 --- a/rs/consensus/utils/src/lib.rs +++ b/rs/consensus/utils/src/lib.rs @@ -12,7 +12,10 @@ use ic_registry_client_helpers::subnet::{NotarizationDelaySettings, SubnetRegist use ic_replicated_state::ReplicatedState; use ic_types::{ Height, NodeId, RegistryVersion, ReplicaVersion, SubnetId, - consensus::{Block, BlockProposal, HasCommittee, HasHeight, HasRank, Threshold}, + consensus::{ + Block, BlockProposal, HasCommittee, HasHeight, HasRank, Threshold, + dkg::SubnetSplittingStatus, + }, crypto::{ Signed, threshold_sig::ni_dkg::{NiDkgId, NiDkgReceivers, NiDkgTag, NiDkgTranscript}, @@ -326,6 +329,14 @@ pub fn active_high_threshold_committee( }) } +/// Return the current high transcript for the given height if it was found. +pub fn subnet_splitting_status_at_height( + reader: &dyn ConsensusPoolCache, + height: Height, +) -> Option { + get_active_data_at(reader, height, get_subnet_splitting_status_at_given_summary) +} + /// Return the active DKGData active at the given height if it was found. fn get_active_data_at( reader: &dyn ConsensusPoolCache, @@ -355,6 +366,19 @@ fn get_active_data_at( .or_else(|| getter(&reader.summary_block(), height)) } +fn get_subnet_splitting_status_at_given_summary( + summary_block: &Block, + height: Height, +) -> Option { + let dkg_summary = &summary_block.payload.as_ref().as_summary().dkg; + + if dkg_summary.current_interval_includes(height) { + Some(dkg_summary.subnet_splitting_status()) + } else { + None + } +} + fn get_registry_version_at_given_summary( summary_block: &Block, height: Height, diff --git a/rs/consensus/utils/src/pool_reader.rs b/rs/consensus/utils/src/pool_reader.rs index cc53af781a50..82c2964ba953 100644 --- a/rs/consensus/utils/src/pool_reader.rs +++ b/rs/consensus/utils/src/pool_reader.rs @@ -558,6 +558,15 @@ impl<'a> PoolReader<'a> { .dkg .get_next_start_height() } + + pub fn get_next_summary_height(&self) -> Height { + self.get_highest_finalized_summary_block() + .payload + .as_ref() + .as_summary() + .dkg + .get_next_start_height() + } } /// Take a slice returned by [`PoolReader::get_payloads_from_height`] diff --git a/rs/consensus/utils/src/subnet_splitting.rs b/rs/consensus/utils/src/subnet_splitting.rs index 17c85e92bcd8..d0a9adf9c7e7 100644 --- a/rs/consensus/utils/src/subnet_splitting.rs +++ b/rs/consensus/utils/src/subnet_splitting.rs @@ -2,9 +2,14 @@ use ic_interfaces_registry::RegistryClient; use ic_protobuf::{ proxy::ProxyDecodeError, registry::subnet::v1::catch_up_package_contents::CupType, }; -use ic_registry_client_helpers::subnet::SubnetRegistry; +use ic_registry_client_helpers::{node::NodeRegistry, subnet::SubnetRegistry}; use ic_types::{ - RegistryVersion, SubnetId, consensus::SubnetSplittingArgs, registry::RegistryClientError, + NodeId, RegistryVersion, SubnetId, + consensus::{ + Block, SubnetSplittingArgs, + dkg::{SplittingArgs, SubnetSplittingStatus}, + }, + registry::RegistryClientError, }; use thiserror::Error; @@ -84,45 +89,107 @@ pub fn get_status( }) } +pub fn is_split_scheduled(summary_block: &Block) -> Option { + match summary_block + .payload + .as_ref() + .as_summary() + .dkg + .subnet_splitting_status() + { + SubnetSplittingStatus::Scheduled(splitting_args) => Some(splitting_args), + SubnetSplittingStatus::NotScheduled | SubnetSplittingStatus::PostSplit(..) => None, + } +} + +#[derive(Debug)] +pub struct PostSplitAssignment { + pub new_subnet_id: SubnetId, + pub other_subnet_id: SubnetId, +} + +#[derive(Debug, Error)] +pub enum PostSplitAssignmentError { + #[error("Error while getting the subnet id from the registry at version {0}: {1}")] + FailedToGetSubnetIdFromTheRegistry(RegistryVersion, RegistryClientError), + #[error("The node is unassigned at registry version {0}")] + Unassigned(RegistryVersion), + #[error("The node changed subnets during subnet splitting")] + DisallowedMembershipChange(SubnetId), +} + +pub fn get_post_split_subnet_assignment( + node_id: NodeId, + summary_block: &Block, + registry_client: &dyn RegistryClient, + SplittingArgs { + destination_subnet_id, + source_subnet_id, + }: SplittingArgs, +) -> Result { + let new_subnet_id = registry_client + .get_subnet_id_from_node_id(node_id, summary_block.context.registry_version) + .map_err(|err| { + PostSplitAssignmentError::FailedToGetSubnetIdFromTheRegistry( + summary_block.context.registry_version, + err, + ) + })? + .ok_or(PostSplitAssignmentError::Unassigned( + summary_block.context.registry_version, + ))?; + + let other_subnet_id = if new_subnet_id == destination_subnet_id { + source_subnet_id + } else if new_subnet_id == source_subnet_id { + destination_subnet_id + } else { + return Err(PostSplitAssignmentError::DisallowedMembershipChange( + new_subnet_id, + )); + }; + + Ok(PostSplitAssignment { + new_subnet_id, + other_subnet_id, + }) +} + #[cfg(test)] mod tests { + use assert_matches::assert_matches; + use std::sync::Arc; + + use ic_interfaces_registry::RegistryClientVersionedResult; use ic_protobuf::registry::subnet::v1::CatchUpPackageContents; use ic_protobuf::registry::subnet::v1::{GenesisArgs, RecoveryArgs}; use ic_registry_keys::make_catch_up_package_contents_key; - use ic_test_utilities_registry::{SubnetRecordBuilder, setup_registry_non_final}; - use ic_test_utilities_types::ids::{NODE_1, SUBNET_1, SUBNET_2}; + use ic_test_utilities_consensus::fake::Fake; + use ic_test_utilities_registry::{ + SubnetRecordBuilder, add_single_subnet_record, add_subnet_list_record, + setup_registry_non_final, + }; + use ic_test_utilities_types::ids::{ + NODE_1, NODE_2, NODE_3, NODE_4, SUBNET_1, SUBNET_2, SUBNET_3, + }; use ic_types::subnet_id_into_protobuf; + use ic_types::{ + Height, ReplicaVersion, Time, + backwards_compatibility::BackwardsCompatible, + batch::ValidationContext, + consensus::{BlockPayload, Payload, Rank, SummaryPayload, dkg::PostSplitArgs}, + crypto::{CryptoHash, CryptoHashOf}, + time::UNIX_EPOCH, + }; use rstest::rstest; - use std::sync::Arc; const SOURCE_SUBNET_ID: SubnetId = SUBNET_1; const DESTINATION_SUBNET_ID: SubnetId = SUBNET_2; + const OTHER_SUBNET_ID: SubnetId = SUBNET_3; const REGISTRY_CUP_REGISTRY_VERSION: RegistryVersion = RegistryVersion::new(2); use super::*; - fn set_up_registry(cup_type: Option) -> Arc { - let (registry_data_provider, registry) = setup_registry_non_final( - SOURCE_SUBNET_ID, - (1..=REGISTRY_CUP_REGISTRY_VERSION.increment().get()) - .map(|version| (version, SubnetRecordBuilder::from(&[NODE_1]).build())) - .collect(), - ); - registry_data_provider - .add( - &make_catch_up_package_contents_key(SOURCE_SUBNET_ID), - REGISTRY_CUP_REGISTRY_VERSION, - Some(CatchUpPackageContents { - cup_type, - ..Default::default() - }), - ) - .unwrap(); - registry.update_to_latest_version(); - - registry - } - #[rstest] fn get_status_should_return_not_scheduled_when_latest_cup_is_not_subnet_splitting_test( #[values( @@ -249,4 +316,198 @@ mod tests { assert_eq!(status, Status::NotScheduled); } + + fn set_up_registry(cup_type: Option) -> Arc { + let (registry_data_provider, registry) = setup_registry_non_final( + SOURCE_SUBNET_ID, + (1..=REGISTRY_CUP_REGISTRY_VERSION.increment().get()) + .map(|version| (version, SubnetRecordBuilder::from(&[NODE_1]).build())) + .collect(), + ); + registry_data_provider + .add( + &make_catch_up_package_contents_key(SOURCE_SUBNET_ID), + REGISTRY_CUP_REGISTRY_VERSION, + Some(CatchUpPackageContents { + cup_type, + ..Default::default() + }), + ) + .unwrap(); + registry.update_to_latest_version(); + + registry + } + + fn make_summary_block_with_status(subnet_splitting_status: SubnetSplittingStatus) -> Block { + let mut summary = SummaryPayload::fake(); + summary.dkg.subnet_splitting_status = + BackwardsCompatible::new_for_test_only(Some(subnet_splitting_status)); + Block { + version: ReplicaVersion::default(), + parent: CryptoHashOf::from(CryptoHash(vec![])), + payload: Payload::new( + ic_types::crypto::crypto_hash, + BlockPayload::Summary(summary), + ), + height: Height::new(0), + rank: Rank(0), + context: ValidationContext { + certified_height: Height::new(0), + registry_version: REGISTRY_CUP_REGISTRY_VERSION, + time: UNIX_EPOCH, + }, + } + } + + fn make_scheduled_summary_block() -> Block { + make_summary_block_with_status(SubnetSplittingStatus::Scheduled(SplittingArgs { + source_subnet_id: SOURCE_SUBNET_ID, + destination_subnet_id: DESTINATION_SUBNET_ID, + })) + } + + fn set_up_post_split_registry( + source_committee: &[NodeId], + destination_committee: &[NodeId], + other_committee: &[NodeId], + ) -> Arc { + let (registry_data_provider, registry) = setup_registry_non_final( + SOURCE_SUBNET_ID, + vec![( + 1, + SubnetRecordBuilder::new() + .with_committee(source_committee) + .build(), + )], + ); + add_single_subnet_record( + ®istry_data_provider, + REGISTRY_CUP_REGISTRY_VERSION.get(), + DESTINATION_SUBNET_ID, + SubnetRecordBuilder::new() + .with_committee(destination_committee) + .build(), + ); + add_single_subnet_record( + ®istry_data_provider, + REGISTRY_CUP_REGISTRY_VERSION.get(), + OTHER_SUBNET_ID, + SubnetRecordBuilder::new() + .with_committee(other_committee) + .build(), + ); + add_subnet_list_record( + ®istry_data_provider, + REGISTRY_CUP_REGISTRY_VERSION.get(), + vec![SOURCE_SUBNET_ID, DESTINATION_SUBNET_ID, OTHER_SUBNET_ID], + ); + registry.update_to_latest_version(); + registry + } + + struct ErrorRegistryClient; + + impl RegistryClient for ErrorRegistryClient { + fn get_versioned_value( + &self, + _key: &str, + version: RegistryVersion, + ) -> RegistryClientVersionedResult> { + Err(RegistryClientError::VersionNotAvailable { version }) + } + + fn get_key_family( + &self, + _key_prefix: &str, + version: RegistryVersion, + ) -> Result, RegistryClientError> { + Err(RegistryClientError::VersionNotAvailable { version }) + } + + fn get_latest_version(&self) -> RegistryVersion { + RegistryVersion::from(0) + } + + fn get_version_timestamp(&self, _registry_version: RegistryVersion) -> Option