feat: libp2p sync [skip-line-limit]

crates/aggregator/src/publickey_aggregator.rs

@@ -42,13 +42,13 @@ pub enum PublicKeyAggregatorState {
         no_proof_parties: Vec<u64>,
     },
     GeneratingC5Proof {
-        public_key: Vec<u8>,
+        public_key: ArcBytes,
         public_key_hash: [u8; 32],
         keyshare_bytes: Vec<ArcBytes>,
         nodes: OrderedSet<String>,
     },
     Complete {
-        public_key: Vec<u8>,
+        public_key: ArcBytes,
         keyshares: OrderedSet<ArcBytes>,
         nodes: OrderedSet<String>,
     },
@@ -293,12 +293,13 @@ impl PublicKeyAggregator {
         let committee_h = honest_keyshares.len();
 
         info!("Publishing PkAggregationProofPending for C5 proof generation...");
+        let pubkey = ArcBytes::from_bytes(&pubkey);
         self.bus.publish(
             PkAggregationProofPending {
                 e3_id: self.e3_id.clone(),
                 proof_request: PkAggregationProofRequest {
                     keyshare_bytes: honest_keyshares.clone(),
-                    aggregated_pk_bytes: ArcBytes::from_bytes(&pubkey),
+                    aggregated_pk_bytes: pubkey.clone(),
                     params_preset: self.params_preset.clone(),
                     // this field is not really used in the circuit, we only use H
                     committee_n: committee_h,
@@ -316,7 +317,7 @@ impl PublicKeyAggregator {
         // Transition to GeneratingC5Proof
         self.state.try_mutate(&ec, |_| {
             Ok(PublicKeyAggregatorState::GeneratingC5Proof {
-                public_key: pubkey,
+                public_key: pubkey.clone(),
                 public_key_hash,
                 keyshare_bytes: honest_keyshares,
                 nodes: honest_nodes_set,

crates/ciphernode-builder/Cargo.toml

@@ -29,6 +29,7 @@ e3-sortition.workspace = true
 e3-sync.workspace = true
 e3-trbfv.workspace = true
 e3-utils.workspace = true
+libp2p.workspace = true
 rayon.workspace = true
 tempfile.workspace = true
 tokio.workspace = true

crates/ciphernode-builder/src/ciphernode.rs

@@ -8,7 +8,8 @@ use actix::Addr;
 use anyhow::Result;
 use e3_data::{DataStore, InMemStore, StoreAddr};
 use e3_events::{BusHandle, EnclaveEvent, HistoryCollector};
-use tokio::task::JoinHandle;
+use e3_net::NetChannelBridge;
+use libp2p::PeerId;
 
 /// A Sharable handle to a Ciphernode. NOTE: clones are available for use in the CiphernodeSystem
 /// but they cannot await the task.
@@ -19,19 +20,27 @@ pub struct CiphernodeHandle {
     pub bus: BusHandle,
     pub history: Option<Addr<HistoryCollector<EnclaveEvent>>>,
     pub errors: Option<Addr<HistoryCollector<EnclaveEvent>>>,
-    pub peer_id: String,
-    pub join_handle: JoinHandle<Result<()>>,
+    pub peer_id: PeerId,
+    pub channel_bridge: Option<NetChannelBridge>,
 }
 
+impl PartialEq for CiphernodeHandle {
+    fn eq(&self, other: &Self) -> bool {
+        self.address == other.address && self.peer_id == other.peer_id
+    }
+}
+
+impl Eq for CiphernodeHandle {}
+
 impl CiphernodeHandle {
     pub fn new(
         address: String,
         store: DataStore,
         bus: BusHandle,
         history: Option<Addr<HistoryCollector<EnclaveEvent>>>,
         errors: Option<Addr<HistoryCollector<EnclaveEvent>>>,
-        peer_id: String,
-        join_handle: JoinHandle<Result<()>>,
+        peer_id: PeerId,
+        channel_bridge: Option<NetChannelBridge>,
     ) -> Self {
         Self {
             address,
@@ -40,7 +49,7 @@ impl CiphernodeHandle {
             history,
             errors,
             peer_id,
-            join_handle,
+            channel_bridge,
         }
     }
 
@@ -64,8 +73,10 @@ impl CiphernodeHandle {
         &self.store
     }
 
-    pub fn split(self) -> (BusHandle, JoinHandle<Result<()>>) {
-        (self.bus, self.join_handle)
+    pub fn channel_bridge(&self) -> Result<NetChannelBridge> {
+        Ok(self.channel_bridge.clone().ok_or(anyhow::anyhow!(
+            "No channel bridge exists. We are likely not in test mode"
+        ))?)
     }
 
     pub fn in_mem_store(&self) -> Option<&Addr<InMemStore>> {

crates/ciphernode-builder/src/ciphernode_builder.rs

@@ -23,7 +23,10 @@ use e3_fhe::ext::FheExtension;
 use e3_fhe_params::{BfvPreset, DEFAULT_BFV_PRESET};
 use e3_keyshare::ext::ThresholdKeyshareExtension;
 use e3_multithread::{Multithread, MultithreadReport, TaskPool};
-use e3_net::{setup_net, NetRepositoryFactory};
+use e3_net::{
+    create_channel_bridge, setup_libp2p_keypair, setup_net, setup_net_interface,
+    NetRepositoryFactory,
+};
 use e3_request::E3Router;
 use e3_sortition::{
     CiphernodeSelector, CiphernodeSelectorFactory, FinalizedCommitteesRepositoryFactory,
@@ -32,6 +35,7 @@ use e3_sortition::{
 use e3_sync::sync;
 use e3_utils::SharedRng;
 use e3_zk_prover::{setup_zk_actors, ZkBackend};
+use libp2p::PeerId;
 use std::time::Duration;
 use std::{collections::HashMap, path::PathBuf, sync::Arc};
 use tracing::{error, info};
@@ -496,28 +500,28 @@ impl CiphernodeBuilder {
             ))
         }
 
-        info!("building...");
-
+        info!("E3Router building...");
         e3_builder.build().await?;
 
-        let (join_handle, peer_id) = if let Some(net_config) = self.net_config {
+        let topic = "enclave-gossip";
+        let (peer_id, interface, channel_bridge) = if let Some(net_config) = self.net_config {
+            // Setup real net interface
             let repositories = store.repositories();
-            setup_net(
-                bus.clone(),
-                net_config.peers,
-                &self.cipher,
-                net_config.quic_port,
-                repositories.libp2p_keypair(),
-                eventstore_ts,
-            )
-            .await?
+            let keypair = setup_libp2p_keypair(repositories.libp2p_keypair(), &self.cipher).await?;
+            let peer_id = keypair.peer_id();
+            let interface =
+                setup_net_interface(topic, keypair, net_config.peers, net_config.quic_port)?;
+            (peer_id, interface, None)
         } else {
-            (
-                tokio::spawn(std::future::ready(Ok(()))),
-                "-not set-".to_string(),
-            )
+            // Setup test net interface with random PeerId
+            let (interface, channel_bridge) = create_channel_bridge();
+            let peer_id = PeerId::random();
+            let channel_bridge = Some(channel_bridge);
+            (peer_id, interface, channel_bridge)
         };
 
+        setup_net(topic, bus.clone(), eventstore_ts, interface)?;
+
         // Run the sync routine
         sync(
             &bus,
@@ -535,7 +539,7 @@ impl CiphernodeBuilder {
             history,
             errors,
             peer_id,
-            join_handle,
+            channel_bridge,
         ))
     }
 

crates/entrypoint/src/helpers/shutdown.rs

@@ -7,53 +7,20 @@
 use e3_ciphernode_builder::CiphernodeHandle;
 use e3_events::{prelude::*, Shutdown};
 use std::time::Duration;
-use tokio::{
-    select,
-    signal::unix::{signal, SignalKind},
-};
+use tokio::signal::unix::{signal, SignalKind};
 use tracing::{error, info};
 
 pub async fn listen_for_shutdown(node: CiphernodeHandle) {
-    let (bus, mut handle) = node.split();
+    let bus = node.bus;
     let mut sigterm =
         signal(SignalKind::terminate()).expect("Failed to create SIGTERM signal stream");
-    select! {
-        _ = sigterm.recv() => {
-            info!("SIGTERM received, initiating graceful shutdown...");
+    sigterm.recv().await;
+    info!("SIGTERM received, initiating graceful shutdown...");
 
-            // Stop the actor system
-            match bus.publish_without_context(Shutdown){
-                Ok(_) => (),
-                Err(e) => error!("Shutdown failed to publish! {e}")
-            }
-
-            // Wait for all events to propagate
-            tokio::time::sleep(Duration::from_secs(2)).await;
-
-            // Abort the spawned task
-            handle.abort();
-
-            // Wait for all actor processes to disconnect
-            tokio::time::sleep(Duration::from_secs(2)).await;
-
-            // Wait for the task to finish
-            let _ = handle.await;
-
-            info!("Graceful shutdown complete");
-
-        }
-        result = &mut handle => {
-            match result {
-                Ok(Ok(_)) => {
-                    info!("Completed");
-                }
-                Ok(Err(e)) => {
-                    error!("Failed: {}", e);
-                }
-                Err(e) => {
-                    error!("Panicked: {}", e);
-                }
-            }
-        }
+    if let Err(e) = bus.publish_without_context(Shutdown) {
+        error!("Shutdown failed to publish! {e}");
     }
+
+    tokio::time::sleep(Duration::from_secs(2)).await;
+    info!("Graceful shutdown complete");
 }

crates/events/src/bus_handle.rs

@@ -7,8 +7,8 @@
 use actix::{Actor, Addr, Handler, Recipient};
 use anyhow::Result;
 use derivative::Derivative;
-use e3_utils::MAILBOX_LIMIT;
-use std::marker::PhantomData;
+use e3_utils::{actix::channel::oneshot, MAILBOX_LIMIT};
+use std::{future::Future, marker::PhantomData, pin::Pin};
 use tracing::error;
 
 use crate::{
@@ -287,6 +287,20 @@ impl<S> EventSubscriber<EnclaveEvent<Sequenced>> for BusHandle<S> {
         self.event_bus
             .do_send(Unsubscribe::new(event_type, recipient));
     }
+
+    fn wait_for(
+        &self,
+        event_type: EventType,
+    ) -> Pin<Box<dyn Future<Output = Result<EnclaveEvent<Sequenced>>> + Send>> {
+        let (addr, rx) = oneshot::<EnclaveEvent<Sequenced>>();
+        self.subscribe(event_type, addr.clone());
+        let bus = self.event_bus.clone();
+        Box::pin(async move {
+            let r = rx.await?;
+            bus.do_send(Unsubscribe::new(event_type, addr));
+            Ok(r)
+        })
+    }
 }
 
 impl<S> EventContextManager for BusHandle<S> {

crates/events/src/enclave_event/enable_effects.rs

@@ -13,7 +13,7 @@ use std::fmt::{self, Display};
 #[derive(Message, Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 #[rtype(result = "()")]
 pub struct EffectsEnabled {
-    pub correlation_id: CorrelationId,
+    correlation_id: CorrelationId,
 }
 
 impl EffectsEnabled {