Merge remote-tracking branch 'origin/main' into 2287-attestation-tasks-permanently-die-after-12h-retry-timeout

Author: pbeza

crates/node/src/coordinator.rs

@@ -703,8 +703,10 @@ where
                 // 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 foreign_tx_threshold =
+                    foreign_tx_reconstruction_threshold(&running_state.domains);
                 let (supporters_by_foreign_chain, _supporters_resolver_task) =
-                    match foreign_tx_reconstruction_threshold(&running_state.domains) {
+                    match foreign_tx_threshold {
                         Some(threshold) => {
                             let (receiver, task) = spawn_supporters_by_foreign_chain(
                                 foreign_chain_supporters_receiver,
@@ -725,6 +727,7 @@ where
                 let verify_foreign_tx_provider = Arc::new(VerifyForeignTxProvider::new(
                     config_file.clone().into(),
                     supporters_by_foreign_chain,
+                    foreign_tx_threshold,
                     verify_foreign_tx_request_store.clone(),
                     ecdsa_signature_provider.clone(),
                 )?);

crates/node/src/foreign_chain_policy.rs

@@ -9,7 +9,9 @@ use tokio::sync::watch;
 use crate::config::ParticipantsConfig;
 use crate::indexer::foreign_chain::ForeignChainSupporters;
 use crate::primitives::ParticipantId;
+use crate::requests::queue::RefineEligibleLeaders;
 use crate::tracking::{self, AutoAbortTask};
+use crate::types::VerifyForeignTxRequest;
 
 /// Participants supporting each available foreign chain; chains without a
 /// signing quorum are omitted.
@@ -129,6 +131,63 @@ fn resolve_participant_ids(
         .collect()
 }
 
+/// Narrows verify-foreign-tx leader selection to the participants supporting the
+/// request's chain, always reading the freshest supporters snapshot. A leader is
+/// only elected when a quorum of supporters is eligible, so an under-quorum
+/// request parks instead of burning attempts that cannot find a compatible
+/// presignature.
+pub(crate) struct ForeignChainLeadersRefiner {
+    supporters_receiver: watch::Receiver<SupportersByForeignChain>,
+    /// [`foreign_tx_reconstruction_threshold`] of the running domains, `None`
+    /// when there is no ForeignTx domain (the snapshot is then always empty).
+    quorum: Option<ReconstructionThreshold>,
+}
+
+impl ForeignChainLeadersRefiner {
+    pub(crate) fn new(
+        supporters_receiver: watch::Receiver<SupportersByForeignChain>,
+        quorum: Option<ReconstructionThreshold>,
+    ) -> Self {
+        ForeignChainLeadersRefiner {
+            supporters_receiver,
+            quorum,
+        }
+    }
+}
+
+impl RefineEligibleLeaders<VerifyForeignTxRequest> for ForeignChainLeadersRefiner {
+    fn refine(
+        &self,
+        request: &VerifyForeignTxRequest,
+        eligible: &HashSet<ParticipantId>,
+    ) -> HashSet<ParticipantId> {
+        let refined = match self
+            .supporters_receiver
+            .borrow()
+            .get(&request.request.chain())
+        {
+            Some(supporters) => supporters & eligible,
+            None => return HashSet::new(),
+        };
+        match self.quorum {
+            None => {
+                tracing::error!(
+                    chain = ?request.request.chain(),
+                    "chain has supporters but there is no ForeignTx domain, this should never happen"
+                );
+                refined
+            }
+            Some(quorum) => {
+                if u64::try_from(refined.len()).is_ok_and(|count| count >= quorum.inner()) {
+                    refined
+                } else {
+                    HashSet::new()
+                }
+            }
+        }
+    }
+}
+
 #[cfg(test)]
 #[expect(non_snake_case)]
 mod tests {
@@ -370,4 +429,131 @@ mod tests {
         });
         root.await;
     }
+
+    fn bitcoin_verify_foreign_tx_request() -> VerifyForeignTxRequest {
+        VerifyForeignTxRequest {
+            id: near_indexer_primitives::CryptoHash([1; 32]),
+            receipt_id: near_indexer_primitives::CryptoHash([2; 32]),
+            request: dtos::ForeignChainRpcRequest::Bitcoin(dtos::BitcoinRpcRequest {
+                tx_id: dtos::BitcoinTxId([3; 32]),
+                confirmations: 2.into(),
+                extractors: vec![dtos::BitcoinExtractor::BlockHash],
+            }),
+            payload_version: dtos::ForeignTxPayloadVersion::V1,
+            expected_payload_hash: None,
+            entropy: [4; 32],
+            timestamp_nanosec: 0,
+            domain_id: mpc_primitives::domain::DomainId(0),
+        }
+    }
+
+    fn participant_set(ids: &[u32]) -> HashSet<ParticipantId> {
+        ids.iter().copied().map(ParticipantId::from_raw).collect()
+    }
+
+    #[test]
+    fn foreign_chain_leaders_refiner__should_allow_nobody_when_chain_has_no_supporters() {
+        // Given
+        let (_sender, receiver) = watch::channel(SupportersByForeignChain::new());
+        let refiner =
+            ForeignChainLeadersRefiner::new(receiver, Some(ReconstructionThreshold::new(1)));
+
+        // When
+        let refined = refiner.refine(
+            &bitcoin_verify_foreign_tx_request(),
+            &participant_set(&[0, 1]),
+        );
+
+        // Then
+        assert!(refined.is_empty());
+    }
+
+    #[test]
+    fn foreign_chain_leaders_refiner__should_intersect_supporters_with_eligible() {
+        // Given
+        let supporters = SupportersByForeignChain::from([(
+            dtos::ForeignChain::Bitcoin,
+            participant_set(&[1, 2]),
+        )]);
+        let (_sender, receiver) = watch::channel(supporters);
+        let refiner =
+            ForeignChainLeadersRefiner::new(receiver, Some(ReconstructionThreshold::new(1)));
+
+        // When
+        let refined = refiner.refine(
+            &bitcoin_verify_foreign_tx_request(),
+            &participant_set(&[0, 1]),
+        );
+
+        // Then
+        assert_eq!(refined, participant_set(&[1]));
+    }
+
+    #[test]
+    fn foreign_chain_leaders_refiner__should_allow_nobody_when_eligible_supporters_below_quorum() {
+        // Given: three supporters, quorum 2, but only one supporter is eligible.
+        let supporters = SupportersByForeignChain::from([(
+            dtos::ForeignChain::Bitcoin,
+            participant_set(&[1, 2, 3]),
+        )]);
+        let (_sender, receiver) = watch::channel(supporters);
+        let refiner =
+            ForeignChainLeadersRefiner::new(receiver, Some(ReconstructionThreshold::new(2)));
+
+        // When
+        let refined = refiner.refine(
+            &bitcoin_verify_foreign_tx_request(),
+            &participant_set(&[0, 1]),
+        );
+
+        // Then
+        assert!(refined.is_empty());
+    }
+
+    #[test]
+    fn foreign_chain_leaders_refiner__should_allow_supporters_when_eligible_quorum_is_met() {
+        // Given: three supporters, quorum 2, two of them eligible.
+        let supporters = SupportersByForeignChain::from([(
+            dtos::ForeignChain::Bitcoin,
+            participant_set(&[1, 2, 3]),
+        )]);
+        let (_sender, receiver) = watch::channel(supporters);
+        let refiner =
+            ForeignChainLeadersRefiner::new(receiver, Some(ReconstructionThreshold::new(2)));
+
+        // When
+        let refined = refiner.refine(
+            &bitcoin_verify_foreign_tx_request(),
+            &participant_set(&[0, 1, 2]),
+        );
+
+        // Then
+        assert_eq!(refined, participant_set(&[1, 2]));
+    }
+
+    #[test]
+    fn foreign_chain_leaders_refiner__should_pick_up_republished_supporters() {
+        // Given
+        let (sender, receiver) = watch::channel(SupportersByForeignChain::new());
+        let refiner =
+            ForeignChainLeadersRefiner::new(receiver, Some(ReconstructionThreshold::new(1)));
+        let eligible = participant_set(&[0, 1]);
+        assert!(
+            refiner
+                .refine(&bitcoin_verify_foreign_tx_request(), &eligible)
+                .is_empty()
+        );
+
+        // When
+        sender
+            .send(SupportersByForeignChain::from([(
+                dtos::ForeignChain::Bitcoin,
+                participant_set(&[0]),
+            )]))
+            .unwrap();
+        let refined = refiner.refine(&bitcoin_verify_foreign_tx_request(), &eligible);
+
+        // Then
+        assert_eq!(refined, participant_set(&[0]));
+    }
 }

crates/node/src/mpc_client.rs

@@ -265,6 +265,7 @@ impl MpcClient {
             self.client.my_participant_id(),
             self.client.clone(),
         );
+
         let mut pending_verify_foreign_txs = PendingRequests::<
             VerifyForeignTxRequest,
             contract_args::VerifyForeignTransactionRespondArgs,
@@ -273,6 +274,10 @@ impl MpcClient {
             self.client.all_participant_ids(),
             self.client.my_participant_id(),
             self.client.clone(),
+        )
+        .with_eligible_leaders_refiner(
+            self.verify_foreign_tx_provider
+                .new_eligible_leaders_refiner(),
         );
 
         let mut recent_blocks = RecentBlocksTracker::new(REQUEST_EXPIRATION_BLOCKS);

crates/node/src/providers/verify_foreign_tx.rs

@@ -1,6 +1,6 @@
 mod sign;
 
-use crate::foreign_chain_policy::SupportersByForeignChain;
+use crate::foreign_chain_policy::{ForeignChainLeadersRefiner, SupportersByForeignChain};
 use crate::network::NetworkTaskChannel;
 use crate::primitives::{MpcTaskId, UniqueId};
 use crate::providers::EcdsaSignatureProvider;
@@ -25,6 +25,7 @@ use foreign_chain_rpc_auth::auth_config_to_rpc_auth;
 use foreign_chain_rpc_interfaces::aptos::ReqwestAptosClient;
 use foreign_chain_rpc_interfaces::sui::GrpcSuiClient;
 use mpc_node_config::{ConfigFile, ForeignChainConfig, ForeignChainsConfig};
+use mpc_primitives::ReconstructionThreshold;
 use near_mpc_contract_interface::types::ProviderId;
 use std::sync::Arc;
 use std::time::Duration;
@@ -161,6 +162,9 @@ pub struct VerifyForeignTxProvider {
     config: Arc<ConfigFile>,
     inspectors: ForeignChainInspectors<HttpClient>,
     supporters_by_foreign_chain: watch::Receiver<SupportersByForeignChain>,
+    /// [`foreign_tx_reconstruction_threshold`](crate::foreign_chain_policy::foreign_tx_reconstruction_threshold)
+    /// of the running domains; `None` when there is no ForeignTx domain.
+    foreign_tx_reconstruction_threshold: Option<ReconstructionThreshold>,
     verify_foreign_tx_request_store: Arc<VerifyForeignTransactionRequestStorage>,
     ecdsa_signature_provider: Arc<EcdsaSignatureProvider>,
 }
@@ -183,6 +187,7 @@ impl VerifyForeignTxProvider {
     pub fn new(
         config: Arc<ConfigFile>,
         supporters_by_foreign_chain: watch::Receiver<SupportersByForeignChain>,
+        foreign_tx_reconstruction_threshold: Option<ReconstructionThreshold>,
         verify_foreign_tx_request_store: Arc<VerifyForeignTransactionRequestStorage>,
         ecdsa_signature_provider: Arc<EcdsaSignatureProvider>,
     ) -> anyhow::Result<Self> {
@@ -191,11 +196,19 @@ impl VerifyForeignTxProvider {
             config,
             inspectors,
             supporters_by_foreign_chain,
+            foreign_tx_reconstruction_threshold,
             verify_foreign_tx_request_store,
             ecdsa_signature_provider,
         })
     }
 
+    pub(crate) fn new_eligible_leaders_refiner(&self) -> ForeignChainLeadersRefiner {
+        ForeignChainLeadersRefiner::new(
+            self.supporters_by_foreign_chain.clone(),
+            self.foreign_tx_reconstruction_threshold,
+        )
+    }
+
     pub async fn process_channel(&self, channel: NetworkTaskChannel) -> anyhow::Result<()> {
         match channel.task_id() {
             MpcTaskId::VerifyForeignTxTaskId(task) => match task {

crates/node/src/providers/verify_foreign_tx/sign.rs

@@ -94,12 +94,16 @@ impl VerifyForeignTxProvider {
             snapshot.get(&requested_chain).cloned().unwrap_or_default()
         };
 
-        // If we don't support the foreign chain, we can't lead the computation.
-        // TODO(#3961): narrow leader selection to only chain supporters.
+        // Leader selection already narrows to chain supporters. Re-check as
+        // defense-in-depth, since the supporters snapshot may have changed
+        // between leader selection and this attempt.
         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:?}");
+            anyhow::bail!(
+                "selected as leader for a {requested_chain:?} request but this node no longer \
+                 supports that chain. Supporters must have changed since leader selection"
+            );
         }
 
         let response_payload = self

crates/node/src/requests.rs

@@ -1,4 +1,5 @@
 pub mod queue;
 
+pub(crate) mod metrics;
+
 mod debug;
-mod metrics;

crates/node/src/requests/debug.rs

@@ -1,5 +1,6 @@
 use super::queue::{
     ComputationProgress, EligibleLeadersAndHeights, PendingRequests, QueuedRequest,
+    RefineEligibleLeaders,
 };
 use crate::indexer::types::ChainRespondArgs;
 use crate::primitives::ParticipantId;
@@ -175,8 +176,12 @@ impl<RequestType: Request, ChainRespondArgsType: ChainRespondArgs>
     }
 }
 
-impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs> Debug
-    for PendingRequests<RequestType, ChainRespondArgsType>
+impl<RequestType, ChainRespondArgsType, Refiner> Debug
+    for PendingRequests<RequestType, ChainRespondArgsType, Refiner>
+where
+    RequestType: Request + Clone,
+    ChainRespondArgsType: ChainRespondArgs,
+    Refiner: RefineEligibleLeaders<RequestType>,
 {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let mut request_lines = Vec::new();
@@ -189,8 +194,14 @@ impl<RequestType: Request + Clone, ChainRespondArgsType: ChainRespondArgs> Debug
         let indexer_heights = self.network_api.indexer_heights();
 
         for request in self.requests.values() {
-            let debug_line =
-                request.debug_print(&self.clock, self.my_participant_id, &eligible_leaders);
+            let request_eligible_leaders = self
+                .refine_eligible_leaders
+                .refine(&request.request, &eligible_leaders);
+            let debug_line = request.debug_print(
+                &self.clock,
+                self.my_participant_id,
+                &request_eligible_leaders,
+            );
             request_lines.push((
                 request.block_height.into(),
                 request.request.get_id(),

crates/node/src/requests/metrics.rs

@@ -238,3 +238,15 @@ pub static MPC_CLUSTER_FAILED_SIGNATURES_COUNT: LazyLock<prometheus::IntCounterV
         )
         .unwrap()
     });
+
+pub static MPC_NUM_REQUESTS_WITHOUT_REFINED_LEADER_TOTAL: LazyLock<prometheus::IntCounterVec> =
+    LazyLock::new(|| {
+        prometheus::register_int_counter_vec!(
+            "mpc_num_requests_without_refined_leader_total",
+            "Number of queue passes where a request had eligible leaders, but none was allowed \
+             by the eligible-leaders refiner (e.g. no participant supports the request's \
+             foreign chain), so the request stays parked",
+            &["request_type"]
+        )
+        .unwrap()
+    });