feat: make sign queue use less shared state

chain-signatures/node/src/cli.rs

@@ -191,7 +191,7 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
             mesh_options,
             message_options,
         } => {
-            let sign_queue = Arc::new(RwLock::new(SignQueue::new()));
+            let (sign_tx, sign_rx) = SignQueue::channel();
             let rt = tokio::runtime::Builder::new_multi_thread()
                 .enable_all()
                 .build()?;
@@ -221,7 +221,7 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
                 &indexer_options,
                 &mpc_contract_id,
                 &account_id,
-                &sign_queue,
+                sign_tx,
                 app_data_storage,
                 rpc_client.clone(),
             )?;
@@ -257,7 +257,7 @@ pub fn run(cmd: Cli) -> anyhow::Result<()> {
                 rpc_client.clone(),
                 signer,
                 receiver,
-                sign_queue,
+                sign_rx,
                 key_storage,
                 triple_storage,
                 presignature_storage,

chain-signatures/node/src/indexer.rs

@@ -1,4 +1,4 @@
-use crate::protocol::{SignQueue, SignRequest};
+use crate::protocol::SignRequest;
 use crate::storage::app_data_storage::AppDataStorage;
 use crypto_shared::{derive_epsilon, ScalarExt};
 use k256::Scalar;
@@ -13,7 +13,7 @@ use std::ops::Mul;
 use std::sync::Arc;
 use std::thread::JoinHandle;
 use std::time::{Duration, Instant};
-use tokio::sync::RwLock;
+use tokio::sync::{mpsc, RwLock};
 
 /// Configures indexer.
 #[derive(Debug, Clone, clap::Parser)]
@@ -169,7 +169,7 @@ impl Indexer {
 struct Context {
     mpc_contract_id: AccountId,
     node_account_id: AccountId,
-    queue: Arc<RwLock<SignQueue>>,
+    sign_tx: mpsc::Sender<SignRequest>,
     indexer: Indexer,
 }
 
@@ -267,14 +267,20 @@ async fn handle_block(
 
     // Add the requests after going through the whole block to avoid partial processing if indexer fails somewhere.
     // This way we can revisit the same block if we failed while not having added the requests partially.
-    let mut queue = ctx.queue.write().await;
     for request in pending_requests {
-        queue.add(request);
+        tracing::info!(
+            request_id = ?near_primitives::hash::CryptoHash(request.request_id),
+            payload = hex::encode(request.request.payload.to_bytes()),
+            entropy = hex::encode(request.entropy),
+            "new sign request"
+        );
+        if let Err(err) = ctx.sign_tx.send(request).await {
+            tracing::error!(?err, "failed to send the sign request into sign queue");
+        }
         crate::metrics::NUM_SIGN_REQUESTS
             .with_label_values(&[ctx.node_account_id.as_str()])
             .inc();
     }
-    drop(queue);
 
     let log_indexing_interval = 1000;
     if block.block_height() % log_indexing_interval == 0 {
@@ -292,7 +298,7 @@ pub fn run(
     options: &Options,
     mpc_contract_id: &AccountId,
     node_account_id: &AccountId,
-    queue: &Arc<RwLock<SignQueue>>,
+    sign_tx: mpsc::Sender<SignRequest>,
     app_data_storage: AppDataStorage,
     rpc_client: near_fetch::Client,
 ) -> anyhow::Result<(JoinHandle<anyhow::Result<()>>, Indexer)> {
@@ -308,7 +314,7 @@ pub fn run(
     let context = Context {
         mpc_contract_id: mpc_contract_id.clone(),
         node_account_id: node_account_id.clone(),
-        queue: queue.clone(),
+        sign_tx,
         indexer: indexer.clone(),
     };
 

chain-signatures/node/src/protocol/consensus.rs

@@ -3,7 +3,6 @@ use super::state::{
     JoiningState, NodeState, PersistentNodeData, RunningState, StartedState,
     WaitingForConsensusState,
 };
-use super::SignQueue;
 use crate::config::Config;
 use crate::gcp::error::SecretStorageError;
 use crate::http_client::MessageQueue;
@@ -12,6 +11,7 @@ use crate::protocol::presignature::PresignatureManager;
 use crate::protocol::signature::SignatureManager;
 use crate::protocol::state::{GeneratingState, ResharingState};
 use crate::protocol::triple::TripleManager;
+use crate::protocol::SignRequest;
 use crate::storage::presignature_storage::PresignatureStorage;
 use crate::storage::secret_storage::SecretNodeStorageBox;
 use crate::storage::triple_storage::TripleStorage;
@@ -25,7 +25,7 @@ use std::sync::Arc;
 use async_trait::async_trait;
 use cait_sith::protocol::InitializationError;
 use serde_json::json;
-use tokio::sync::RwLock;
+use tokio::sync::{mpsc, RwLock};
 use url::Url;
 
 use near_account_id::AccountId;
@@ -38,7 +38,7 @@ pub trait ConsensusCtx {
     fn signer(&self) -> &InMemorySigner;
     fn mpc_contract_id(&self) -> &AccountId;
     fn my_address(&self) -> &Url;
-    fn sign_queue(&self) -> Arc<RwLock<SignQueue>>;
+    fn sign_rx(&self) -> Arc<RwLock<mpsc::Receiver<SignRequest>>>;
     fn secret_storage(&self) -> &SecretNodeStorageBox;
     fn triple_storage(&self) -> &TripleStorage;
     fn presignature_storage(&self) -> &PresignatureStorage;
@@ -118,7 +118,6 @@ impl ConsensusProtocol for StartedState {
                         }
                         Ordering::Less => Err(ConsensusError::EpochRollback),
                         Ordering::Equal => {
-                            let sign_queue = ctx.sign_queue();
                             match contract_state
                                 .participants
                                 .find_participant(ctx.my_account_id())
@@ -147,9 +146,11 @@ impl ConsensusProtocol for StartedState {
                                     let signature_manager =
                                         Arc::new(RwLock::new(SignatureManager::new(
                                             me,
+                                            ctx.my_account_id(),
+                                            contract_state.threshold,
                                             public_key,
                                             epoch,
-                                            ctx.my_account_id(),
+                                            ctx.sign_rx(),
                                         )));
 
                                     Ok(NodeState::Running(RunningState {
@@ -158,7 +159,6 @@ impl ConsensusProtocol for StartedState {
                                         threshold: contract_state.threshold,
                                         private_share,
                                         public_key,
-                                        sign_queue,
                                         triple_manager,
                                         presignature_manager,
                                         signature_manager,
@@ -388,9 +388,11 @@ impl ConsensusProtocol for WaitingForConsensusState {
 
                     let signature_manager = Arc::new(RwLock::new(SignatureManager::new(
                         me,
+                        ctx.my_account_id(),
+                        self.threshold,
                         self.public_key,
                         self.epoch,
-                        ctx.my_account_id(),
+                        ctx.sign_rx(),
                     )));
 
                     Ok(NodeState::Running(RunningState {
@@ -399,7 +401,6 @@ impl ConsensusProtocol for WaitingForConsensusState {
                         threshold: self.threshold,
                         private_share: self.private_share,
                         public_key: self.public_key,
-                        sign_queue: ctx.sign_queue(),
                         triple_manager,
                         presignature_manager,
                         signature_manager,

chain-signatures/node/src/protocol/cryptography.rs

@@ -379,10 +379,9 @@ impl CryptographicProtocol for RunningState {
 
         let presig_task = PresignatureManager::execute(&self, &active, &cfg.protocol);
 
-        let me = ctx.me().await;
         let stable = mesh_state.stable_participants;
         tracing::debug!(?stable, "stable participants");
-        let sig_task = SignatureManager::execute(&self, &stable, me, &cfg.protocol, &ctx);
+        let sig_task = SignatureManager::execute(&self, &stable, &cfg.protocol, &ctx);
 
         match tokio::try_join!(triple_task, presig_task, sig_task) {
             Ok(_result) => (),
@@ -398,7 +397,7 @@ impl CryptographicProtocol for RunningState {
 
         let failures = messages
             .send_encrypted(
-                me,
+                ctx.me().await,
                 &cfg.local.network.sign_sk,
                 ctx.http_client(),
                 &active,

chain-signatures/node/src/protocol/mod.rs

@@ -49,7 +49,7 @@ struct Ctx {
     signer: InMemorySigner,
     rpc_client: near_fetch::Client,
     http_client: reqwest::Client,
-    sign_queue: Arc<RwLock<SignQueue>>,
+    sign_rx: Arc<RwLock<mpsc::Receiver<SignRequest>>>,
     secret_storage: SecretNodeStorageBox,
     triple_storage: TripleStorage,
     presignature_storage: PresignatureStorage,
@@ -81,8 +81,8 @@ impl ConsensusCtx for &mut MpcSignProtocol {
         &self.ctx.my_address
     }
 
-    fn sign_queue(&self) -> Arc<RwLock<SignQueue>> {
-        self.ctx.sign_queue.clone()
+    fn sign_rx(&self) -> Arc<RwLock<mpsc::Receiver<SignRequest>>> {
+        self.ctx.sign_rx.clone()
     }
 
     fn secret_storage(&self) -> &SecretNodeStorageBox {
@@ -155,7 +155,7 @@ impl MpcSignProtocol {
         rpc_client: near_fetch::Client,
         signer: InMemorySigner,
         receiver: mpsc::Receiver<MpcMessage>,
-        sign_queue: Arc<RwLock<SignQueue>>,
+        sign_rx: mpsc::Receiver<SignRequest>,
         secret_storage: SecretNodeStorageBox,
         triple_storage: TripleStorage,
         presignature_storage: PresignatureStorage,
@@ -179,7 +179,7 @@ impl MpcSignProtocol {
             mpc_contract_id,
             rpc_client,
             http_client: reqwest::Client::new(),
-            sign_queue,
+            sign_rx: Arc::new(RwLock::new(sign_rx)),
             signer,
             secret_storage,
             triple_storage,