diff --git a/crates/node/src/assets.rs b/crates/node/src/assets.rs index a4fe2b1c4..576fea9c0 100644 --- a/crates/node/src/assets.rs +++ b/crates/node/src/assets.rs @@ -165,6 +165,32 @@ impl ColdQueue { self.cold_queue.push_back((id, value)); ColdQueueAddIfNotSatisfiedResult::Enqueued } + + /// Adds an element whose status against the stored condition is unknown: + /// inserted at the `cold_available` barrier so both sweep barriers stay + /// valid for `take`/`discard`. + pub(self) fn ingest(&mut self, id: UniqueId, value: T) { + self.cold_queue.insert(self.cold_available, (id, value)); + self.cold_available += 1; + } + + /// Removes the first element matching `cond_val` (a caller-supplied value, + /// independent of the stored condition value), shifting the barriers that + /// lie beyond the removed position. + pub(self) fn take_first_matching(&mut self, cond_val: &CondVal) -> Option<(UniqueId, T)> { + let pos = self + .cold_queue + .iter() + .position(|(_, value)| (self.condition)(cond_val, value))?; + let result = self.cold_queue.remove(pos); + if pos < self.cold_ready { + self.cold_ready -= 1; + } + if pos < self.cold_available { + self.cold_available -= 1; + } + result + } } enum ColdQueueTakeResult { @@ -270,6 +296,32 @@ where } } + pub async fn take_owned_matching(&self, cond_val: CondVal) -> (UniqueId, T) { + loop { + { + let mut cold = self.cold_queue.lock().unwrap(); + // The stored condition value doesn't affect matching (the + // predicate runs against `cond_val`); refreshed anyway so + // matching-heavy periods don't leave it stale for + // `take_owned`/`discard`. + cold.update_condition_value_if_due(); + while let Some(Ok((id, value))) = self.hot_receiver.recv_async().now_or_never() { + cold.ingest(id, value); + } + if let Some(taken) = cold.take_first_matching(&cond_val) { + return taken; + } + } + tokio::select! { + _ = self.clock.sleep(near_time::Duration::seconds(1)) => {} + received = self.hot_receiver.recv_async() => { + let (id, value) = received.unwrap(); + self.cold_queue.lock().unwrap().ingest(id, value); + } + } + } + } + /// Process `num_elements_to_process`, removing any that doesn't satisfy condition. /// Return ids, that were removed from cold storage. pub async fn maybe_discard_owned(&self, mut num_elements_to_process: usize) -> Vec { @@ -504,6 +556,18 @@ where (id, asset) } + /// Takes an owned asset whose participants are all in `active`, waiting for + /// background generation if none is currently available. + pub async fn take_owned_matching(&self, active: Vec) -> (UniqueId, T) { + let (id, asset) = self.owned_queue.take_owned_matching(active).await; + let mut update = self.db.update(); + update.delete(self.col, &self.make_key(id)); + update + .commit() + .expect("Unrecoverable error writing to database"); + (id, asset) + } + /// Adds an owned asset to the storage. pub fn add_owned(&self, id: UniqueId, value: T) { let key = self.make_key(id); @@ -748,6 +812,158 @@ mod tests { assert_eq!(queue.available(), 0); } + #[tokio::test] + #[expect(non_snake_case)] + async fn take_owned_matching__should_return_matching_asset_immediately() { + // Given: a queue holding one non-matching and one matching asset. + let clock = FakeClock::default(); + let queue = DoubleQueue::new(clock.clock(), |cond, val| val % 2 == *cond, Arc::new(|| 0)); + let id1 = UniqueId::new(ParticipantId::from_raw(42), 123, 456); + let id2 = id1.add_to_counter(1).unwrap(); + queue.add_owned(id1, 2); + queue.add_owned(id2, 3); + + // When: taking with a condition value matching only the second asset. + let taken = queue.take_owned_matching(1).now_or_never(); + + // Then + assert_eq!(taken, Some((id2, 3))); + } + + #[tokio::test] + #[expect(non_snake_case)] + async fn take_owned_matching__should_wait_until_matching_asset_is_added() { + // Given: a queue holding only a non-matching asset. + let clock = FakeClock::default(); + let queue = DoubleQueue::new(clock.clock(), |cond, val| val % 2 == *cond, Arc::new(|| 0)); + let id1 = UniqueId::new(ParticipantId::from_raw(42), 123, 456); + let id2 = id1.add_to_counter(1).unwrap(); + queue.add_owned(id1, 2); + + // When: taking with a condition value nothing matches yet. + let mut take = Box::pin(queue.take_owned_matching(1)); + assert!((&mut take).now_or_never().is_none()); + + // Then: the take completes once a matching asset is generated. + queue.add_owned(id2, 3); + assert_eq!(take.await, (id2, 3)); + // And: the non-matching asset is still in the queue. + assert_eq!(queue.available(), 1); + } + + #[test] + #[expect(non_snake_case)] + fn cold_queue_ingest_and_take_first_matching__should_preserve_barriers() { + // Given: a ready prefix and a non-satisfying tail (stored condition: + // even values satisfy condition value 0). + let clock = FakeClock::default(); + let cond_value = Arc::new(AtomicI32::new(0)); + let mut queue = ColdQueue::new(clock.clock(), |cond, val| val % 2 == *cond, { + let cond_value = cond_value.clone(); + Arc::new(move || cond_value.load(Ordering::Relaxed)) + }); + let id1 = UniqueId::new(ParticipantId::from_raw(42), 1, 0); + let id2 = id1.add_to_counter(1).unwrap(); + let id3 = id1.add_to_counter(2).unwrap(); + let id4 = id1.add_to_counter(3).unwrap(); + queue.add_if_condition_satisfied(id1, 2); + queue.add_if_condition_not_satisfied(id2, 3); + verify_cold_queue_internal_consistency(&queue, 2); + + // When: ingesting unknowns and taking one by a caller-side condition. + queue.ingest(id3, 4); + queue.ingest(id4, 5); + verify_cold_queue_internal_consistency(&queue, 4); + let taken = queue.take_first_matching(&1); + + // Then: the matching element is removed, the barriers stay valid, and + // a regular take still serves the ready prefix. + assert_eq!(taken, Some((id4, 5))); + verify_cold_queue_internal_consistency(&queue, 3); + let super::ColdQueueTakeResult::Taken(pair) = queue.take() else { + panic!("expected the ready asset to be taken"); + }; + assert_eq!(pair, (id1, 2)); + verify_cold_queue_internal_consistency(&queue, 2); + } + + #[tokio::test] + #[expect(non_snake_case)] + async fn double_queue__should_keep_take_and_discard_working_across_matching_takes() { + // Given: even values satisfy the stored condition (condition value 0). + let clock = FakeClock::default(); + let queue = DoubleQueue::new(clock.clock(), |cond, val| val % 2 == *cond, Arc::new(|| 0)); + let id1 = UniqueId::new(ParticipantId::from_raw(42), 123, 456); + let id2 = id1.add_to_counter(1).unwrap(); + let id3 = id1.add_to_counter(2).unwrap(); + let id4 = id1.add_to_counter(3).unwrap(); + queue.add_owned(id1, 2); + queue.add_owned(id2, 3); + queue.add_owned(id3, 4); + queue.add_owned(id4, 5); + + // When: a matching take (odd values) interleaves with the regular flow. + assert_eq!(queue.take_owned_matching(1).now_or_never(), Some((id2, 3))); + + // Then: a regular take still yields a condition-satisfying asset, + assert_eq!(queue.take_owned().now_or_never(), Some((id1, 2))); + // discard removes exactly the non-satisfying leftover, + assert_eq!(queue.maybe_discard_owned(2).await, vec![id4]); + assert_eq!(queue.available(), 1); + // and the surviving satisfying asset remains takeable. + assert_eq!(queue.take_owned().now_or_never(), Some((id3, 4))); + } + + #[tokio::test] + #[expect(non_snake_case)] + async fn distributed_store_take_owned_matching__should_delete_taken_asset_from_disk() { + // Given: two persisted assets; the condition keys on the asset value. + let dir = tempfile::tempdir().unwrap(); + let db = crate::db::SecretDB::new(dir.path(), [1; 16]).unwrap(); + let condition: fn(&Vec, &u32) -> bool = + |eligible, val| eligible.contains(&ParticipantId::from_raw(*val)); + let store = DistributedAssetStorage::::new( + FakeClock::default().clock(), + db.clone(), + crate::db::DBCol::TripleV2, + Vec::new(), + ParticipantId::from_raw(42), + condition, + Arc::new(std::vec::Vec::new), + ) + .unwrap(); + let id1 = store.generate_and_reserve_id(); + let id2 = store.generate_and_reserve_id(); + store.add_owned(id1, 123); + store.add_owned(id2, 456); + + // When: taking the matching asset, then reopening the store from disk. + let taken = store + .take_owned_matching(vec![ParticipantId::from_raw(456)]) + .await; + drop(store); + let reopened = DistributedAssetStorage::::new( + FakeClock::default().clock(), + db, + crate::db::DBCol::TripleV2, + Vec::new(), + ParticipantId::from_raw(42), + condition, + Arc::new(std::vec::Vec::new), + ) + .unwrap(); + + // Then: only the taken asset was removed from disk. + assert_eq!(taken, (id2, 456)); + assert_eq!(reopened.num_owned(), 1); + assert_eq!( + reopened + .take_owned_matching(vec![ParticipantId::from_raw(123)]) + .await, + (id1, 123) + ); + } + // This test covers tricky cases around updates to the condition value #[test] fn test_double_queue_condition_value() { diff --git a/crates/node/src/metrics.rs b/crates/node/src/metrics.rs index 373689ac1..a4be5cae4 100644 --- a/crates/node/src/metrics.rs +++ b/crates/node/src/metrics.rs @@ -157,8 +157,8 @@ pub static MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS: LazyLock< prometheus::register_int_counter!( "mpc_num_verify_foreign_tx_unavailable_chain_rejections", "Number of gate rejections of verify foreign tx attempts, at most one per node per \ - attempt: the requested chain is not available or the supporters snapshot has not \ - been received yet" + attempt: the requested chain is not available, the supporters snapshot has not \ + been received yet, or too few supporting participants are alive" ) .unwrap() }); diff --git a/crates/node/src/providers/ecdsa.rs b/crates/node/src/providers/ecdsa.rs index 24120e8c7..01e522092 100644 --- a/crates/node/src/providers/ecdsa.rs +++ b/crates/node/src/providers/ecdsa.rs @@ -87,6 +87,10 @@ impl EcdsaSignatureProvider { }) } + pub(crate) fn all_alive_participant_ids(&self) -> Vec { + self.client.all_alive_participant_ids() + } + pub(super) fn keyshare(&self, domain_id: DomainId) -> anyhow::Result { ecdsa_common::lookup_keyshare(&self.keyshares, domain_id) } diff --git a/crates/node/src/providers/verify_foreign_tx/sign.rs b/crates/node/src/providers/verify_foreign_tx/sign.rs index 5004b1539..df5adbe28 100644 --- a/crates/node/src/providers/verify_foreign_tx/sign.rs +++ b/crates/node/src/providers/verify_foreign_tx/sign.rs @@ -15,6 +15,7 @@ use tokio_util::time::FutureExt; use crate::foreign_chain_policy::SupportersByForeignChain; use crate::metrics; +use crate::primitives::ParticipantId; use crate::providers::verify_foreign_tx::VerifyForeignTxTaskId; use crate::types::{SignatureRequest, VerifyForeignTxRequest}; use crate::{ @@ -24,6 +25,7 @@ use crate::{ use near_mpc_bounded_collections::BoundedVec; use near_mpc_contract_interface::types::{self as dtos, ECDSA_PAYLOAD_SIZE_BYTES}; use near_mpc_contract_interface::types::{Payload, Tweak}; +use std::collections::HashSet; use tokio::time::{Duration, timeout}; const FOREIGN_CHAIN_INSPECTION_TIMEOUT: Duration = Duration::from_secs(5); @@ -54,21 +56,44 @@ impl VerifyForeignTxProvider { id: SignatureId, ) -> anyhow::Result<((dtos::ForeignTxSignPayload, Signature), VerifyingKey)> { let foreign_tx_request = self.verify_foreign_tx_request_store.get(id).await?; + let requested_chain = foreign_tx_request.request.chain(); // Also checked in `execute_foreign_chain_request`; checked early here - // because `take_owned` below irreversibly consumes a presignature. An - // availability flip between the two checks still costs one presignature. - ensure_chain_is_available( - &self.supporters_by_foreign_chain.borrow(), - &foreign_tx_request.request, - ) - .inspect_err(|_| metrics::MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS.inc())?; + // so an unavailable chain is rejected up front. + let chain_supporters: HashSet = { + let snapshot = self.supporters_by_foreign_chain.borrow(); + ensure_chain_is_available(&snapshot, &foreign_tx_request.request).inspect_err( + |_| metrics::MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS.inc(), + )?; + snapshot + .as_ref() + .and_then(|supporters| supporters.get(&requested_chain)) + .cloned() + .unwrap_or_default() + }; let keyshare = self .ecdsa_signature_provider .keyshare(foreign_tx_request.domain_id)?; - let (presignature_id, presignature) = keyshare.presignature_store.take_owned().await; - let participants = presignature.participants.clone(); + let eligible_participants: Vec = self + .ecdsa_signature_provider + .all_alive_participant_ids() + .into_iter() + .filter(|participant_id| chain_supporters.contains(participant_id)) + .collect(); + ensure_enough_alive_supporters( + requested_chain, + eligible_participants.len(), + keyshare.reconstruction_threshold.inner(), + ) + .inspect_err(|_| metrics::MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS.inc())?; + + let (presignature_id, presignature) = keyshare + .presignature_store + .take_owned_matching(eligible_participants) + .await; + let participants: Vec = presignature.participants.clone(); + let channel = self.ecdsa_signature_provider.new_channel_for_task( VerifyForeignTxTaskId::VerifyForeignTx { id, @@ -389,6 +414,29 @@ enum ChainAvailabilityError { "requested chain {requested:?} is not in the list of available foreign chains on the MPC contract" )] ChainNotAvailable { requested: dtos::ForeignChain }, + #[error( + "requested chain {requested:?} has {alive} alive supporting participants, {required} required" + )] + NotEnoughAliveSupporters { + requested: dtos::ForeignChain, + alive: usize, + required: u64, + }, +} + +fn ensure_enough_alive_supporters( + requested: dtos::ForeignChain, + alive: usize, + required: u64, +) -> Result<(), ChainAvailabilityError> { + if (alive as u64) < required { + return Err(ChainAvailabilityError::NotEnoughAliveSupporters { + requested, + alive, + required, + }); + } + Ok(()) } /// A chain counts as available when the supporters map has an entry for it: @@ -414,9 +462,8 @@ fn ensure_chain_is_available( #[expect(non_snake_case)] mod tests { use super::*; - use crate::primitives::ParticipantId; use assert_matches::assert_matches; - use std::collections::{BTreeMap, HashSet}; + use std::collections::BTreeMap; fn bitcoin_supporters() -> Option { Some(BTreeMap::from([( @@ -475,4 +522,24 @@ mod tests { Err(ChainAvailabilityError::SupportersSnapshotNotReady) ); } + + #[test] + fn ensure_enough_alive_supporters__should_fail_below_reconstruction_threshold() { + assert_matches!( + ensure_enough_alive_supporters(dtos::ForeignChain::Bitcoin, 2, 3), + Err(ChainAvailabilityError::NotEnoughAliveSupporters { + requested: dtos::ForeignChain::Bitcoin, + alive: 2, + required: 3, + }) + ); + } + + #[test] + fn ensure_enough_alive_supporters__should_succeed_at_reconstruction_threshold() { + assert_matches!( + ensure_enough_alive_supporters(dtos::ForeignChain::Bitcoin, 3, 3), + Ok(()) + ); + } }