feat: sync mode preparation [skip-line-limit]

Cargo.toml

@@ -29,6 +29,7 @@ members = [
   "crates/sdk",
   "crates/sortition",
   "crates/support-scripts",
+  "crates/sync",
   "crates/test-helpers",
   "crates/tests",
   "crates/trbfv",
@@ -51,6 +52,8 @@ exclude = [
 ]
 resolver = "3"
 msrv = "1.86.0"
+
+[workspace.metadata.release]
 shared-version = true
 pre-release-commit-message = "chore: Release {{crate_name}} v{{version}}"
 pre-release-replacements = [
@@ -98,6 +101,7 @@ e3-sortition = { version = "0.1.7", path = "./crates/sortition" }
 e3-program-server = { version = "0.1.7", path = "./crates/program-server" }
 e3-polynomial = { version = "0.1.7", path = "./crates/polynomial" }
 e3-support-scripts = { version = "0.1.7", path = "./crates/support-scripts" }
+e3-sync = { version = "0.1.7", path = "./crates/sync" }
 e3-test-helpers = { version = "0.1.7", path = "./crates/test-helpers" }
 e3-tests = { version = "0.1.7", path = "./crates/tests" }
 e3-trbfv = { version = "0.1.7", path = "./crates/trbfv" }
@@ -195,6 +199,8 @@ libp2p = { version = "=0.54.1", features = [
   "ping",
   "quic",
   "tokio",
+  "request-response",
+  "cbor"
 ]}
 zeroize = "=1.8.1"
 

crates/Dockerfile

@@ -74,6 +74,7 @@ COPY crates/safe/Cargo.toml ./safe/Cargo.toml
 COPY crates/sdk/Cargo.toml ./sdk/Cargo.toml
 COPY crates/sortition/Cargo.toml ./sortition/Cargo.toml
 COPY crates/support-scripts/Cargo.toml ./support-scripts/Cargo.toml
+COPY crates/sync/Cargo.toml ./sync/Cargo.toml
 COPY crates/test-helpers/Cargo.toml ./test-helpers/Cargo.toml
 COPY crates/tests/Cargo.toml ./tests/Cargo.toml
 COPY crates/trbfv/Cargo.toml ./trbfv/Cargo.toml

crates/aggregator/src/committee_finalizer.rs

@@ -9,6 +9,7 @@ use e3_events::{
     prelude::*, trap, BusHandle, CommitteeFinalizeRequested, CommitteeRequested, EType,
     EnclaveEvent, EnclaveEventData, EventType, Shutdown,
 };
+use e3_utils::NotifySync;
 use std::collections::HashMap;
 use std::time::Duration;
 use tracing::{error, info};
@@ -48,8 +49,8 @@ impl Handler<EnclaveEvent> for CommitteeFinalizer {
     type Result = ();
     fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
         match msg.into_data() {
-            EnclaveEventData::CommitteeRequested(data) => ctx.notify(data),
-            EnclaveEventData::Shutdown(data) => ctx.notify(data),
+            EnclaveEventData::CommitteeRequested(data) => self.notify_sync(ctx, data),
+            EnclaveEventData::Shutdown(data) => self.notify_sync(ctx, data),
             _ => (),
         }
     }

crates/aggregator/src/publickey_aggregator.rs

@@ -12,8 +12,10 @@ use e3_events::{
     prelude::*, BusHandle, Die, E3id, EnclaveEvent, EnclaveEventData, KeyshareCreated, OrderedSet,
     PublicKeyAggregated, Seed,
 };
+use e3_events::{trap, EType};
 use e3_fhe::{Fhe, GetAggregatePublicKey};
 use e3_utils::ArcBytes;
+use e3_utils::NotifySync;
 use std::sync::Arc;
 use tracing::{error, info};
 
@@ -139,11 +141,14 @@ impl Actor for PublicKeyAggregator {
 impl Handler<EnclaveEvent> for PublicKeyAggregator {
     type Result = ();
     fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
-        match msg.into_data() {
-            EnclaveEventData::KeyshareCreated(data) => ctx.notify(data),
-            EnclaveEventData::E3RequestComplete(_) => ctx.notify(Die),
-            _ => (),
-        }
+        trap(EType::KeyGeneration, &self.bus.clone(), || {
+            match msg.into_data() {
+                EnclaveEventData::KeyshareCreated(data) => self.notify_sync(ctx, data)?,
+                EnclaveEventData::E3RequestComplete(_) => self.notify_sync(ctx, Die),
+                _ => (),
+            };
+            Ok(())
+        });
     }
 }
 
@@ -163,10 +168,13 @@ impl Handler<KeyshareCreated> for PublicKeyAggregator {
         self.add_keyshare(pubkey, node)?;
 
         if let Some(PublicKeyAggregatorState::Computing { keyshares, .. }) = &self.state.get() {
-            ctx.notify(ComputeAggregate {
-                keyshares: keyshares.clone(),
-                e3_id,
-            })
+            self.notify_sync(
+                ctx,
+                ComputeAggregate {
+                    keyshares: keyshares.clone(),
+                    e3_id,
+                },
+            )?
         }
 
         Ok(())
@@ -215,6 +223,6 @@ impl Handler<ComputeAggregate> for PublicKeyAggregator {
 impl Handler<Die> for PublicKeyAggregator {
     type Result = ();
     fn handle(&mut self, _: Die, ctx: &mut Self::Context) -> Self::Result {
-        ctx.stop()
+        ctx.stop();
     }
 }

crates/aggregator/src/threshold_plaintext_aggregator.rs

@@ -20,6 +20,7 @@ use e3_trbfv::{
     TrBFVResponse,
 };
 use e3_utils::utility_types::ArcBytes;
+use e3_utils::NotifySync;
 use tracing::{debug, error, info, trace};
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
@@ -263,8 +264,8 @@ impl Handler<EnclaveEvent> for ThresholdPlaintextAggregator {
     fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
         match msg.into_data() {
             EnclaveEventData::DecryptionshareCreated(data) => ctx.notify(data),
-            EnclaveEventData::E3RequestComplete(_) => ctx.notify(Die),
-            EnclaveEventData::ComputeResponse(data) => ctx.notify(data),
+            EnclaveEventData::E3RequestComplete(_) => self.notify_sync(ctx, Die),
+            EnclaveEventData::ComputeResponse(data) => self.notify_sync(ctx, data),
             _ => (),
         }
     }
@@ -318,12 +319,15 @@ impl Handler<DecryptionshareCreated> for ThresholdPlaintextAggregator {
                         ..
                     })) = act.state.get()
                     {
-                        ctx.notify(ComputeAggregate {
-                            shares: shares.clone(),
-                            ciphertext_output: ciphertext_output.clone(),
-                            threshold_m,
-                            threshold_n,
-                        })
+                        act.notify_sync(
+                            ctx,
+                            ComputeAggregate {
+                                shares: shares.clone(),
+                                ciphertext_output: ciphertext_output.clone(),
+                                threshold_m,
+                                threshold_n,
+                            },
+                        )
                     }
 
                     Ok(())

crates/ciphernode-builder/Cargo.toml

@@ -21,8 +21,10 @@ e3-evm.workspace = true
 e3-fhe.workspace = true
 e3-keyshare.workspace = true
 e3-multithread.workspace = true
+e3-net.workspace = true
 e3-request.workspace = true
 e3-sortition.workspace = true
+e3-sync.workspace = true
 e3-trbfv.workspace = true
 e3-utils.workspace = true
 rayon.workspace = true

crates/ciphernode-builder/src/ciphernode.rs

@@ -5,16 +5,22 @@
 // or FITNESS FOR A PARTICULAR PURPOSE.
 
 use actix::Addr;
+use anyhow::Result;
 use e3_data::{DataStore, InMemStore, StoreAddr};
 use e3_events::{BusHandle, EnclaveEvent, HistoryCollector};
+use tokio::task::JoinHandle;
 
-#[derive(Clone, Debug)]
+/// A Sharable handle to a Ciphernode. NOTE: clones are available for use in the CiphernodeSystem
+/// but they cannot await the task.
+#[derive(Debug)]
 pub struct CiphernodeHandle {
     pub address: String,
     pub store: DataStore,
     pub bus: BusHandle,
     pub history: Option<Addr<HistoryCollector<EnclaveEvent>>>,
     pub errors: Option<Addr<HistoryCollector<EnclaveEvent>>>,
+    pub peer_id: String,
+    pub join_handle: JoinHandle<Result<()>>,
 }
 
 impl CiphernodeHandle {
@@ -24,13 +30,17 @@ impl CiphernodeHandle {
         bus: BusHandle,
         history: Option<Addr<HistoryCollector<EnclaveEvent>>>,
         errors: Option<Addr<HistoryCollector<EnclaveEvent>>>,
+        peer_id: String,
+        join_handle: JoinHandle<Result<()>>,
     ) -> Self {
         Self {
             address,
             store,
             bus,
             history,
             errors,
+            peer_id,
+            join_handle,
         }
     }
 
@@ -54,6 +64,10 @@ impl CiphernodeHandle {
         &self.store
     }
 
+    pub fn split(self) -> (BusHandle, JoinHandle<Result<()>>) {
+        (self.bus, self.join_handle)
+    }
+
     pub fn in_mem_store(&self) -> Option<&Addr<InMemStore>> {
         let addr = self.store.get_addr();
         if let StoreAddr::InMem(ref store) = addr {