Merge branch 'main' into fix/solver-setup

Author: AnomalRoil

Cargo.lock

@@ -100,5 +100,5 @@ tower-http = { version = "0.6.6" }
 
 [patch.crates-io]
 # Use custom patch to fix arkworks' FieldHasher
-ark-ff = { git = 'https://github.com/azixus/algebra', branch = "fix/fieldHasher-v0.5.0" }
-libp2p-floodsub = { git = "https://github.com/azixus/rust-libp2p", branch = "fix/floodsub-msg-size" }
+ark-ff = { git = 'https://github.com/randa-mu/algebra', branch = "fix/fieldHasher-v0.5.0" }
+libp2p-floodsub = { git = "https://github.com/randa-mu/rust-libp2p", branch = "fix/floodsub-msg-size" }

Cargo.toml

@@ -128,7 +128,7 @@ async fn adkg_pairing_out_g2<'a, E, S, TBT>(
     adkg_scheme: S,
     topic_transport: Arc<TBT>,
     writer: Option<InMemoryWriter>,
-    mut rng: impl AdkgRng + 'static,
+    rng: impl AdkgRng + 'static,
 ) -> anyhow::Result<()>
 where
     E: Pairing,
@@ -171,7 +171,7 @@ where
                 group_config,
                 topic_transport,
                 adkg_scheme,
-                &mut rng,
+                rng,
                 tx_adkg_out,
             )
             .await
@@ -315,7 +315,7 @@ async fn adkg_dxkr23<S, TBT>(
     group_config: GroupConfig,
     topic_transport: Arc<TBT>,
     adkg_scheme: S,
-    rng: &mut impl AdkgRng,
+    rng: impl AdkgRng + 'static,
     out: oneshot::Sender<AdkgOutput<S::Curve>>,
 ) -> anyhow::Result<()>
 where
@@ -325,9 +325,9 @@ where
     S::ABAConfig: AbaConfig<'static, PartyId, Input = AbaCrainInput<S::Curve>>,
     <S::ACSSConfig as AcssConfig<'static, S::Curve, PartyId>>::Output:
         Into<ShareWithPoly<S::Curve>>,
-    TBT: TopicBasedTransport<Identity = PartyId>,
+    TBT: TopicBasedTransport<Identity = PartyId> + Send + Sync + 'static,
 {
-    let mut adkg = adkg_scheme.new_adkg(
+    let adkg = adkg_scheme.new_adkg(
         adkg_config.id,
         group_config.n,
         group_config.t,
@@ -336,6 +336,10 @@ where
         pks.clone(),
     )?;
 
+    let (adkg_start_tx, adkg_start_rx) = oneshot::channel();
+    let (adkg_stop_tx, adkg_stop_rx) = oneshot::channel();
+    let adkg_out = adkg.run(adkg_start_rx, adkg_stop_rx, rng, topic_transport);
+
     // Calculate time to sleep before actively executing the adkg
     let sleep_duration = (group_config.start_time - chrono::Utc::now())
         .to_std() // TimeDelta to positive duration
@@ -353,11 +357,14 @@ where
         "Executing ADKG with a timeout of {}",
         humantime::format_duration(adkg_config.timeout)
     );
+    if adkg_start_tx.send(()).is_err() {
+        anyhow::bail!("Failed to send ADKG start signal");
+    }
 
     let res = tokio::select! {
-        output = adkg.start(rng, topic_transport) => {
-            let output = match output {
-                Ok(adkg_out) => {
+        output = adkg_out => {
+            let output: anyhow::Result<_> = match output {
+                Some(Ok(adkg_out)) => {
                     tracing::info!(used_sessions = ?adkg_out.used_sessions, "Successfully obtained secret key & output from ADKG");
                     if out.send(adkg_out).is_err() {
                         // fails if the receiver side is dropped early
@@ -368,9 +375,13 @@ where
                     tokio::time::sleep(adkg_config.grace_period).await;
                     Ok(())
                 }
-                Err(e) => {
+                Some(Err(e)) => {
                     tracing::error!("failed to obtain output from ADKG: {e:?}");
-                    Err(e)
+                    Err(e.into())
+                }
+                None => {
+                    tracing::error!("failed to obtain output from ADKG: stopped before an output");
+                    Err(anyhow!("ADKG stopped before output"))
                 }
             };
 
@@ -384,9 +395,9 @@ where
     };
 
     tracing::warn!("Stopping ADKG...");
-    adkg.stop().await;
+    let _ = adkg_stop_tx.send(());
 
-    Ok(res??)
+    res?
 }
 
 /// Pairing-based DLEQ proof that there exists an s_j s.t. P_1 = [s_j] G_1 \land P_2 = [s_j] G_2,

bin/adkg-cli/src/adkg_dxkr23.rs

@@ -98,7 +98,7 @@ where
         decryption_sender_contract.provider().to_owned(),
         block_poll_interval,
     )
-    .await?;
+    .await;
     // Transform each stream into a stream of events
     let decryption_requested_stream =
         decryption_requested_stream.map(|(req, _)| ChainEvent::DecryptionRequested(req));

bin/blocklock-agent/src/lib.rs

@@ -99,7 +99,7 @@ where
         signature_sender_contract.provider().to_owned(),
         block_poll_interval,
     )
-    .await?;
+    .await;
 
     // Transform each stream into a stream of events
     let signature_requested_stream =