feat(node): chain-compatible presignature selection for verifying foreign transactions

crates/node/Cargo.toml

@@ -116,6 +116,7 @@ average = { workspace = true }
 blstrs = { workspace = true }
 chain-gateway = { workspace = true, features = ["test-utils"] }
 elliptic-curve = { workspace = true }
+httpmock = { workspace = true }
 insta = { workspace = true }
 itertools = { workspace = true }
 mockall = { workspace = true }

crates/node/src/assets.rs

@@ -31,6 +31,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 +168,39 @@ impl<T, CondVal: Default + Eq> ColdQueue<T, CondVal> {
         self.cold_queue.push_back((id, value));
         ColdQueueAddIfNotSatisfiedResult::Enqueued
     }
+
+    /// Adds an element to the cold queue unconditionally, never returned.
+    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
+    /// 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.condition)(&self.last_condition_value, val) && (self.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)
+    }
 }
 
 enum ColdQueueTakeResult<T> {
@@ -270,6 +306,31 @@ where
         }
     }
 
+    pub async fn take_owned_matching(&self, cond_val: CondVal) -> (UniqueId, T) {
+        loop {
+            {
+                let mut cold = self.cold_queue.lock().unwrap();
+                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;
+                }
+            }
+            // If the cold queue is exhausted, wait for a new element.
+            tokio::select! {
+                _ = self.clock.sleep(near_time::Duration::seconds(1)) => {
+                    continue;
+                }
+                received = self.hot_receiver.recv_async() => {
+                    // can't fail, because self keeps a sender.
+                    let (id, val) = received.unwrap();
+                    self.cold_queue.lock().unwrap().ingest(id, val);
+                }
+            }
+        }
+    }
+
     /// 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<UniqueId> {
@@ -569,6 +630,20 @@ 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<ParticipantId>) -> (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()
+            .expect("Unrecoverable error writing to database");
+        (id, val)
+    }
+
     fn take_unowned_inner(&self, id: UniqueId) -> anyhow::Result<T> {
         let key = self.make_key(id);
         let value_ser = self.db.get(self.col, &key)?.ok_or_else(|| {
@@ -1398,4 +1473,173 @@ mod tests {
             }
         }
     }
+
+    // The standing condition holds for 2 and 3, the supplied one for 3 and 4:
+    // only 3 satisfies both and may be taken; the rest stay in the queue.
+    #[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<i32>, 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(), 1);
+        assert_eq!(queue.offline(), 1);
+    }
+
+    // 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<i32>, 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<i32>, 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<i32>, 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<ParticipantId>, &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<crate::db::SecretDB>| {
+            DistributedAssetStorage::<u32>::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)
+        );
+    }
 }

crates/node/src/coordinator.rs

@@ -1,12 +1,17 @@
 use crate::assets::cleanup::{EpochData, delete_stale_triples_and_presignatures};
 use crate::config::{MpcConfig, ParticipantInfo, ParticipantsConfig, SecretsConfig};
 use crate::db::SecretDB;
+use crate::foreign_chain_policy::{
+    SupportersByForeignChain, foreign_tx_reconstruction_threshold,
+    spawn_supporters_by_foreign_chain,
+};
+use crate::indexer::foreign_chain::ForeignChainSupporters;
 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::indexer::{IndexerAPI, tx_sender};
 use crate::key_events::{
     ResharingArgs, keygen_follower, keygen_leader, resharing_follower, resharing_leader,
 };
@@ -55,7 +60,7 @@ use tracing::{error, info};
 /// 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<TransactionSender, ForeignChainPolicyReader> {
+pub struct Coordinator<TransactionSender> {
     pub clock: Clock,
     pub secrets: SecretsConfig,
     pub config_file: ConfigFile,
@@ -65,7 +70,7 @@ pub struct Coordinator<TransactionSender, ForeignChainPolicyReader> {
     /// Storage for keyshares.
     pub keyshare_storage: Arc<RwLock<KeyshareStorage>>,
     /// For interaction with the indexer.
-    pub indexer: IndexerAPI<TransactionSender, ForeignChainPolicyReader>,
+    pub indexer: IndexerAPI<TransactionSender>,
 
     /// For testing, to know what the current state is.
     pub currently_running_job_name: Arc<Mutex<String>>,
@@ -101,11 +106,9 @@ enum MpcJobResult {
     HaltUntilInterrupted,
 }
 
-impl<TransactionSender, ForeignChainPolicyReader>
-    Coordinator<TransactionSender, ForeignChainPolicyReader>
+impl<TransactionSender> Coordinator<TransactionSender>
 where
     TransactionSender: tx_sender::TransactionSender + 'static,
-    ForeignChainPolicyReader: ReadSupportedForeignChain + Clone + Send + Sync + 'static,
 {
     pub async fn run(mut self) -> anyhow::Result<()> {
         loop {
@@ -176,7 +179,7 @@ where
                                 self.keyshare_storage.clone(),
                                 running_state.clone(),
                                 self.indexer.txn_sender.clone(),
-                                self.indexer.foreign_chain_policy_reader.clone(),
+                                self.indexer.foreign_chain_supporters_receiver.clone(),
                                 self.indexer
                                     .block_update_receiver
                                     .clone()
@@ -371,7 +374,7 @@ where
         keyshare_storage: Arc<RwLock<KeyshareStorage>>,
         running_state: ContractRunningState,
         chain_txn_sender: TransactionSender,
-        foreign_chain_policy_reader: ForeignChainPolicyReader,
+        foreign_chain_supporters_receiver: watch::Receiver<ForeignChainSupporters>,
         block_update_receiver: tokio::sync::OwnedMutexGuard<
             mpsc::UnboundedReceiver<ChainBlockUpdate>,
         >,
@@ -694,9 +697,34 @@ where
                     ckd_keyshares,
                 ));
 
+                // `running_mpc_config.participants` is the running set retained
+                // to resharing survivors (active ∩ prospective), so a chain only
+                // counts as available when a quorum of nodes that can sign now
+                // and remain after the reshare supports it. With no ForeignTx
+                // domain nothing can be available, so the resolver isn't
+                // spawned and the provider sees a constant empty map.
+                let (supporters_by_foreign_chain, _supporters_resolver_task) =
+                    match foreign_tx_reconstruction_threshold(&running_state.domains) {
+                        Some(threshold) => {
+                            let (receiver, task) = spawn_supporters_by_foreign_chain(
+                                foreign_chain_supporters_receiver,
+                                running_mpc_config.participants.clone(),
+                                threshold,
+                            );
+                            (receiver, Some(task))
+                        }
+                        None => {
+                            // No resolver to feed it: the sender is dropped on
+                            // purpose and the provider sees a constant empty map.
+                            let (_sender, receiver) =
+                                watch::channel(SupportersByForeignChain::new());
+                            (receiver, None)
+                        }
+                    };
+
                 let verify_foreign_tx_provider = Arc::new(VerifyForeignTxProvider::new(
                     config_file.clone().into(),
-                    foreign_chain_policy_reader.clone(),
+                    supporters_by_foreign_chain,
                     verify_foreign_tx_request_store.clone(),
                     ecdsa_signature_provider.clone(),
                 )?);