use crate::assets::cleanup::{EpochData, delete_stale_triples_and_presignatures}; use crate::config::{MpcConfig, ParticipantInfo, ParticipantsConfig, SecretsConfig}; use crate::db::SecretDB; use crate::indexer::handler::ChainBlockUpdate; use crate::indexer::participants::{ ContractKeyEventInstance, ContractResharingState, ContractRunningState, ContractState, }; use crate::indexer::types::ChainSendTransactionRequest; use crate::indexer::{IndexerAPI, ReadSupportedForeignChain, tx_sender}; use crate::key_events::{ ResharingArgs, keygen_follower, keygen_leader, resharing_follower, resharing_leader, }; use crate::keyshare::{KeyshareData, KeyshareStorage}; use crate::metrics; use crate::metrics::tokio_runtime_metrics::run_monitor_loop; use crate::mpc_client::MpcClient; use crate::network::{ MeshNetworkClient, MeshNetworkTransportSender, NetworkTaskChannel, run_network_client, }; use crate::p2p::{new_tls_mesh_network, new_tls_mesh_network_with_address_updates}; use crate::primitives::{MpcTaskId, ParticipantId}; use crate::providers::ckd::CKDProvider; use crate::providers::ecdsa::triple; use crate::providers::eddsa::{EddsaSignatureProvider, EddsaTaskId}; use crate::providers::robust_ecdsa::RobustEcdsaSignatureProvider; use crate::providers::verify_foreign_tx::VerifyForeignTxProvider; use crate::providers::{DomainKeyshare, EcdsaSignatureProvider, EcdsaTaskId}; use crate::runtime::{AsyncDroppableRuntime, build_lower_priority_runtime}; use crate::storage::SignRequestStorage; use crate::storage::{CKDRequestStorage, VerifyForeignTransactionRequestStorage}; use crate::tracking::{self}; use crate::web::DebugRequest; use futures::FutureExt; use futures::future::BoxFuture; use mpc_node_config::ConfigFile; use mpc_primitives::domain::{Curve, DomainId, Protocol}; use mpc_primitives::{EpochId, ReconstructionThreshold}; use near_account_id::AccountId; use near_mpc_contract_interface::call_args as contract_args; use near_mpc_contract_interface::types as dtos; use near_time::Clock; use std::collections::HashMap; use std::future::Future; use std::sync::{Arc, Mutex}; use threshold_signatures::{confidential_key_derivation, ecdsa, frost::eddsa}; use tokio::select; use tokio::sync::mpsc::unbounded_channel; use tokio::sync::{RwLock, broadcast, mpsc, watch}; use tokio_metrics::RuntimeMonitor; use tokio_util::sync::CancellationToken; use tracing::{error, info}; /// Main entry point for the MPC node logic. Assumes the existence of an /// indexer. Queries and monitors the contract for state transitions, and act /// accordingly: if the contract says we need to generate keys, we generate /// keys; if the contract says we're running, we run the MPC protocol; if the /// contract says we need to perform key resharing, we perform key resharing. pub struct Coordinator { pub clock: Clock, pub secrets: SecretsConfig, pub config_file: ConfigFile, /// Storage for triples, presignatures, signing requests. pub secret_db: Arc, /// Storage for keyshares. pub keyshare_storage: Arc>, /// For interaction with the indexer. pub indexer: IndexerAPI, /// For testing, to know what the current state is. pub currently_running_job_name: Arc>, /// For debug UI to send us debug requests. pub debug_request_sender: broadcast::Sender, } type StopFn = Box bool + Send>; /// Represents a top-level task that we run for the current contract state. /// There is a different one of these for each contract state. struct MpcJob { /// Friendly name for the currently running task. name: &'static str, /// The future for the MPC task (keygen, resharing, or normal run). fut: BoxFuture<'static, anyhow::Result>, /// a function that looks at a new contract state and returns true iff the /// current task should be killed. stop_fn: StopFn, } /// When an MpcJob future returns successfully, it returns one of the following. #[derive(Debug)] enum MpcJobResult { /// This MpcJob has been completed successfully. Done, /// This MpcJob could not run because the contract is in a state that we /// cannot handle (such as the contract being invalid or we're not a current /// participant). If this is returned, the coordinator should do nothing /// until either timeout or the contract state changed. During this time, /// block updates are buffered. HaltUntilInterrupted, } impl Coordinator where TransactionSender: tx_sender::TransactionSender + 'static, ForeignChainPolicyReader: ReadSupportedForeignChain + Clone + Send + Sync + 'static, { pub async fn run(mut self) -> anyhow::Result<()> { loop { let state = self.indexer.contract_state_receiver.borrow().clone(); if let Some(epoch_id) = current_epoch_id(&state) { metrics::MPC_CURRENT_EPOCH_ID.set(epoch_id); } let mut job: MpcJob = match state { ContractState::Invalid => { // Invalid state. Similar to initial state; we do nothing until the state changes. MpcJob { name: "Invalid", fut: futures::future::ready(Ok(MpcJobResult::HaltUntilInterrupted)).boxed(), stop_fn: Box::new(|_| true), } } ContractState::Initializing(state) => { // For initialization state, we generate keys and vote for the public key. // We give it a timeout, so that if somehow the keygen and voting fail to // progress, we can retry. let (key_event_receiver, stop_fn) = make_initializing_stop_fn(state.key_event); MpcJob { name: "Initializing", fut: Self::create_runtime_and_run( "Initializing", self.config_file.cores, Self::run_initialization( self.secrets.clone(), self.config_file.clone(), self.keyshare_storage.clone(), state.participants.clone(), self.indexer.txn_sender.clone(), key_event_receiver, ), )?, stop_fn, } } ContractState::Running(running_state) => { tracing::info!("Resharing process is: {:?}", &running_state.resharing_state); let (job_name, key_event_receiver, stop_fn): (_, _, StopFn) = match running_state.resharing_state.clone() { Some(resharing_state) => { let (receiver, stop_fn) = make_resharing_stop_fn(resharing_state); ("Resharing", Some(receiver), stop_fn) } None => { let stop_fn = make_running_stop_fn( running_state.keyset.epoch_id, running_state.participants.clone(), self.config_file.my_near_account_id.clone(), ); ("Running", None, stop_fn) } }; MpcJob { name: job_name, fut: Self::create_runtime_and_run( "Running", self.config_file.cores, Self::run_mpc( self.clock.clone(), self.secret_db.clone(), self.secrets.clone(), self.config_file.clone(), self.keyshare_storage.clone(), running_state.clone(), self.indexer.txn_sender.clone(), self.indexer.foreign_chain_policy_reader.clone(), self.indexer .block_update_receiver .clone() .lock_owned() .await, self.debug_request_sender.subscribe(), key_event_receiver, self.indexer.contract_state_receiver.clone(), ), )?, stop_fn, } } }; tracing::info!("[{}] Starting", job.name); let _report_guard = ReportCurrentJobGuard::new(job.name, self.currently_running_job_name.clone()); loop { tokio::select! { res = &mut job.fut => { match res { Err(e) => { tracing::error!("[{}] failed: {:?}", job.name, e); break; } Ok(MpcJobResult::Done) => { tracing::info!("[{}] finished successfully", job.name); break; } Ok(MpcJobResult::HaltUntilInterrupted) => { tracing::info!("[{}] halted; waiting for state change or timeout", job.name); // Replace it with a never-completing future so next iteration we wait for // only state change or timeout. job.fut = futures::future::pending().boxed(); continue; } } } res = self.indexer.contract_state_receiver.changed() => { if res.is_err() { anyhow::bail!("[{}] contract state receiver closed", job.name); } if (job.stop_fn)(&self.indexer.contract_state_receiver.borrow()) { tracing::info!( "[{}] contract state changed incompatibly, stopping", job.name ); break; } } } } } } fn create_runtime_and_run( description: &str, cores: Option, task: impl Future> + Send + 'static, ) -> anyhow::Result>> { let task_handle = tracking::current_task(); // Create a separate runtime, as opposed to making a runtime when the // binary starts, for these reasons: // - so that we can limit the number of cores used for MPC tasks, // in order to avoid starving the indexer, causing it to fall behind. // - so that we can ensure that all MPC tasks are shut down when we // encounter contract state transitions. By dropping the entire // runtime, we can ensure that all tasks are stopped. Otherwise, it // would be very difficult and error-prone to ensure we don't leave // some long-running task behind. let mpc_runtime = if let Some(n_threads) = cores { tokio::runtime::Builder::new_multi_thread() .worker_threads(std::cmp::max(n_threads, 1)) .enable_all() .build()? } else { tokio::runtime::Runtime::new()? }; let runtime_handle = mpc_runtime.handle(); let runtime_monitor = RuntimeMonitor::new(runtime_handle); // run as long as the runtime is alive mpc_runtime.spawn(run_monitor_loop("mpc", runtime_monitor)); let mpc_runtime = AsyncDroppableRuntime::new(mpc_runtime); let fut = mpc_runtime.spawn(task_handle.scope(description, task)); Ok(async move { let _mpc_runtime = mpc_runtime; anyhow::Ok(fut.await??) } .boxed()) } /// Builds the lower-priority runtime that CPU-heavy asset generation runs on, /// so the OS preempts it whenever signing is ready. Returns the runtime — to /// be kept alive for the duration of the run — alongside the handle that /// generation tasks spawn on. When disabled there is no separate runtime and /// the handle is the current one, so generation shares the MPC runtime. Must /// be called from within the MPC runtime so `Handle::current()` resolves to it. fn build_gen_runtime( config_file: &ConfigFile, ) -> anyhow::Result<(Option, tokio::runtime::Handle)> { let gen_runtime = config_file .separate_asset_generation_runtime .then(|| { let worker_threads = config_file.cores.unwrap_or_else(|| { std::thread::available_parallelism() .map(|n| n.get()) .unwrap_or(1) }); build_lower_priority_runtime(worker_threads, "mpc-gen") .map(AsyncDroppableRuntime::new) }) .transpose()?; if let Some(runtime) = &gen_runtime { // Metrics published under the "gen" runtime label (the MPC runtime // uses "mpc"), so the two runtimes stay distinct series. runtime.spawn(run_monitor_loop( "gen", RuntimeMonitor::new(runtime.handle()), )); } let gen_runtime_handle = gen_runtime .as_ref() .map_or_else(tokio::runtime::Handle::current, |runtime| { runtime.handle().clone() }); Ok((gen_runtime, gen_runtime_handle)) } /// Entry point to handle the Initializing state of the contract. async fn run_initialization( secrets: SecretsConfig, config_file: ConfigFile, keyshare_storage: Arc>, participants: ParticipantsConfig, chain_txn_sender: TransactionSender, key_event_receiver: watch::Receiver, ) -> anyhow::Result { let p2p_key = &secrets.persistent_secrets.p2p_private_key; let Some(mpc_config) = MpcConfig::from_participants_with_near_account_id( participants, &config_file.my_near_account_id, &p2p_key.verifying_key(), ) else { tracing::info!( "We are not a participant in the current epoch; doing nothing until contract state change" ); return Ok(MpcJobResult::HaltUntilInterrupted); }; tracking::set_progress(&format!( "Generating key(s) as participant {}", mpc_config.my_participant_id )); let (sender, receiver) = new_tls_mesh_network(&mpc_config, p2p_key).await?; let (network_client, channel_receiver, _handle) = run_network_client(Arc::new(sender), Box::new(receiver)); if mpc_config.is_leader_for_key_event() { keygen_leader( network_client, keyshare_storage, key_event_receiver, chain_txn_sender, ) .await?; } else { keygen_follower( channel_receiver, keyshare_storage, key_event_receiver, chain_txn_sender, ) .await?; } Ok(MpcJobResult::Done) } /// Entry point to handle the Running state of the contract. /// In this state, we generate triples and presignatures, and listen to /// signature requests and submit signature responses. #[expect(clippy::too_many_arguments)] async fn run_mpc( clock: Clock, secret_db: Arc, secrets: SecretsConfig, config_file: ConfigFile, keyshare_storage: Arc>, running_state: ContractRunningState, chain_txn_sender: TransactionSender, foreign_chain_policy_reader: ForeignChainPolicyReader, block_update_receiver: tokio::sync::OwnedMutexGuard< mpsc::UnboundedReceiver, >, debug_request_receiver: broadcast::Receiver, resharing_state_receiver: Option>, contract_state_receiver: watch::Receiver, ) -> anyhow::Result { tracing::info!("Entering running state."); // `_gen_runtime` is kept alive for the lifetime of `run` below; // `AsyncDroppableRuntime` lets it be dropped from this async context on // teardown. let (_gen_runtime, gen_runtime_handle) = Self::build_gen_runtime(&config_file)?; let my_participant_id = running_state .participants .get_participant_id(&config_file.my_near_account_id); if let Some(my_participant_id) = my_participant_id { let current_participants_config = running_state.participants.clone(); let current_epoch_id = running_state.keyset.epoch_id; let all_domains: Vec<_> = running_state.keyset.get_domain_ids(); let current_epoch_data = EpochData { epoch_id: current_epoch_id, participants: current_participants_config, }; let triple_thresholds = triple::caitsith_triple_thresholds(&running_state.domains); delete_stale_triples_and_presignatures( &secret_db, current_epoch_data, my_participant_id, all_domains, triple_thresholds, )?; } let mut running_participants = running_state.participants.clone(); let participants_config = match &running_state.resharing_state { Some(resharing_state) => resharing_state.new_participants.clone(), None => running_participants.clone(), }; // Only consider the running participants that are also members of the new resharing state. running_participants .participants .retain(|p| participants_config.participants.contains(p)); let p2p_key = &secrets.persistent_secrets.p2p_private_key; let Some(mpc_config) = MpcConfig::from_participants_with_near_account_id( participants_config, &config_file.my_near_account_id, &p2p_key.verifying_key(), ) else { tracing::info!( "We are not a participant in the current epoch; doing nothing until contract state change" ); return Ok(MpcJobResult::HaltUntilInterrupted); }; register_foreign_chains(&chain_txn_sender, &config_file.foreign_chains).await; let resolve_peer_address = move |participant_id| peer_address_from_state(&contract_state_receiver, participant_id); tracing::info!("Creating tls mesh"); let (sender, receiver) = new_tls_mesh_network_with_address_updates(&mpc_config, p2p_key, resolve_peer_address) .await?; let sender = Arc::new(sender); tracing::info!("Creating network client."); let (network_client, mut channel_receiver, _handle) = run_network_client(sender.clone(), Box::new(receiver)); let cancellation_token = CancellationToken::new(); let cancellation_token_child = cancellation_token.child_token(); let _drop_guard = cancellation_token.drop_guard(); let (running_network_receiver, resharing_network_receiver) = { let (running_sender, running_receiver) = unbounded_channel(); let (resharing_sender, resharing_receiver) = unbounded_channel(); let _multiplexer_handle = tokio::spawn(async move { loop { select! { network_channel = channel_receiver.recv() => { let Some(network_channel) = network_channel else { tracing::info!("Network channel dropped."); break; }; let is_resharing_message = matches!( network_channel.task_id(), MpcTaskId::EcdsaTaskId(EcdsaTaskId::KeyResharing { .. }) | MpcTaskId::EddsaTaskId(EddsaTaskId::KeyResharing { .. }) ); if is_resharing_message { let send_result = resharing_sender.send(network_channel); if send_result.is_err() { error!("resharing receiver dropped."); } } else { let send_result = running_sender.send(network_channel); if send_result.is_err() { error!("running receiver dropped."); } } } _ = cancellation_token_child.cancelled() => { info!("Network multiplexer cancelled."); break; } } } info!("Exiting network multiplexer."); }); (running_receiver, resharing_receiver) }; // This handle must be alive, otherwise the AutoAbortTask will get cancelled on drop. let resharing_handle = resharing_state_receiver.map(|resharing_state_receiver| { let config_file = config_file.clone(); let running_state = running_state.clone(); let keyshare_storage = keyshare_storage.clone(); let chain_txn_sender = chain_txn_sender.clone(); let network_client = network_client.clone(); let mpc_config = mpc_config.clone(); tracking::spawn_checked("key resharing", async move { Self::run_key_resharing( &config_file, keyshare_storage.clone(), running_state.clone(), &mpc_config, network_client, resharing_network_receiver, chain_txn_sender, resharing_state_receiver, ) .await }) }); let p2p_public_key = p2p_key.verifying_key(); let running_handle = tracking::spawn::<_, anyhow::Result>( "running mpc job", async move { let Some(running_mpc_config) = MpcConfig::from_participants_with_near_account_id( running_participants.clone(), &config_file.my_near_account_id, &p2p_public_key, ) else { tracing::info!( "We are not a participant in the current epoch; doing nothing until contract state change" ); return Ok(MpcJobResult::HaltUntilInterrupted); }; let keyshares = match keyshare_storage .write() .await .update_permanent_keyshares(&running_state.keyset) .await { Ok(keyshares) => keyshares, Err(e) => { tracing::error!( "Failed to load keyshares: {:?}; doing nothing until contract state changes.", e ); return Ok(MpcJobResult::HaltUntilInterrupted); } }; if keyshares.is_empty() { tracing::info!("We have no keyshares. Waiting for Initialization."); return Ok(MpcJobResult::HaltUntilInterrupted); } tracking::set_progress(&format!( "Running epoch {:?} as participant {}", running_state.keyset.epoch_id, running_mpc_config.my_participant_id )); tracing::info!("wait for ready."); let running_participant_ids = running_mpc_config .participants .participants .iter() .map(|p| p.id) .collect::>(); sender .wait_for_ready( running_mpc_config.participants.threshold.try_into()?, &running_participant_ids, ) .await?; let sign_request_store = Arc::new(SignRequestStorage::new(secret_db.clone())?); let ckd_request_store = Arc::new(CKDRequestStorage::new(secret_db.clone())?); let verify_foreign_tx_request_store = Arc::new( VerifyForeignTransactionRequestStorage::new(secret_db.clone())?, ); let mut ecdsa_keyshares: HashMap< mpc_primitives::domain::DomainId, DomainKeyshare, > = HashMap::new(); let mut robust_ecdsa_keyshares: HashMap< mpc_primitives::domain::DomainId, DomainKeyshare, > = HashMap::new(); let mut eddsa_keyshares: HashMap< mpc_primitives::domain::DomainId, DomainKeyshare, > = HashMap::new(); let mut ckd_keyshares: HashMap< mpc_primitives::domain::DomainId, DomainKeyshare, > = HashMap::new(); let domain_registry: HashMap = running_state .domains .iter() .map(|d| (d.id, (d.protocol, d.reconstruction_threshold))) .collect(); for keyshare in keyshares { let domain_id = keyshare.key_id.domain_id; let Some((protocol, reconstruction_threshold)) = domain_registry.get(&domain_id).copied() else { anyhow::bail!( "Keyshare references domain {domain_id:?} which is not in the contract registry", ); }; let expected_curve = Curve::from(protocol); match (expected_curve, keyshare.data) { (Curve::Secp256k1, KeyshareData::Secp256k1(data)) => match protocol { Protocol::CaitSith => { ecdsa_keyshares.insert( domain_id, DomainKeyshare::new(data, reconstruction_threshold), ); } Protocol::DamgardEtAl => { robust_ecdsa_keyshares.insert( domain_id, DomainKeyshare::new(data, reconstruction_threshold), ); } other => anyhow::bail!( "Unexpected protocol {other:?} for Secp256k1 keyshare on domain {domain_id:?}", ), }, (Curve::Edwards25519, KeyshareData::Ed25519(data)) => { eddsa_keyshares.insert( domain_id, DomainKeyshare::new(data, reconstruction_threshold), ); } (Curve::Bls12381, KeyshareData::Bls12381(data)) => { ckd_keyshares.insert( domain_id, DomainKeyshare::new(data, reconstruction_threshold), ); } (expected, data) => anyhow::bail!( "Keyshare data does not match the domain protocol's expected curve: domain_id={:?}, protocol={:?}, expected_curve={:?}, data_kind={:?}", domain_id, protocol, expected, std::mem::discriminant(&data), ), } } let domain_to_protocol: HashMap = domain_registry .into_iter() .map(|(id, (protocol, _))| (id, protocol)) .collect(); let ecdsa_signature_provider = Arc::new(EcdsaSignatureProvider::new( config_file.clone().into(), running_mpc_config.clone().into(), network_client.clone(), clock.clone(), secret_db.clone(), sign_request_store.clone(), ecdsa_keyshares, )?); let robust_ecdsa_signature_provider = Arc::new(RobustEcdsaSignatureProvider::new( config_file.clone().into(), running_mpc_config.clone().into(), network_client.clone(), clock, secret_db, sign_request_store.clone(), robust_ecdsa_keyshares, )?); let eddsa_signature_provider = Arc::new(EddsaSignatureProvider::new( config_file.clone().into(), running_mpc_config.clone().into(), network_client.clone(), sign_request_store.clone(), eddsa_keyshares, )); let ckd_provider = Arc::new(CKDProvider::new( config_file.clone().into(), running_mpc_config.clone().into(), network_client.clone(), ckd_request_store.clone(), ckd_keyshares, )); let verify_foreign_tx_provider = Arc::new(VerifyForeignTxProvider::new( config_file.clone().into(), foreign_chain_policy_reader.clone(), verify_foreign_tx_request_store.clone(), ecdsa_signature_provider.clone(), )?); let mpc_client = Arc::new(MpcClient::new( config_file.into(), network_client, sign_request_store, ckd_request_store, verify_foreign_tx_request_store, ecdsa_signature_provider, robust_ecdsa_signature_provider, eddsa_signature_provider, ckd_provider, verify_foreign_tx_provider, domain_to_protocol, gen_runtime_handle, )); mpc_client .run( running_network_receiver, block_update_receiver, chain_txn_sender, debug_request_receiver, ) .await?; Ok(MpcJobResult::Done) }, ); if let Some(resharing_handle) = resharing_handle { tracing::info!("Waiting on resharing handle."); resharing_handle.await?; } running_handle.await? } /// Entry point to handle the Resharing state of the contract. #[expect(clippy::too_many_arguments)] async fn run_key_resharing( config_file: &ConfigFile, keyshare_storage: Arc>, current_running_state: ContractRunningState, mpc_config: &MpcConfig, network_client: Arc, channel_receiver: mpsc::UnboundedReceiver, chain_txn_sender: TransactionSender, key_event_receiver: watch::Receiver, ) -> anyhow::Result { tracing::info!("Starting key resharing."); let previous_keyset = current_running_state.keyset; let was_participant_last_epoch = current_running_state .participants .participants .iter() .any(|p| p.near_account_id == config_file.my_near_account_id); let existing_keyshares = if was_participant_last_epoch { let keyshares = match keyshare_storage .write() .await .update_permanent_keyshares(&previous_keyset) .await { Ok(x) => x, Err(e) => { tracing::error!( "Failed to load keyshare for epoch {:?}: {:?}; doing nothing until contract state change", previous_keyset.epoch_id, e ); return Ok(MpcJobResult::HaltUntilInterrupted); } }; Some(keyshares) } else { info!("Not participant in last epoch."); if keyshare_storage .write() .await .update_permanent_keyshares(&previous_keyset) .await .is_ok() { tracing::warn!( "We should not have the previous keyshares when we were not a participant last epoch" ); } None }; let old_reconstruction_thresholds: HashMap = current_running_state .domains .iter() .map(|d| (d.id, d.reconstruction_threshold)) .collect(); let args = Arc::new(ResharingArgs { previous_keyset, existing_keyshares, old_reconstruction_thresholds, old_participants: current_running_state.participants, }); if mpc_config.is_leader_for_key_event() { resharing_leader( network_client, keyshare_storage, key_event_receiver, chain_txn_sender, args, ) .await?; } else { resharing_follower( channel_receiver, keyshare_storage, key_event_receiver, chain_txn_sender, args, ) .await?; } Ok(MpcJobResult::Done) } } /// Simple RAII to export current job name to metrics and /debug/tasks. struct ReportCurrentJobGuard { name: String, currently_running_job_name: Arc>, } impl ReportCurrentJobGuard { fn new(name: &str, currently_running_job_name: Arc>) -> Self { metrics::MPC_CURRENT_JOB_STATE .with_label_values(&[name]) .inc(); tracking::set_progress(name); *currently_running_job_name.lock().unwrap() = name.to_string(); Self { name: name.to_string(), currently_running_job_name, } } } impl Drop for ReportCurrentJobGuard { fn drop(&mut self) { metrics::MPC_CURRENT_JOB_STATE .with_label_values(&[&self.name]) .dec(); tracking::set_progress("Transitioning state"); *self.currently_running_job_name.lock().unwrap() = "".to_string(); } } fn current_epoch_id(state: &ContractState) -> Option { state .epoch_id() .and_then(|epoch_id| i64::try_from(epoch_id.get()).ok()) } /// The `host:port` a peer is currently reachable at in live contract state, re-read on every /// (re)connect so a peer's URL update is picked up without a restart. fn peer_address_from_state( contract_state_receiver: &watch::Receiver, participant_id: ParticipantId, ) -> Option { contract_state_receiver .borrow() .mesh_participants() .and_then(|participants| { participants .get_info(participant_id) .map(|info| format!("{}:{}", info.address, info.port)) }) } /// Whether a participant-set change forces a job restart rather than being absorbed live. A peer /// address/port change is hot-swapped ([`peer_address_from_state`]); only a change to identity or /// our *own* listening port (which re-binds the listener) needs a restart. The governance /// `threshold` does not: per-domain reconstruction thresholds drive the running protocol, so a /// threshold-only change is absorbed live. fn participants_change_requires_restart( old: &ParticipantsConfig, new: &ParticipantsConfig, my_near_account_id: &AccountId, ) -> bool { // Destructured exhaustively so a new field forces a restart-vs-hot-swap decision here. let identities = |cfg: &ParticipantsConfig| { let mut ids: Vec<_> = cfg .participants .iter() .map(|p| { let ParticipantInfo { id, address: _, port: _, p2p_public_key, near_account_id, } = p; (*id, near_account_id.clone(), p2p_public_key.to_bytes()) }) .collect(); ids.sort(); ids }; if identities(old) != identities(new) { return true; } let my_port = |cfg: &ParticipantsConfig| { cfg.get_info_by_account_id(my_near_account_id) .map(|p| p.port) }; my_port(old) != my_port(new) } /// returns true if one of the following occurs: /// - the epoch id changes /// - a resharing starts /// - the participant set changes in a way that requires a restart /// (see [`participants_change_requires_restart`]) fn stop_running( new_state: &ContractState, current_running_epoch_id: EpochId, current_participant_set: &ParticipantsConfig, my_near_account_id: &AccountId, ) -> bool { match new_state { ContractState::Running(new_state) => { if new_state.keyset.epoch_id != current_running_epoch_id { tracing::info!("Epoch id changed."); return true; } if new_state.resharing_state.is_some() { tracing::info!("A resharing started."); return true; } if participants_change_requires_restart( current_participant_set, &new_state.participants, my_near_account_id, ) { tracing::info!("Participant set changed in a way that requires a restart."); return true; } false } _ => { tracing::info!("No longer in Running state."); true } } } fn make_running_stop_fn( current_running_epoch_id: EpochId, current_participant_set: ParticipantsConfig, my_near_account_id: AccountId, ) -> StopFn { Box::new(move |new_state| { stop_running( new_state, current_running_epoch_id, ¤t_participant_set, &my_near_account_id, ) }) } /// returns true if one of the following occurs: /// - epoch id changed /// - resharing concludes /// - key event receiver closes the channel. fn stop_resharing( new_state: &ContractState, current_resharing_epoch_id: EpochId, key_event_sender: &tokio::sync::watch::Sender, ) -> bool { match new_state { ContractState::Running(new_state) => { let Some(new_resharing_state) = &new_state.resharing_state else { tracing::info!("Concluded resharing state."); return true; }; if new_resharing_state.key_event.id.epoch_id != current_resharing_epoch_id { tracing::info!("Epoch changed. We exit resharing state."); return true; } if key_event_sender .send(new_resharing_state.key_event.clone()) .is_err() { tracing::info!("Key event receiver closed."); return true; } false } _ => true, } } fn make_resharing_stop_fn( resharing_state: ContractResharingState, ) -> (watch::Receiver, StopFn) { let (key_event_sender, key_event_receiver) = watch::channel(resharing_state.key_event.clone()); let current_resharing_epoch_id = resharing_state.key_event.id.epoch_id; let stop_fn = Box::new(move |new_state: &ContractState| { stop_resharing(new_state, current_resharing_epoch_id, &key_event_sender) }); (key_event_receiver, stop_fn) } fn stop_initializing( new_state: &ContractState, current_epoch_id: EpochId, key_event_sender: &tokio::sync::watch::Sender, ) -> bool { match new_state { ContractState::Initializing(new_state) => { if new_state.key_event.id.epoch_id != current_epoch_id { tracing::info!("Epoch id changed"); return true; } if key_event_sender.send(new_state.key_event.clone()).is_err() { tracing::info!("Key event receiver closed"); return true; } false } _ => { tracing::info!("Protocol State changed."); true } } } /// Dual-writes the node's foreign-chain registration (legacy + new endpoint); /// an empty config still registers so that dropping every chain propagates. /// TODO(#3630): drop the legacy RegisterForeignChainConfig half. async fn register_foreign_chains( chain_txn_sender: &impl tx_sender::TransactionSender, foreign_chains: &mpc_node_config::ForeignChainsConfig, ) { let foreign_chain_configuration = foreign_chains.configured_chains(); if let Err(err) = chain_txn_sender .send(ChainSendTransactionRequest::RegisterForeignChainConfig( contract_args::RegisterForeignChainConfigArgs::new(foreign_chain_configuration), )) .await { tracing::warn!(error = ?err, "failed to send register supported foreign chains transaction"); } let foreign_chains_config: dtos::ForeignChainsConfig = foreign_chains .iter_chains() .map(|(chain, _)| chain) .collect::>() .into(); if let Err(err) = chain_txn_sender .send(ChainSendTransactionRequest::RegisterForeignChainsConfig( contract_args::RegisterForeignChainsConfigArgs::new(foreign_chains_config), )) .await { tracing::warn!(error = ?err, "failed to send register foreign chains config transaction"); } } fn make_initializing_stop_fn( key_event: ContractKeyEventInstance, ) -> (watch::Receiver, StopFn) { let (key_event_sender, key_event_receiver) = watch::channel(key_event.clone()); let key_event_sender = key_event_sender.clone(); ( key_event_receiver, Box::new(move |new_state| { stop_initializing(new_state, key_event.id.epoch_id, &key_event_sender) }), ) } #[cfg(test)] #[expect(non_snake_case)] mod tests { use super::{ current_epoch_id, participants_change_requires_restart, peer_address_from_state, register_foreign_chains, stop_running, }; use crate::indexer::participants::ContractState; use crate::indexer::participants::test_utils::{ base_config, me, participant, resharing, running, }; use crate::indexer::types::ChainSendTransactionRequest; use crate::primitives::ParticipantId; use crate::tests::common::MockTransactionSender; use assert_matches::assert_matches; use ed25519_dalek::SigningKey; use mpc_node_config::foreign_chains::RpcProviderName; use mpc_node_config::{ForeignChainConfig, ForeignChainProviderConfig, ForeignChainsConfig}; use mpc_primitives::EpochId; use near_mpc_contract_interface::types as dtos; use rand::SeedableRng; use rand::rngs::StdRng; use rstest::rstest; use std::collections::BTreeSet; use std::num::NonZeroU64; use tokio::sync::mpsc; use tokio::sync::mpsc::error::TryRecvError; use tokio::sync::watch; #[test] fn participants_change_requires_restart__should_be_false_for_peer_address_change() { // Given let old = base_config(); let mut new = base_config(); new.participants[1].address = "bob-new.example.com".to_string(); new.participants[1].port = 9090; // When / Then assert!(!participants_change_requires_restart(&old, &new, &me())); } #[test] fn participants_change_requires_restart__should_be_true_for_own_port_change() { // Given let old = base_config(); let mut new = base_config(); new.participants[0].port = 9090; // When / Then assert!(participants_change_requires_restart(&old, &new, &me())); } /// Our own host is not dialed by us (we bind 0.0.0.0), so a change to it needs no restart. #[test] fn participants_change_requires_restart__should_be_false_for_own_address_change() { // Given let old = base_config(); let mut new = base_config(); new.participants[0].address = "alice-new.example.com".to_string(); // When / Then assert!(!participants_change_requires_restart(&old, &new, &me())); } #[test] fn participants_change_requires_restart__should_be_true_for_tls_key_change() { // Given let old = base_config(); let mut new = base_config(); new.participants[1].p2p_public_key = SigningKey::generate(&mut StdRng::seed_from_u64(42)).verifying_key(); // When / Then assert!(participants_change_requires_restart(&old, &new, &me())); } #[test] fn participants_change_requires_restart__should_be_true_for_membership_change() { // Given let old = base_config(); let mut new = base_config(); new.participants .push(participant(2, "carol.near", "carol.example.com", 8080, 3)); // When / Then assert!(participants_change_requires_restart(&old, &new, &me())); } #[test] fn participants_change_requires_restart__should_be_false_for_governance_threshold_change() { // Given let old = base_config(); let mut new = base_config(); new.threshold = 1; // When / Then assert!(!participants_change_requires_restart(&old, &new, &me())); } #[test] fn stop_running__should_be_false_for_peer_address_change() { // Given let current = base_config(); let mut updated = base_config(); updated.participants[1].address = "bob-new.example.com".to_string(); // When / Then assert!(!stop_running( &running(updated, 5), EpochId::new(5), ¤t, &me() )); } #[test] fn stop_running__should_be_true_for_epoch_change() { // Given let current = base_config(); // When / Then assert!(stop_running( &running(base_config(), 6), EpochId::new(5), ¤t, &me() )); } #[test] fn stop_running__should_be_true_when_no_longer_running() { // Given let current = base_config(); // When / Then assert!(stop_running( &ContractState::Invalid, EpochId::new(5), ¤t, &me() )); } #[rstest] #[case::running(running(base_config(), 7), Some(7))] #[case::resharing_keeps_the_old_epoch(resharing(base_config(), base_config(), 7), Some(7))] #[case::invalid(ContractState::Invalid, None)] #[case::epoch_id_beyond_i64(running(base_config(), u64::MAX), None)] fn current_epoch_id__should_report_the_running_keyset_epoch( #[case] state: ContractState, #[case] expected: Option, ) { // When / Then assert_eq!(current_epoch_id(&state), expected); } #[test] fn peer_address_from_state__should_resolve_running_peer_host_and_port() { // Given let (_tx, rx) = watch::channel(running(base_config(), 5)); // When let address = peer_address_from_state(&rx, ParticipantId::from_raw(1)); // Then assert_eq!(address, Some("bob.example.com:8080".to_string())); } #[test] fn peer_address_from_state__should_reflect_updated_address() { // Given let (tx, rx) = watch::channel(running(base_config(), 5)); let mut updated = base_config(); updated.participants[1].address = "bob-new.example.com".to_string(); updated.participants[1].port = 9090; // When tx.send(running(updated, 5)).unwrap(); // Then assert_eq!( peer_address_from_state(&rx, ParticipantId::from_raw(1)), Some("bob-new.example.com:9090".to_string()) ); } #[test] fn peer_address_from_state__should_be_none_when_not_running() { // Given let (_tx, rx) = watch::channel(ContractState::Invalid); // When / Then assert_eq!( peer_address_from_state(&rx, ParticipantId::from_raw(1)), None ); } #[test] fn peer_address_from_state__should_resolve_against_new_participants_during_resharing() { // Given let mut new_participants = base_config(); new_participants.participants[1].address = "bob-reshared.example.com".to_string(); new_participants.participants[1].port = 9090; new_participants.participants.push(participant( 2, "carol.near", "carol.example.com", 7000, 3, )); let (_tx, rx) = watch::channel(resharing(base_config(), new_participants, 5)); // When let moved_peer = peer_address_from_state(&rx, ParticipantId::from_raw(1)); let joining_peer = peer_address_from_state(&rx, ParticipantId::from_raw(2)); // Then assert_eq!( moved_peer, Some("bob-reshared.example.com:9090".to_string()) ); assert_eq!(joining_peer, Some("carol.example.com:7000".to_string())); } /// Guards the upgrade-window dual-write: the legacy registration must keep /// being emitted alongside the new one until #3630 drops it. #[tokio::test] async fn register_foreign_chains__should_send_legacy_and_new_registrations() { // Given: a node config covering Solana. let foreign_chains = ForeignChainsConfig { solana: Some(ForeignChainConfig { timeout_sec: NonZeroU64::new(30).unwrap(), max_retries: NonZeroU64::new(3).unwrap(), expected_network_fingerprint: None, providers: near_mpc_bounded_collections::NonEmptyBTreeMap::new( RpcProviderName::from("public".to_string()), ForeignChainProviderConfig { rpc_url: "https://rpc.public.example.com".to_string(), auth: Default::default(), }, ), }), ..Default::default() }; let (sender, mut receiver) = mpsc::channel(10); let txn_sender = MockTransactionSender { transaction_sender: sender, }; // When register_foreign_chains(&txn_sender, &foreign_chains).await; // Then: the legacy registration is emitted first, then the new one. let expected_legacy = foreign_chains.configured_chains(); assert_matches!( receiver.try_recv(), Ok(ChainSendTransactionRequest::RegisterForeignChainConfig(args)) if args.foreign_chain_configuration == expected_legacy ); let expected: dtos::ForeignChainsConfig = BTreeSet::from([dtos::ForeignChain::Solana]).into(); assert_matches!( receiver.try_recv(), Ok(ChainSendTransactionRequest::RegisterForeignChainsConfig(args)) if args.foreign_chains_config == expected ); assert_matches!(receiver.try_recv(), Err(TryRecvError::Empty)); } }