Merge branch 'kpop/CON-979/unhalt_on_cup_2' into 'master'

feat(consensus): [CON-979] set `halt_at_cup_height` to `false` during CUP creation.

In order to unhalt a Subnet which was halted due to setting the `halt_at_cup_height` flag to `true`, currently we have to do it in two steps:
1) create a proposal which sets `halt_at_cup_height` to `false`; and then
2) create a new cup proposal.

After this step we will still require two steps, namely:
1) create a new cup proposal, and then
2) create a proposal which sets `is_halted` to false,

but they will fit nicer in the process of subnet splitting where, on high level, we want to do the following:
1) halt a subnet at cup height
2) split state
3) create a new cup with the new state
4) do some extra validation
5) unhalt the subnet; 

See merge request dfinity-lab/public/ic!13001

rs/protobuf/def/registry/subnet/v1/subnet.proto

@@ -93,6 +93,10 @@ message SubnetRecord {
 
   // If `true`, the subnet will be halted after reaching the next cup height: it will no longer
   // create or execute blocks.
+  //
+  // Note: this flag is reset automatically when a new CUP proposal is approved. When that
+  // happens, the `is_halted` flag is set to `true`, so the Subnet remains halted until an
+  // appropriate proposal which sets `is_halted` to `false` is approved.
   bool halt_at_cup_height = 28;
 }
 

rs/protobuf/src/gen/registry/registry.subnet.v1.rs

@@ -79,6 +79,10 @@ pub struct SubnetRecord {
     pub ecdsa_config: ::core::option::Option<EcdsaConfig>,
     /// If `true`, the subnet will be halted after reaching the next cup height: it will no longer
     /// create or execute blocks.
+    ///
+    /// Note: this flag is reset automatically when a new CUP proposal is approved. When that
+    /// happens, the `is_halted` flag is set to `true`, so the Subnet remains halted until an
+    /// appropriate proposal which sets `is_halted` to `false` is approved.
     #[prost(bool, tag = "28")]
     pub halt_at_cup_height: bool,
 }

rs/protobuf/src/gen/state/registry.subnet.v1.rs

@@ -78,6 +78,10 @@ pub struct SubnetRecord {
     pub ecdsa_config: ::core::option::Option<EcdsaConfig>,
     /// If `true`, the subnet will be halted after reaching the next cup height: it will no longer
     /// create or execute blocks.
+    ///
+    /// Note: this flag is reset automatically when a new CUP proposal is approved. When that
+    /// happens, the `is_halted` flag is set to `true`, so the Subnet remains halted until an
+    /// appropriate proposal which sets `is_halted` to `false` is approved.
     #[prost(bool, tag = "28")]
     pub halt_at_cup_height: bool,
 }

rs/protobuf/src/gen/types/registry.subnet.v1.rs

@@ -79,6 +79,10 @@ pub struct SubnetRecord {
     pub ecdsa_config: ::core::option::Option<EcdsaConfig>,
     /// If `true`, the subnet will be halted after reaching the next cup height: it will no longer
     /// create or execute blocks.
+    ///
+    /// Note: this flag is reset automatically when a new CUP proposal is approved. When that
+    /// happens, the `is_halted` flag is set to `true`, so the Subnet remains halted until an
+    /// appropriate proposal which sets `is_halted` to `false` is approved.
     #[prost(bool, tag = "28")]
     pub halt_at_cup_height: bool,
 }

rs/registry/canister/src/mutations/do_recover_subnet.rs

@@ -5,9 +5,11 @@
 //! bad state) and optionally replacing any (potentially) broken nodes in the
 //! subnet with a set of known-good nodes
 
-use crate::mutations::do_create_subnet::EcdsaInitialConfig;
-use crate::registry::Version;
-use crate::{common::LOG_PREFIX, mutations::common::encode_or_panic, registry::Registry};
+use crate::{
+    common::LOG_PREFIX,
+    mutations::{common::encode_or_panic, do_create_subnet::EcdsaInitialConfig},
+    registry::{Registry, Version},
+};
 use candid::{CandidType, Deserialize, Encode};
 use dfn_core::api::{call, CanisterId};
 #[cfg(target_arch = "wasm32")]
@@ -19,8 +21,10 @@ use ic_registry_keys::{
     make_catch_up_package_contents_key, make_crypto_threshold_signing_pubkey_key,
     make_subnet_record_key,
 };
-use ic_registry_transport::pb::v1::{registry_mutation, RegistryMutation};
-use ic_registry_transport::upsert;
+use ic_registry_transport::{
+    pb::v1::{registry_mutation, RegistryMutation},
+    upsert,
+};
 use on_wire::bytes;
 use serde::Serialize;
 use std::convert::TryFrom;
@@ -57,6 +61,17 @@ impl Registry {
 
             let mut subnet_record = self.get_subnet_or_panic(subnet_id);
 
+            // If [SubnetRecord::halt_at_cup_height] is set to `true`, then the Subnet was
+            // instructed to halt after reaching the next CUP height. Since the Consensus is
+            // looking at the registry version from the highest CUP when considering this flag,
+            // we reset it to `false` but flip the `is_halted` flag to `true`, so the subnet
+            // remains halted but can be later unhalted by sending an appropriate proposal which
+            // resets `is_halted` to `false`.
+            if subnet_record.halt_at_cup_height {
+                subnet_record.halt_at_cup_height = false;
+                subnet_record.is_halted = true;
+            }
+
             let dkg_nodes = if let Some(replacement_nodes) = payload.replacement_nodes.clone() {
                 self.replace_subnet_record_membership(
                     subnet_id,
@@ -95,7 +110,6 @@ impl Registry {
             // and make sure the subnet is not listed as signing_subnet for keys it no longer holds
             if let Some(ref new_ecdsa_config) = payload.ecdsa_config {
                 // get current set of keys
-
                 let new_key_list: Vec<EcdsaKeyId> = new_ecdsa_config
                     .keys
                     .iter()
@@ -106,9 +120,11 @@ impl Registry {
                     subnet_id,
                     &self.get_keys_that_will_be_removed_from_subnet(subnet_id, new_key_list),
                 ));
+
                 // Update ECDSA configuration on subnet record to reflect new holdings
                 subnet_record.ecdsa_config = Some(new_ecdsa_config.clone().into());
             }
+
             // Push all of our subnet_record mutations
             mutations.push(upsert(
                 make_subnet_record_key(subnet_id),
@@ -260,16 +276,17 @@ fn get_record_version_as_of_registry_version(
 
 #[cfg(test)]
 mod test {
-    use crate::common::test_helpers::{
-        add_fake_subnet, get_invariant_compliant_subnet_record, invariant_compliant_registry,
-        prepare_registry_with_nodes,
-    };
-    use crate::mutations::do_create_subnet::EcdsaInitialConfig;
-    use crate::mutations::do_create_subnet::EcdsaKeyRequest;
-    use crate::mutations::do_recover_subnet::{
-        panic_if_record_changed_across_versions, RecoverSubnetPayload,
+    use crate::{
+        common::test_helpers::{
+            add_fake_subnet, get_invariant_compliant_subnet_record, invariant_compliant_registry,
+            prepare_registry_with_nodes,
+        },
+        mutations::{
+            do_create_subnet::{EcdsaInitialConfig, EcdsaKeyRequest},
+            do_recover_subnet::{panic_if_record_changed_across_versions, RecoverSubnetPayload},
+        },
+        registry::Registry,
     };
-    use crate::registry::Registry;
     use ic_base_types::SubnetId;
     use ic_ic00_types::{EcdsaCurve, EcdsaKeyId};
     use ic_protobuf::registry::subnet::v1::SubnetRecord;

rs/registry/canister/tests/recover_subnet.rs

@@ -5,34 +5,33 @@ use ic_config::Config;
 use ic_ic00_types::{EcdsaCurve, EcdsaKeyId};
 use ic_interfaces_registry::RegistryClient;
 use ic_nns_common::registry::encode_or_panic;
-use ic_nns_test_utils::itest_helpers::try_call_via_universal_canister;
 use ic_nns_test_utils::{
     itest_helpers::{
         forward_call_via_universal_canister, local_test_on_nns_subnet_with_mutations,
-        set_up_registry_canister, set_up_universal_canister,
+        set_up_registry_canister, set_up_universal_canister, try_call_via_universal_canister,
     },
     registry::{get_value_or_panic, prepare_registry},
 };
-use ic_protobuf::registry::crypto::v1::EcdsaSigningSubnetList;
-use ic_protobuf::registry::crypto::v1::{EcdsaCurve as pbEcdsaCurve, EcdsaKeyId as pbEcdsaKeyId};
-use ic_protobuf::registry::subnet::v1::{CatchUpPackageContents, EcdsaConfig};
-use ic_protobuf::registry::subnet::v1::{SubnetListRecord, SubnetRecord};
+use ic_protobuf::registry::{
+    crypto::v1::{EcdsaCurve as pbEcdsaCurve, EcdsaKeyId as pbEcdsaKeyId, EcdsaSigningSubnetList},
+    subnet::v1::{CatchUpPackageContents, EcdsaConfig, SubnetListRecord, SubnetRecord},
+};
 use ic_registry_keys::{
     make_catch_up_package_contents_key, make_crypto_threshold_signing_pubkey_key,
     make_ecdsa_signing_subnet_list_key, make_subnet_list_record_key, make_subnet_record_key,
 };
 use ic_registry_subnet_features::DEFAULT_ECDSA_MAX_QUEUE_SIZE;
-use ic_registry_transport::pb::v1::RegistryAtomicMutateRequest;
-use ic_registry_transport::{insert, upsert};
+use ic_registry_transport::{insert, pb::v1::RegistryAtomicMutateRequest, upsert};
 use ic_replica_tests::{canister_test_with_config_async, get_ic_config};
 use ic_test_utilities::types::ids::subnet_test_id;
 use ic_types::ReplicaVersion;
-use registry_canister::mutations::common::decode_registry_value;
-use registry_canister::mutations::do_create_subnet::{
-    CreateSubnetPayload, EcdsaInitialConfig, EcdsaKeyRequest,
-};
 use registry_canister::{
-    init::RegistryCanisterInitPayloadBuilder, mutations::do_recover_subnet::RecoverSubnetPayload,
+    init::RegistryCanisterInitPayloadBuilder,
+    mutations::{
+        common::decode_registry_value,
+        do_create_subnet::{CreateSubnetPayload, EcdsaInitialConfig, EcdsaKeyRequest},
+        do_recover_subnet::RecoverSubnetPayload,
+    },
 };
 use std::convert::TryFrom;
 
@@ -587,6 +586,116 @@ fn test_recover_subnet_without_ecdsa_key_removes_it_from_signing_list() {
     });
 }
 
+#[test]
+fn test_recover_subnet_resets_the_halt_at_cup_height_flag() {
+    let ic_config = get_ic_config();
+    let (config, _tmpdir) = Config::temp_config();
+    canister_test_with_config_async(config, ic_config, |local_runtime| async move {
+        let data_provider = local_runtime.registry_data_provider.clone();
+        let fake_client = local_runtime.registry_client.clone();
+
+        let runtime = Runtime::Local(local_runtime);
+        // get some nodes for our tests
+        let (init_mutate, mut node_ids) = prepare_registry_with_nodes(5, 0);
+
+        let subnet_to_recover_nodes = vec![node_ids.pop().unwrap()];
+        let mut subnet_to_recover: SubnetRecord = CreateSubnetPayload {
+            unit_delay_millis: 10,
+            gossip_retransmission_request_ms: 10_000,
+            gossip_registry_poll_period_ms: 2000,
+            gossip_pfn_evaluation_period_ms: 50,
+            gossip_receive_check_cache_size: 1,
+            gossip_max_duplicity: 1,
+            gossip_max_chunk_wait_ms: 200,
+            gossip_max_artifact_streams_per_peer: 1,
+            replica_version_id: ReplicaVersion::default().into(),
+            node_ids: subnet_to_recover_nodes.clone(),
+            ..Default::default()
+        }
+        .into();
+
+        // Set the `halt_at_cup_height` to `true`, to verify that it will be later set to `false`.
+        subnet_to_recover.halt_at_cup_height = true;
+        subnet_to_recover.is_halted = false;
+
+        let mut subnet_list_record = decode_registry_value::<SubnetListRecord>(
+            fake_client
+                .get_value(
+                    &make_subnet_list_record_key(),
+                    fake_client.get_latest_version(),
+                )
+                .unwrap()
+                .unwrap(),
+        );
+
+        let subnet_to_recover_subnet_id = subnet_test_id(1003);
+
+        subnet_list_record
+            .subnets
+            .push(subnet_to_recover_subnet_id.get().into_vec());
+
+        let mutations = vec![
+            upsert(
+                make_subnet_record_key(subnet_to_recover_subnet_id).into_bytes(),
+                encode_or_panic(&subnet_to_recover),
+            ),
+            upsert(
+                make_subnet_list_record_key().into_bytes(),
+                encode_or_panic(&subnet_list_record),
+            ),
+            insert(
+                make_crypto_threshold_signing_pubkey_key(subnet_to_recover_subnet_id).as_bytes(),
+                encode_or_panic(&vec![]),
+            ),
+            insert(
+                make_catch_up_package_contents_key(subnet_to_recover_subnet_id).as_bytes(),
+                encode_or_panic(&dummy_cup_for_subnet(subnet_to_recover_nodes)),
+            ),
+        ];
+
+        let add_subnets_mutate = RegistryAtomicMutateRequest {
+            preconditions: vec![],
+            mutations,
+        };
+
+        let registry = setup_registry_synced_with_fake_client(
+            &runtime,
+            fake_client,
+            data_provider,
+            vec![init_mutate, add_subnets_mutate],
+        )
+        .await;
+
+        // Install the universal canister in place of the governance canister
+        let fake_governance_canister = set_up_universal_canister_as_governance(&runtime).await;
+
+        let payload = RecoverSubnetPayload {
+            subnet_id: subnet_to_recover_subnet_id.get(),
+            height: 10,
+            time_ns: 1200,
+            state_hash: vec![10, 20, 30],
+            replacement_nodes: None,
+            registry_store_uri: None,
+            ecdsa_config: None,
+        };
+
+        try_call_via_universal_canister(
+            &fake_governance_canister,
+            &registry,
+            "recover_subnet",
+            Encode!(&payload).unwrap(),
+        )
+        .await
+        .unwrap();
+
+        // Verify that the `halt_at_cup_height` and `is_halted` flags are correctly set.
+        let subnet_record = get_subnet_record(&registry, subnet_to_recover_subnet_id).await;
+
+        assert!(!subnet_record.halt_at_cup_height);
+        assert!(subnet_record.is_halted);
+    });
+}
+
 pub async fn ecdsa_signing_subnet_list(
     registry: &Canister<'_>,
     key_id: &EcdsaKeyId,