Draft: create configs early

rs/consensus/dkg/src/payload_validator.rs

@@ -1,5 +1,5 @@
-use self::payload_builder::create_early_remote_transcripts;
 use super::{crypto_validate_dealing, payload_builder, utils};
+use crate::remote::{build_callback_id_config_map, merge_configs};
 use ic_consensus_utils::{crypto::ConsensusCrypto, pool_reader::PoolReader};
 use ic_interfaces::{
     dkg::{DkgPayloadValidationError, DkgPool},
@@ -15,7 +15,10 @@ use ic_types::{
     batch::ValidationContext,
     consensus::{
         Block, BlockPayload,
-        dkg::{DkgDataPayload, DkgPayloadValidationFailure, DkgSummary, InvalidDkgPayloadReason},
+        dkg::{
+            DkgDataPayload, DkgPayloadCreationError, DkgPayloadValidationFailure, DkgSummary,
+            InvalidDkgPayloadReason,
+        },
     },
 };
 use prometheus::IntCounterVec;
@@ -117,6 +120,8 @@ pub fn validate_payload(
                 });
 
             validate_dealings_payload(
+                subnet_id,
+                registry_client,
                 crypto,
                 pool_reader,
                 dkg_pool,
@@ -134,8 +139,11 @@ pub fn validate_payload(
 }
 
 // Validates the payload containing dealings.
+#[allow(clippy::too_many_arguments)]
 #[allow(clippy::result_large_err)]
 fn validate_dealings_payload(
+    subnet_id: SubnetId,
+    registry_client: &dyn RegistryClient,
     crypto: &dyn ConsensusCrypto,
     pool_reader: &PoolReader<'_>,
     dkg_pool: &dyn DkgPool,
@@ -181,6 +189,20 @@ fn validate_dealings_payload(
         return Err(InvalidDkgPayloadReason::DealerAlreadyDealt(dealer_id).into());
     }
 
+    let state = state_reader
+        .get_state_at(validation_context.certified_height)
+        .map_err(DkgPayloadCreationError::StateManagerError)?;
+
+    let remote_config_results = build_callback_id_config_map(
+        subnet_id,
+        registry_client,
+        state.get_ref(),
+        validation_context.registry_version,
+        last_summary,
+        log,
+    )?;
+    let configs = merge_configs(&last_summary.configs, &remote_config_results);
+
     // Check that all messages have a valid DKG config from the summary and the
     // dealer is valid, then verify each dealing.
     for message in &dealings.messages {
@@ -192,7 +214,7 @@ fn validate_dealings_payload(
             continue;
         }
 
-        let Some(config) = last_summary.configs.get(&message.content.dkg_id) else {
+        let Some(config) = configs.get(&message.content.dkg_id) else {
             return Err(InvalidDkgPayloadReason::MissingDkgConfigForDealing.into());
         };
 
@@ -202,14 +224,12 @@ fn validate_dealings_payload(
 
     // If we have early transcripts, we compare them
     if !dealings.transcripts_for_remote_subnets.is_empty() {
-        let expected_transcripts = create_early_remote_transcripts(
+        let expected_transcripts = payload_builder::create_early_remote_transcripts(
             pool_reader,
             crypto,
             parent,
-            last_summary,
-            state_reader,
-            validation_context,
-            log.clone(),
+            remote_config_results,
+            log,
         )?;
 
         if dealings.transcripts_for_remote_subnets != expected_transcripts {
@@ -246,11 +266,13 @@ mod tests {
         dkg::ChangeAction,
         p2p::consensus::{MutablePool, PoolMutationsProducer},
     };
+    use ic_interfaces_state_manager::Labeled;
     use ic_logger::no_op_logger;
     use ic_metrics::MetricsRegistry;
     use ic_registry_keys::make_subnet_record_key;
     use ic_test_utilities_consensus::fake::FakeContentSigner;
     use ic_test_utilities_registry::SubnetRecordBuilder;
+    use ic_test_utilities_state::get_initial_state;
     use ic_test_utilities_types::ids::{
         NODE_1, NODE_2, NODE_3, SUBNET_1, SUBNET_2, node_test_id, subnet_test_id,
     };
@@ -707,6 +729,13 @@ mod tests {
                         .build(),
                 )],
             );
+            state_manager
+                .get_mut()
+                .expect_get_latest_certified_state()
+                .return_const(Some(Labeled::new(
+                    Height::new(0),
+                    Arc::new(get_initial_state(0, 0)),
+                )));
 
             // Both summary registry versions should be 1 initially
             let summary_block = pool.as_cache().summary_block();
@@ -758,6 +787,9 @@ mod tests {
             let key_manager = Arc::new(Mutex::new(key_manager));
             let dkg_impl = DkgImpl::new(
                 node_id,
+                subnet_id,
+                registry.clone(),
+                state_manager.clone(),
                 crypto.clone(),
                 pool.get_cache(),
                 key_manager,
@@ -778,9 +810,10 @@ mod tests {
             };
 
             // It should be possible to validate the dealing
+            let configs = dkg_summary.configs.iter().collect();
             let result = dkg_impl.validate_dealings_for_dealer(
                 &dkg_pool,
-                &dkg_summary.configs,
+                &configs,
                 start_height,
                 vec![dealing],
             );

rs/consensus/dkg/src/test_utils.rs

@@ -26,7 +26,7 @@ use ic_types::{
 };
 use std::{
     collections::{BTreeMap, BTreeSet},
-    sync::Arc,
+    sync::{Arc, Mutex},
 };
 
 pub(super) fn make_setup_initial_dkg_context(
@@ -59,6 +59,23 @@ pub(super) fn make_reshare_chain_key_context(
     })
 }
 
+fn clone_subnet_call_context(context: &SubnetCallContext) -> SubnetCallContext {
+    match context {
+        SubnetCallContext::SetupInitialDKG(c) => SubnetCallContext::SetupInitialDKG(c.clone()),
+        SubnetCallContext::CanisterHttpRequest(c) => {
+            SubnetCallContext::CanisterHttpRequest(c.clone())
+        }
+        SubnetCallContext::ReshareChainKey(c) => SubnetCallContext::ReshareChainKey(c.clone()),
+        SubnetCallContext::BitcoinGetSuccessors(c) => {
+            SubnetCallContext::BitcoinGetSuccessors(c.clone())
+        }
+        SubnetCallContext::BitcoinSendTransactionInternal(c) => {
+            SubnetCallContext::BitcoinSendTransactionInternal(c.clone())
+        }
+        SubnetCallContext::SignWithThreshold(c) => SubnetCallContext::SignWithThreshold(c.clone()),
+    }
+}
+
 /// Set up the state manager mock to return an initial state containing the
 /// given subnet call contexts.
 pub(super) fn complement_state_manager_with_dkg_contexts(
@@ -73,13 +90,53 @@ pub(super) fn complement_state_manager_with_dkg_contexts(
             .subnet_call_context_manager
             .push_context(context);
     }
+    let state = Arc::new(state);
     let mut mock = state_manager.get_mut();
     let expectation = mock
         .expect_get_state_at()
-        .return_const(Ok(Labeled::new(Height::new(0), Arc::new(state))));
+        .return_const(Ok(Labeled::new(Height::new(0), state.clone())));
+    if let Some(times) = times {
+        expectation.times(times);
+    }
+
+    mock.expect_get_latest_certified_state()
+        .return_const(Some(Labeled::new(Height::new(0), state.clone())));
+}
+
+/// Set up the state manager mock to return an initial state built from a
+/// mutable context vector on every call.
+pub(super) fn complement_state_manager_with_dkg_contexts_mut(
+    state_manager: Arc<RefMockStateManager>,
+    contexts: Arc<Mutex<Vec<SubnetCallContext>>>,
+    times: Option<usize>,
+) {
+    let mut mock = state_manager.get_mut();
+    let state_for_get_state_at = contexts.clone();
+    let expectation = mock.expect_get_state_at().returning(move |_| {
+        let mut state = ic_test_utilities_state::get_initial_state(0, 0);
+        for context in state_for_get_state_at.lock().unwrap().iter() {
+            state
+                .metadata
+                .subnet_call_context_manager
+                .push_context(clone_subnet_call_context(context));
+        }
+        Ok(Labeled::new(Height::new(0), Arc::new(state)))
+    });
     if let Some(times) = times {
         expectation.times(times);
     }
+
+    let state_for_latest_certified = contexts;
+    mock.expect_get_latest_certified_state().returning(move || {
+        let mut state = ic_test_utilities_state::get_initial_state(0, 0);
+        for context in state_for_latest_certified.lock().unwrap().iter() {
+            state
+                .metadata
+                .subnet_call_context_manager
+                .push_context(clone_subnet_call_context(context));
+        }
+        Some(Labeled::new(Height::new(0), Arc::new(state)))
+    });
 }
 
 pub(super) fn complement_state_manager_with_setup_initial_dkg_request(

rs/consensus/tests/framework/runner.rs

@@ -172,6 +172,9 @@ impl<'a> ConsensusRunner<'a> {
         );
         let dkg = ic_consensus_dkg::DkgImpl::new(
             deps.replica_config.node_id,
+            deps.replica_config.subnet_id,
+            Arc::clone(&deps.registry_client),
+            deps.state_manager.clone(),
             Arc::clone(&consensus_crypto),
             deps.consensus_pool.read().unwrap().get_cache(),
             dkg_key_manager,

rs/consensus/tests/payload.rs

@@ -91,6 +91,12 @@ fn consensus_produces_expected_batches() {
                 Height::new(0),
                 Arc::new(get_initial_state(0, 0)),
             )));
+        state_manager
+            .expect_get_latest_certified_state()
+            .return_const(Some(Labeled::new(
+                Height::new(0),
+                Arc::new(get_initial_state(0, 0)),
+            )));
         state_manager
             .expect_get_certified_state_snapshot()
             .returning(|| None);
@@ -185,6 +191,9 @@ fn consensus_produces_expected_batches() {
             ic_consensus::consensus::ConsensusBouncer::new(&metrics_registry, router.clone());
         let dkg = ic_consensus_dkg::DkgImpl::new(
             replica_config.node_id,
+            replica_config.subnet_id,
+            Arc::clone(&registry_client) as Arc<_>,
+            Arc::clone(&state_manager) as Arc<_>,
             Arc::clone(&fake_crypto) as Arc<_>,
             Arc::clone(&consensus_cache),
             dkg_key_manager,

rs/replica/setup_ic_network/src/lib.rs

@@ -616,6 +616,9 @@ fn start_consensus(
         abortable_broadcast_channels.dkg,
         ic_consensus_dkg::DkgImpl::new(
             node_id,
+            subnet_id,
+            Arc::clone(&registry_client),
+            Arc::clone(&state_manager) as Arc<_>,
             Arc::clone(&consensus_crypto),
             Arc::clone(&consensus_pool_cache),
             dkg_key_manager,