diff --git a/crates/node/src/assets.rs b/crates/node/src/assets.rs index 1062cd6c5..6c42188e2 100644 --- a/crates/node/src/assets.rs +++ b/crates/node/src/assets.rs @@ -6,7 +6,6 @@ use crate::db::{DBCol, SecretDB, SecretDBUpdate}; use crate::primitives::{ParticipantId, UniqueId}; use crate::providers::HasParticipants; use borsh::BorshDeserialize; -use futures::FutureExt; use near_time::Clock; use serde::Serialize; use serde::de::DeserializeOwned; @@ -31,6 +30,9 @@ use std::sync::{Arc, Mutex}; /// If the element *doesn't satisfy* the condition it is inserted at the back. /// 4. When the condition changes the barriers are reset, marking /// the entire queue as unknown. +/// 5. When taking an asset matching a caller-supplied condition value we may +/// remove from any position before the cold_available barrier. Barriers +/// past the removed position shift down by one. /// /// NB: Assets may be reordered by these operations. No guarantees are made on the order in which /// assets are taken or discarded from the queue. @@ -165,6 +167,42 @@ impl ColdQueue { self.cold_queue.push_back((id, value)); ColdQueueAddIfNotSatisfiedResult::Enqueued } + + /// Adds an element to the cold queue unconditionally. + pub(self) fn ingest(&mut self, id: UniqueId, value: T) { + self.update_condition_value_if_due(); + if (self.condition)(&self.last_condition_value, &value) { + self.cold_queue.push_front((id, value)); + self.cold_ready += 1; + self.cold_available += 1; + } else { + self.cold_queue.push_back((id, value)); + } + } + + /// Removes and returns the first element satisfying both the standing + /// condition and the caller-supplied `cond_val`, shifting the barriers + /// ([`ColdQueue::cold_ready`] and [`ColdQueue::cold_available`]) that lie + /// past the removed position. + pub(self) fn take_first_matching(&mut self, cond_val: &CondVal) -> Option<(UniqueId, T)> { + self.update_condition_value_if_due(); + let pos = self + .cold_queue + .iter() + .take(self.cold_available) + .position(|(_, val)| self.satisfies_condition(cond_val, val))?; + if pos < self.cold_ready { + self.cold_ready -= 1; + } + if pos < self.cold_available { + self.cold_available -= 1; + } + self.cold_queue.remove(pos) + } + + pub(self) fn satisfies_condition(&self, cond_val: &CondVal, val: &T) -> bool { + (self.condition)(&self.last_condition_value, val) && (self.condition)(cond_val, val) + } } enum ColdQueueTakeResult { @@ -197,6 +235,7 @@ where hot_receiver: flume::Receiver<(UniqueId, T)>, cold_queue: Arc>>, clock: Clock, + cold_queue_new_elements: tokio::sync::Notify, } impl DoubleQueue @@ -218,6 +257,7 @@ where condition_value_fetcher, ))), clock, + cold_queue_new_elements: tokio::sync::Notify::new(), } } @@ -233,6 +273,7 @@ where // away and we quickly exhaust the available assets. self.cold_queue.lock().unwrap().update_condition_value(); loop { + let cold_queue_new_elements = self.cold_queue_new_elements.notified(); let taken = self.cold_queue.lock().unwrap().take(); match taken { ColdQueueTakeResult::Taken(result) => { @@ -252,13 +293,17 @@ where // making a cold queue element eligible. continue; } + _ = cold_queue_new_elements => { + continue; + } received = self.hot_receiver.recv_async() => { - // can't fail, because self keeps a sender. - let (id, value) = received.unwrap(); + let (id, value) = received.expect("should never fail because self keeps a sender"); match self.cold_queue.lock().unwrap().add_if_condition_not_satisfied(id, value) { ColdQueueAddIfNotSatisfiedResult::ConditionSatisfied(value) => { return (id, value); } + // Element failed the queue condition (aliveness) + // so it would not be suitable for any taker. Hence, no notify. ColdQueueAddIfNotSatisfiedResult::Enqueued => { continue; } @@ -270,6 +315,51 @@ where } } + pub async fn take_owned_matching(&self, cond_val: CondVal) -> (UniqueId, T) { + loop { + let cold_queue_new_elements = self.cold_queue_new_elements.notified(); + let (taken, ingested) = { + let mut cold = self.cold_queue.lock().unwrap(); + let mut ingested = false; + let mut taken = None; + while let Ok((id, value)) = self.hot_receiver.try_recv() { + if cold.satisfies_condition(&cond_val, &value) { + taken = Some((id, value)); + break; + } + cold.ingest(id, value); + ingested = true; + } + ( + taken.or_else(|| cold.take_first_matching(&cond_val)), + ingested, + ) + }; + + if ingested { + self.cold_queue_new_elements.notify_waiters(); + } + if let Some(taken) = taken { + return taken; + } + + // If the cold queue is exhausted, wait for a new element. + tokio::select! { + _ = self.clock.sleep(near_time::Duration::seconds(1)) => { + continue; + } + _ = cold_queue_new_elements => { + continue; + } + received = self.hot_receiver.recv_async() => { + let (id, value) = received.expect("should never fail because self keeps a sender"); + self.cold_queue.lock().unwrap().ingest(id, value); + self.cold_queue_new_elements.notify_waiters(); + } + } + } + } + /// 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 { @@ -298,8 +388,8 @@ where // If the cold queue is exhausted, process elements buffered in the hot queue while num_elements_to_process > 0 { - match self.hot_receiver.recv_async().now_or_never() { - Some(Ok((id, value))) => { + match self.hot_receiver.try_recv().ok() { + Some((id, value)) => { num_elements_to_process -= 1; let _ = self .cold_queue @@ -569,6 +659,22 @@ where result } + /// Takes an owned asset satisfying both the standing alive-condition and + /// the supplied `eligible` set. Blocks indefinitely if none becomes + /// available. + /// Callers are expected to enforce their own timeout. + pub async fn take_owned_matching(&self, eligible: Vec) -> (UniqueId, T) { + let (id, val) = self.owned_queue.take_owned_matching(eligible).await; + let mut update = self.db.update(); + update.delete(self.col, &self.make_key(id)); + update + .commit() + // TODO(#4090): propagate err instead in here and rest of the functions + // in this file. + .expect("Unrecoverable error writing to database"); + (id, val) + } + fn take_unowned_inner(&self, id: UniqueId) -> anyhow::Result { let key = self.make_key(id); let value_ser = self.db.get(self.col, &key)?.ok_or_else(|| { @@ -1398,4 +1504,230 @@ mod tests { } } } + + // The standing condition holds for 2 and 3, the supplied one for 3 and 4: + // only 3 satisfies both and is taken. The drain stops at the match, so 2 + // is parked in the cold queue while 4 stays buffered in the hot queue + // (counted by `available`, invisible to `offline`). + #[test] + #[expect(non_snake_case)] + fn take_owned_matching__should_only_take_asset_satisfying_both_conditions() { + // Given + let clock = FakeClock::default(); + let queue = DoubleQueue::new( + clock.clock(), + |cond: &Vec, val| cond.contains(val), + Arc::new(|| vec![2, 3]), + ); + 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(); + queue.add_owned(id1, 2); + queue.add_owned(id2, 3); + queue.add_owned(id3, 4); + + // When + let taken = queue.take_owned_matching(vec![3, 4]).now_or_never(); + + // Then + assert_eq!(taken, Some((id2, 3))); + assert_eq!(queue.available(), 2); + assert_eq!(queue.offline(), 0); + } + + // A take with a supplied value nothing matches yet parks; it completes once + // a matching asset is added, without consuming the non-matching one. + #[test] + #[expect(non_snake_case)] + fn take_owned_matching__should_wait_until_matching_asset_is_added() { + // Given + let clock = FakeClock::default(); + let queue = DoubleQueue::new( + clock.clock(), + |cond: &Vec, val| cond.contains(val), + Arc::new(|| vec![2, 3]), + ); + let id1 = UniqueId::new(ParticipantId::from_raw(42), 123, 456); + let id2 = id1.add_to_counter(1).unwrap(); + queue.add_owned(id1, 2); + + // When + let fut = queue.take_owned_matching(vec![3]); + let MaybeReady::Future(fut) = run_future_once(fut) else { + panic!("should not take a value when no element matches"); + }; + + // Then + queue.add_owned(id2, 3); + assert_eq!(fut.now_or_never().unwrap(), (id2, 3)); + assert_eq!(queue.available(), 1); + } + + // Takes from the middle and the front of the ready section, then attempts an + // element failing the standing condition (never returned even when it + // satisfies the supplied value), checking barrier consistency at every step. + #[test] + #[expect(non_snake_case)] + fn take_first_matching__should_maintain_barrier_invariants() { + // Given + let clock = FakeClock::default(); + let mut queue = ColdQueue::new( + clock.clock(), + |cond: &Vec, val| cond.contains(val), + Arc::new(|| vec![2, 4]), + ); + 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(); + queue.ingest(id1, 2); + queue.ingest(id2, 4); + queue.ingest(id3, 3); + verify_cold_queue_internal_consistency(&queue, 3); + + // When + let taken = queue.take_first_matching(&vec![2, 3]); + + // Then + assert_eq!(taken, Some((id1, 2))); + verify_cold_queue_internal_consistency(&queue, 2); + + assert_eq!(queue.take_first_matching(&vec![4]), Some((id2, 4))); + verify_cold_queue_internal_consistency(&queue, 1); + + assert_eq!(queue.take_first_matching(&vec![3]), None); + verify_cold_queue_internal_consistency(&queue, 1); + } + + // Flips the standing condition between operations (advancing the fake + // clock past the refresh interval): the barrier reset must keep the + // sections consistent, and takes must honor the new standing value. + #[test] + #[expect(non_snake_case)] + fn take_first_matching__should_stay_consistent_when_condition_value_changes() { + // Given + let clock = FakeClock::default(); + let standing = Arc::new(Mutex::new(vec![2, 4])); + let mut queue = ColdQueue::new(clock.clock(), |cond: &Vec, val| cond.contains(val), { + let standing = standing.clone(); + Arc::new(move || standing.lock().unwrap().clone()) + }); + let id1 = UniqueId::new(ParticipantId::from_raw(42), 1, 0); + let id2 = id1.add_to_counter(1).unwrap(); + queue.ingest(id1, 2); + queue.ingest(id2, 3); + verify_cold_queue_internal_consistency(&queue, 2); + + // When: the standing condition changes and the refresh comes due. + *standing.lock().unwrap() = vec![3]; + clock.advance(near_time::Duration::seconds(1)); + + // Then: 2 no longer satisfies the standing condition and cannot be + // taken, while 3 (previously non-satisfying) now can. + assert_eq!(queue.take_first_matching(&vec![2]), None); + verify_cold_queue_internal_consistency(&queue, 2); + assert_eq!(queue.take_first_matching(&vec![3]), Some((id2, 3))); + verify_cold_queue_internal_consistency(&queue, 1); + } + + // Takes the asset matching the supplied participant set, then reopens the + // store from the same DB: only the taken asset is deleted from disk. + #[tokio::test] + #[expect(non_snake_case)] + async fn distributed_store_take_owned_matching__should_delete_taken_asset_from_disk() { + // Given + 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 alive = || vec![ParticipantId::from_raw(123), ParticipantId::from_raw(456)]; + let new_store = |db: Arc| { + DistributedAssetStorage::::new( + FakeClock::default().clock(), + db, + crate::db::DBCol::TripleV2, + Vec::new(), + ParticipantId::from_raw(42), + condition, + Arc::new(alive), + ) + .unwrap() + }; + let store = new_store(db.clone()); + 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 + let taken = store + .take_owned_matching(vec![ParticipantId::from_raw(456)]) + .await; + drop(store); + let reopened = new_store(db); + + // Then + 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) + ); + } + + /// A `take_owned_matching` taker drains a buffered asset it can't use + /// (even fails its odd condition) into the cold queue on its first scan; a + /// parked `take_owned` taker must be woken for it immediately — the clock + /// never advances, so a missing wakeup leaves it pending on its 1s tick. + #[test] + #[expect(non_snake_case)] + fn take_owned__should_wake_when_matching_take_drains_incompatible_asset() { + // Given + let clock = FakeClock::default(); + let queue = DoubleQueue::new(clock.clock(), |cond, val| val % 2 == *cond, Arc::new(|| 0)); + let MaybeReady::Future(take_owned) = run_future_once(queue.take_owned()) else { + panic!("take_owned should park on an empty queue"); + }; + let id = UniqueId::new(ParticipantId::from_raw(42), 123, 456); + queue.add_owned(id, 2); + + // When + let MaybeReady::Future(_take_matching) = run_future_once(queue.take_owned_matching(1)) + else { + panic!("the asset must not satisfy the matching taker's condition"); + }; + + // Then + assert_eq!(take_owned.now_or_never().unwrap(), (id, 2)); + } + + /// Same as above, but the asset arrives while the matching taker is + /// already parked, so it is ingested by the taker's hot-receiver select + /// arm rather than by the drain on the first scan; the parked `take_owned` + /// taker must be woken all the same. + #[test] + #[expect(non_snake_case)] + fn take_owned__should_wake_when_parked_matching_take_receives_incompatible_asset() { + // Given + let clock = FakeClock::default(); + let queue = DoubleQueue::new(clock.clock(), |cond, val| val % 2 == *cond, Arc::new(|| 0)); + let MaybeReady::Future(take_owned) = run_future_once(queue.take_owned()) else { + panic!("take_owned should park on an empty queue"); + }; + let MaybeReady::Future(take_matching) = run_future_once(queue.take_owned_matching(1)) + else { + panic!("take_owned_matching should park on an empty queue"); + }; + + // When + let id = UniqueId::new(ParticipantId::from_raw(42), 123, 456); + queue.add_owned(id, 2); + let MaybeReady::Future(_take_matching) = run_future_once(take_matching) else { + panic!("the asset must not satisfy the matching taker's condition"); + }; + + // Then + assert_eq!(take_owned.now_or_never().unwrap(), (id, 2)); + } } diff --git a/crates/node/src/metrics.rs b/crates/node/src/metrics.rs index 047eef3db..506e3273d 100644 --- a/crates/node/src/metrics.rs +++ b/crates/node/src/metrics.rs @@ -162,6 +162,16 @@ pub static MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS: LazyLock< .unwrap() }); +pub static MPC_NUM_VERIFY_FOREIGN_TX_PRESIGNATURE_WAITS: LazyLock = + LazyLock::new(|| { + prometheus::register_int_counter!( + "mpc_num_verify_foreign_tx_presignature_waits", + "Number of verify foreign tx attempts that found no chain-compatible presignature \ + immediately and had to wait for one" + ) + .unwrap() + }); + pub static MPC_NUM_SIGN_RESPONSES_INDEXED: LazyLock = LazyLock::new(|| { prometheus::register_int_counter!( "mpc_num_signature_responses_indexed", diff --git a/crates/node/src/providers/ecdsa.rs b/crates/node/src/providers/ecdsa.rs index ac20f5717..92ead1efd 100644 --- a/crates/node/src/providers/ecdsa.rs +++ b/crates/node/src/providers/ecdsa.rs @@ -121,6 +121,10 @@ impl EcdsaSignatureProvider { ) -> anyhow::Result { self.client.new_channel_for_task(task_id, participants) } + + pub(super) fn my_participant_id(&self) -> ParticipantId { + self.client.my_participant_id() + } } #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, BorshSerialize, BorshDeserialize)] diff --git a/crates/node/src/providers/verify_foreign_tx/sign.rs b/crates/node/src/providers/verify_foreign_tx/sign.rs index 1604bbc4e..f0a51d8aa 100644 --- a/crates/node/src/providers/verify_foreign_tx/sign.rs +++ b/crates/node/src/providers/verify_foreign_tx/sign.rs @@ -1,3 +1,5 @@ +use std::collections::HashSet; + use anyhow::{Context, bail}; use foreign_chain_inspector::abstract_chain::inspector::AbstractExtractor; use foreign_chain_inspector::adi::inspector::AdiExtractor; @@ -17,6 +19,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::{ @@ -29,6 +32,7 @@ use near_mpc_contract_interface::types::{Payload, Tweak}; use tokio::time::{Duration, timeout}; const FOREIGN_CHAIN_INSPECTION_TIMEOUT: Duration = Duration::from_secs(5); +const PRESIGNATURE_TAKE_GRACE_PERIOD: Duration = Duration::from_secs(1); fn build_signature_request( request: &VerifyForeignTxRequest, @@ -57,21 +61,46 @@ fn build_signature_request( }) } +// Awaits on the future for specified grace duration, and calls on_slow +// if grace period expires. +async fn await_with_slow_hook( + grace: Duration, + fut: F, + on_slow: impl FnOnce(), +) -> F::Output { + tokio::pin!(fut); + match timeout(grace, &mut fut).await { + Ok(output) => output, + Err(_) => { + on_slow(); + fut.await + } + } +} + impl VerifyForeignTxProvider { pub(crate) async fn make_verify_foreign_tx_leader( &self, 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(); + + let chain_supporters: HashSet = { + let snapshot = self.supporters_by_foreign_chain.borrow().clone(); + ensure_chain_is_available(&snapshot, foreign_tx_request.request.chain()).inspect_err( + |_| metrics::MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS.inc(), + )?; + snapshot.get(&requested_chain).cloned().unwrap_or_default() + }; - // 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())?; + // If we don't support the foreign chain, we can't lead the computation. + // TODO(#3961): narrow leader selection to only chain supporters. + let my_participant_id = self.ecdsa_signature_provider.my_participant_id(); + if !chain_supporters.contains(&my_participant_id) { + metrics::MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS.inc(); + anyhow::bail!("this node does not support the requested chain {requested_chain:?}"); + } let response_payload = self .execute_foreign_chain_request( @@ -87,7 +116,20 @@ impl VerifyForeignTxProvider { let keyshare = self .ecdsa_signature_provider .keyshare(foreign_tx_request.domain_id)?; - let (presignature_id, presignature) = keyshare.presignature_store.take_owned().await; + let (presignature_id, presignature) = await_with_slow_hook( + PRESIGNATURE_TAKE_GRACE_PERIOD, + keyshare + .presignature_store + .take_owned_matching(chain_supporters.iter().copied().collect()), + || { + metrics::MPC_NUM_VERIFY_FOREIGN_TX_PRESIGNATURE_WAITS.inc(); + tracing::warn!( + ?requested_chain, + "no chain-compatible presignatures available, waiting" + ) + }, + ) + .await; let participants = presignature.participants.clone(); let channel = self.ecdsa_signature_provider.new_channel_for_task( VerifyForeignTxTaskId::VerifyForeignTx { @@ -137,7 +179,9 @@ impl VerifyForeignTxProvider { request: &dtos::ForeignChainRpcRequest, payload_version: dtos::ForeignTxPayloadVersion, ) -> anyhow::Result { - ensure_chain_is_available(&self.supporters_by_foreign_chain.borrow(), request) + // Check that the requested chain is still available when this + // point is reached. + ensure_chain_is_available(&self.supporters_by_foreign_chain.borrow(), request.chain()) .inspect_err(|_| { metrics::MPC_NUM_VERIFY_FOREIGN_TX_UNAVAILABLE_CHAIN_REJECTIONS.inc() })?; @@ -449,13 +493,14 @@ struct ChainNotAvailableError { /// participants supports it. fn ensure_chain_is_available( supporters_by_foreign_chain: &SupportersByForeignChain, - request: &dtos::ForeignChainRpcRequest, + foreign_chain: dtos::ForeignChain, ) -> Result<(), ChainNotAvailableError> { - let requested = request.chain(); - if supporters_by_foreign_chain.contains_key(&requested) { + if supporters_by_foreign_chain.contains_key(&foreign_chain) { Ok(()) } else { - Err(ChainNotAvailableError { requested }) + Err(ChainNotAvailableError { + requested: foreign_chain, + }) } } @@ -510,7 +555,7 @@ mod tests { // When, then assert_matches!( - ensure_chain_is_available(&supporters, &bitcoin_request()), + ensure_chain_is_available(&supporters, bitcoin_request().chain()), Ok(_) ); } @@ -527,7 +572,7 @@ mod tests { // When, then assert_matches!( - ensure_chain_is_available(&supporters, ðereum_request), + ensure_chain_is_available(&supporters, ethereum_request.chain()), Err(ChainNotAvailableError { requested: dtos::ForeignChain::Ethereum }) diff --git a/docs/asset-generation.md b/docs/asset-generation.md index f1e6ebe04..ff27b30ca 100644 --- a/docs/asset-generation.md +++ b/docs/asset-generation.md @@ -74,6 +74,7 @@ operations: |--------|-------------| | `add_owned(id, value)` | Stores a newly generated asset as owned. Persists to `RocksDB` and pushes to the in-memory queue. | | `take_owned()` | Blocks until an online asset is available, removes it from storage, and returns it. | +| `take_owned_matching(eligible)` | Like `take_owned()`, but only returns an asset whose borrowers are additionally all in `eligible`. Used by the verify-foreign-tx leader with the requested chain's supporters. | | `maybe_discard_owned(n)` | Examines up to `n` assets. Discards those with offline borrowers; keeps online ones aside as ready. | | `add_unowned(id, value)` | Stores another owner's asset share in `RocksDB`. | | `take_unowned(id)` | Looks up an unowned asset by ID, removes it from `RocksDB`, and returns it. Fails if not found. | @@ -85,7 +86,8 @@ operations: ### Properties of an asset taken from the queue -When `take_owned()` returns an asset, the following holds: +When `take_owned()` or `take_owned_matching(eligible)` returns an asset, +the following holds: 1. **All borrowers were online at check time.** The queue verifies that the asset is online by the time it is taken from the queue @@ -94,9 +96,13 @@ When `take_owned()` returns an asset, the following holds: 2. **The asset has not been used before.** It is removed from storage atomically on retrieval. Each asset is consumed exactly once. -`take_unowned(id)` does not guarantee either property — it performs a -plain database lookup with no liveness check. Borrowers trust the -owner's choice. +3. **All borrowers are in `eligible`** — `take_owned_matching` only. + The asset's borrowers are checked against the caller-supplied set at + the same time as the liveness classification. + +`take_unowned(id)` does not guarantee any of these properties — it +performs a plain database lookup with no liveness check. Borrowers trust +the owner's choice. **Remark: No real-time liveness guarantee.** A participant can go offline between the liveness check and the start of the protocol. If that @@ -142,8 +148,9 @@ The `DoubleQueue` that holds owned assets has two layers: ### Hot queue An unbounded multi-producer multi-consumer (MPMC) channel. Newly generated assets are pushed here by -`add_owned()`. The hot queue is drained into the cold queue the first -time an asset is needed. +`add_owned()`. Takers move buffered hot-queue assets into the cold queue +while searching for a usable asset; a take stops as soon as it finds one, +so assets may remain buffered in the hot queue. ### Cold queue @@ -174,6 +181,13 @@ will never return an offline asset — if all owned assets are offline, it blocks and re-checks the set of online participants every second until an asset comes back online. +### take_first_matching() + +Backing `take_owned_matching`, this is the exception to the cold queue's +front/back access: it removes the first online asset also matching the +caller's set from *any* position before `cold_available`, shifting the +barriers past the removal point down by one. + ### take_owned() flow 1. Force-refresh the condition value. @@ -245,7 +259,9 @@ Unlike triples and presignatures, signatures are not pre-generated. When a signature request arrives: 1. **Leader** calls `presignature_store.take_owned()` for the relevant - domain, consuming one presignature. + domain, consuming one presignature. Verify-foreign-tx leaders call + `take_owned_matching(supporters)` instead, so the borrowers can all + inspect the requested chain. 2. Leader opens a network channel with the presignature's borrowers and broadcasts the presignature ID along with the signature request. 3. **Followers** call `presignature_store.take_unowned(id)` to retrieve diff --git a/docs/design/calculating-supported-foreign-chains.md b/docs/design/calculating-supported-foreign-chains.md index ff01044c3..9fd7f1ec9 100644 --- a/docs/design/calculating-supported-foreign-chains.md +++ b/docs/design/calculating-supported-foreign-chains.md @@ -66,14 +66,28 @@ result as non-retryable. (Open: whether a sub-quorum from purely *transient* failures — timeouts, finality not reached — should still retry, vs. only genuine disagreement being terminal. Tracked in [#3477](https://github.com/near/mpc/issues/3477).) -## Participant election +## Participant selection -Foreign-tx signing must elect participants that **cover** the requested chain +Foreign-tx signing must select participants that **cover** the requested chain (report ≥ `rpc_quorum(C)` providers for `C`), not merely online ones — a non-covering participant produces no share and can stall the request. -Implementation requirement, not current behavior: today the signing set is inherited -from a presignature, whose -participants were chosen for liveness, not chain coverage. + +Implemented on the presignature-selection side: the leader only takes a +presignature whose participants are all alive **and** supporters of `C`, and +refuses to lead a request for a chain it does not itself support (every owned +presignature includes the leader). Leader selection itself is not yet +chain-aware — a non-supporting leader rejects the attempt instead of serving +it — tracked in [#3961](https://github.com/near/mpc/issues/3961). + +Residual limitation, accepted as-is: presignature generation remains +liveness-driven, so participant sets are random `t`-subsets of the alive set. +When not all alive participants support `C`, many presignatures are +incompatible, and a request may wait until a compatible one is generated or +the request times out. We do not plan chain-aware generation, the +`mpc_num_verify_foreign_tx_presignature_waits` metric (plus a warning log) +makes such waits observable. A broader redesign of asset selection and the +underlying queue — which could subsume this limitation — is tracked in +[#377](https://github.com/near/mpc/issues/377). ## Per-node registration @@ -109,13 +123,17 @@ chain leaves the available set only when more than `n − signing_threshold` nod it. This strictly improves on the intersection rule, where one non-registering node dropped a chain to zero availability. -## Known tradeoff - -A node that's up but not covering a chain only sidelines the `ForeignTx` **presignatures** it co-owns -(discarded if they stay offline long enough). Its **triples are not lost** — they're shared across -domains and stay in use, so triples go offline only if the node is genuinely down. Mitigation is -operational: alerting keeps coverage high and operators are expected to configure every node for -every chain. +## Known limitations + +A node that's up but not covering a chain `C` shrinks the eligible presignature +pool for `C`: presignatures it co-owns are excluded from selection for `C`'s +requests (they stay usable for chains the node does cover), and the smaller +the supporter set, the fewer generated presignatures qualify (see +[Participant selection](#participant-selection)). Its **triples are +not lost** — they're shared across domains and stay in use, so triples go +offline only if the node is genuinely down. Mitigation is operational: +alerting keeps coverage high and operators are expected to configure every +node for every chain. ## Migration