Prepare for sequencers

packages/ciphernode/core/src/ciphernode_sequencer.rs

@@ -0,0 +1,40 @@
+// sequence and persist events for a single E3 request in the correct order
+// TODO: spawn and store a ciphernode upon start and forward all events to it in order
+// TODO: if the ciphernode fails restart the node by replaying all stored events back to it
+
+use actix::prelude::*;
+
+use crate::{Ciphernode, Data, EnclaveEvent, EventBus, Fhe};
+
+pub struct CiphernodeSequencer {
+    fhe: Addr<Fhe>,
+    data: Addr<Data>,
+    bus: Addr<EventBus>,
+    child: Option<Addr<Ciphernode>>,
+}
+impl CiphernodeSequencer {
+   pub fn new(fhe: Addr<Fhe>, data: Addr<Data>, bus: Addr<EventBus>) -> Self {
+        Self {
+            fhe,
+            bus,
+            data,
+            child: None,
+        }
+    }
+}
+impl Actor for CiphernodeSequencer {
+    type Context = Context<Self>;
+}
+
+impl Handler<EnclaveEvent> for CiphernodeSequencer {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
+        let bus = self.bus.clone();
+        let fhe = self.fhe.clone();
+        let data = self.data.clone();
+        let sink = self
+            .child
+            .get_or_insert_with(|| Ciphernode::new(bus, fhe, data).start());
+        sink.do_send(msg);
+    }
+}

packages/ciphernode/core/src/ciphernode_supervisor.rs

@@ -89,7 +89,7 @@ impl Handler<EnclaveEvent> for CiphernodeSupervisor {
             }
             EnclaveEvent::KeyshareCreated { data, .. } => {
                 if let Some(key) = self.publickey_aggregators.get(&data.e3_id) {
-                    key.do_send(data);
+                    key.do_send(EnclaveEvent::from(data));
                 }
             }
             EnclaveEvent::PublicKeyAggregated { data, .. } => {
@@ -120,7 +120,7 @@ impl Handler<EnclaveEvent> for CiphernodeSupervisor {
             }
             EnclaveEvent::DecryptionshareCreated { data, .. } => {
                 if let Some(decryption) = self.plaintext_aggregators.get(&data.e3_id) {
-                    decryption.do_send(data);
+                    decryption.do_send(EnclaveEvent::from(data));
                 }
             }
             EnclaveEvent::PlaintextAggregated { data, .. } => {

packages/ciphernode/core/src/events.rs

@@ -120,6 +120,20 @@ impl From<EnclaveEvent> for EventId {
     }
 }
 
+impl From<EnclaveEvent> for E3id {
+    fn from(value: EnclaveEvent) -> Self {
+        match value {
+            EnclaveEvent::KeyshareCreated { data, .. } => data.e3_id,
+            EnclaveEvent::CommitteeRequested { data, .. } => data.e3_id,
+            EnclaveEvent::PublicKeyAggregated { data, .. } => data.e3_id,
+            EnclaveEvent::CiphertextOutputPublished { data, .. } => data.e3_id,
+            EnclaveEvent::DecryptionshareCreated { data, .. } => data.e3_id,
+            EnclaveEvent::PlaintextAggregated { data, .. } => data.e3_id,
+            EnclaveEvent::CiphernodeSelected { data, .. } => data.e3_id,
+        }
+    }
+}
+
 impl From<KeyshareCreated> for EnclaveEvent {
     fn from(data: KeyshareCreated) -> Self {
         EnclaveEvent::KeyshareCreated {

packages/ciphernode/core/src/fhe.rs

@@ -53,26 +53,38 @@ impl Actor for Fhe {
 }
 
 impl Fhe {
-    pub fn new(
-        params: Arc<BfvParameters>,
-        crp: CommonRandomPoly,
-        rng: ChaCha20Rng,
-    ) -> Result<Self> {
-        Ok(Self { params, crp, rng })
+    pub fn new(params: Arc<BfvParameters>, crp: CommonRandomPoly, rng: ChaCha20Rng) -> Self {
+        Self { params, crp, rng }
     }
-    pub fn try_default() -> Result<Self> {
+
+     pub fn try_default() -> Result<Self> {
         let moduli = &vec![0x3FFFFFFF000001];
         let degree = 2048usize;
         let plaintext_modulus = 1032193u64;
-        let mut rng = ChaCha20Rng::from_entropy();
+        let rng = ChaCha20Rng::from_entropy();
+
+        Ok(Fhe::from_raw_params(
+            moduli,
+            degree,
+            plaintext_modulus,
+            rng,
+        )?)
+    }
+
+    pub fn from_raw_params(
+        moduli: &[u64],
+        degree: usize,
+        plaintext_modulus: u64,
+        mut rng: ChaCha20Rng,
+    ) -> Result<Self> {
         let params = BfvParametersBuilder::new()
             .set_degree(degree)
             .set_plaintext_modulus(plaintext_modulus)
             .set_moduli(&moduli)
             .build_arc()?;
         let crp = CommonRandomPoly::new(&params, &mut rng)?;
 
-        Ok(Fhe::new(params, crp, rng)?)
+        Ok(Fhe::new(params, crp, rng))
     }
 }
 

packages/ciphernode/core/src/lib.rs

@@ -16,6 +16,10 @@ mod p2p;
 mod plaintext_aggregator;
 mod publickey_aggregator;
 mod serializers;
+mod ciphernode_sequencer;
+mod plaintext_sequencer;
+mod publickey_sequencer;
+mod registry;
 
 // TODO: this is too permissive
 pub use actix::prelude::*;
@@ -28,17 +32,11 @@ pub use fhe::*;
 pub use logger::*;
 pub use p2p::*;
 pub use publickey_aggregator::*;
-
-pub use actix::prelude::*;
-pub use ciphernode::*;
+pub use publickey_sequencer::*;
+pub use plaintext_sequencer::*;
+pub use plaintext_aggregator::*;
 pub use ciphernode_selector::*;
-pub use ciphernode_supervisor::*;
-pub use data::*;
-pub use eventbus::*;
-pub use events::*;
-pub use fhe::*;
-pub use p2p::*;
-pub use publickey_aggregator::*;
+pub use ciphernode_sequencer::*;
 
 // TODO: move these out to a test folder
 #[cfg(test)]
@@ -129,7 +127,7 @@ mod tests {
     ) -> Result<(Addr<Fhe>, Arc<BfvParameters>, CommonRandomPoly)> {
         let (params, crp) = setup_bfv_params(&moduli, degree, plaintext_modulus, rng1)?;
         Ok((
-            Fhe::new(params.clone(), crp.clone(), rng2)?.start(),
+            Fhe::new(params.clone(), crp.clone(), rng2).start(),
             params,
             crp,
         ))

packages/ciphernode/core/src/plaintext_aggregator.rs

@@ -75,6 +75,16 @@ impl Actor for PlaintextAggregator {
     type Context = Context<Self>;
 }
 
+impl Handler<EnclaveEvent> for PlaintextAggregator {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
+        match msg {
+            EnclaveEvent::DecryptionshareCreated { data, .. } => ctx.notify(data),
+            _ => ()
+        }
+    }
+}
+
 impl Handler<DecryptionshareCreated> for PlaintextAggregator {
     type Result = Result<()>;
     fn handle(&mut self, event: DecryptionshareCreated, ctx: &mut Self::Context) -> Self::Result {
@@ -94,7 +104,7 @@ impl Handler<DecryptionshareCreated> for PlaintextAggregator {
 
         // Check the state and if it has changed to the computing
         if let PlaintextAggregatorState::Computing { shares } = &self.state {
-            ctx.address().do_send(ComputeAggregate {
+            ctx.notify(ComputeAggregate {
                 shares: shares.clone(),
             })
         }

packages/ciphernode/core/src/plaintext_sequencer.rs

@@ -0,0 +1,43 @@
+// sequence and persist events for a single E3 request in the correct order
+// TODO: spawn and store a ciphernode upon start and forward all events to it in order
+// TODO: if the ciphernode fails restart the node by replaying all stored events back to it
+
+use actix::prelude::*;
+
+use crate::{E3id, EnclaveEvent, EventBus, Fhe, PlaintextAggregator};
+
+pub struct PlaintextSequencer {
+    fhe: Addr<Fhe>,
+    e3_id: E3id,
+    bus: Addr<EventBus>,
+    nodecount: usize,
+    child: Option<Addr<PlaintextAggregator>>,
+}
+impl PlaintextSequencer {
+   pub fn new(fhe: Addr<Fhe>, e3_id: E3id, bus: Addr<EventBus>, nodecount: usize) -> Self {
+        Self {
+            fhe,
+            e3_id,
+            bus,
+            nodecount,
+            child: None,
+        }
+    }
+}
+impl Actor for PlaintextSequencer {
+    type Context = Context<Self>;
+}
+
+impl Handler<EnclaveEvent> for PlaintextSequencer {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
+        let fhe = self.fhe.clone();
+        let bus = self.bus.clone();
+        let nodecount = self.nodecount;
+        let e3_id = self.e3_id.clone();
+        let sink = self
+            .child
+            .get_or_insert_with(|| PlaintextAggregator::new(fhe, bus, e3_id, nodecount).start());
+        sink.do_send(msg);
+    }
+}

packages/ciphernode/core/src/publickey_aggregator.rs

@@ -91,6 +91,16 @@ impl Actor for PublicKeyAggregator {
     type Context = Context<Self>;
 }
 
+impl Handler<EnclaveEvent> for PublicKeyAggregator {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
+        match msg {
+            EnclaveEvent::KeyshareCreated { data, .. } => ctx.notify(data),
+            _ => ()
+        }
+    }
+}
+
 impl Handler<KeyshareCreated> for PublicKeyAggregator {
     type Result = Result<()>;
 
@@ -112,7 +122,7 @@ impl Handler<KeyshareCreated> for PublicKeyAggregator {
 
         // Check the state and if it has changed to the computing
         if let PublicKeyAggregatorState::Computing { keyshares } = &self.state {
-            ctx.address().do_send(ComputeAggregate {
+            ctx.notify(ComputeAggregate {
                 keyshares: keyshares.clone(),
             })
         }

packages/ciphernode/core/src/publickey_sequencer.rs

@@ -0,0 +1,45 @@
+// sequence and persist events for a single E3 request in the correct order
+// TODO: spawn and store a ciphernode upon start and forward all events to it in order
+// TODO: if the ciphernode fails restart the node by replaying all stored events back to it
+
+use actix::prelude::*;
+
+use crate::{E3id, EnclaveEvent, EventBus, Fhe, PublicKeyAggregator};
+
+pub struct PublicKeySequencer {
+    fhe: Addr<Fhe>,
+    e3_id: E3id,
+    bus: Addr<EventBus>,
+    nodecount: usize,
+    child: Option<Addr<PublicKeyAggregator>>,
+}
+
+impl PublicKeySequencer {
+    pub fn new(fhe: Addr<Fhe>, e3_id: E3id, bus: Addr<EventBus>, nodecount: usize) -> Self {
+        Self {
+            fhe,
+            e3_id,
+            bus,
+            nodecount,
+            child: None,
+        }
+    }
+}
+
+impl Actor for PublicKeySequencer {
+    type Context = Context<Self>;
+}
+
+impl Handler<EnclaveEvent> for PublicKeySequencer {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
+        let fhe = self.fhe.clone();
+        let bus = self.bus.clone();
+        let nodecount = self.nodecount;
+        let e3_id = self.e3_id.clone();
+        let sink = self
+            .child
+            .get_or_insert_with(|| PublicKeyAggregator::new(fhe, bus, e3_id, nodecount).start());
+        sink.do_send(msg);
+    }
+}