feat: ciphernode economic contracts [skip-line-limit]

.gitmodules

@@ -3,4 +3,4 @@
 	url = https://github.com/risc0/risc0-ethereum
 [submodule "templates/default/lib/risc0-ethereum"]
 	path = templates/default/lib/risc0-ethereum
-  url = https://github.com/gnosisguild/risc0-ethereum
+  	url = https://github.com/gnosisguild/risc0-ethereum

README.md

@@ -156,7 +156,7 @@ sequenceDiagram
     E3Program-->>Enclave: inputValidator
     Enclave->>ComputeProvider: validate(computeProviderParams)
     ComputeProvider-->>Enclave: decryptionVerifier
-    Enclave->>CiphernodeRegistry: requestCommittee(e3Id, filter, threshold)
+    Enclave->>CiphernodeRegistry: requestCommittee(e3Id, seed, threshold)
     CiphernodeRegistry-->>Enclave: success
     Enclave-->>Users: e3Id, E3 struct
 

crates/aggregator/Cargo.toml

@@ -14,6 +14,7 @@ bincode = { workspace = true }
 e3-config = { workspace = true }
 e3-data = { workspace = true }
 e3-events = { workspace = true }
+e3-evm = { workspace = true }
 e3-fhe = { workspace = true }
 e3-multithread = { workspace = true }
 e3-trbfv = { workspace = true }

crates/aggregator/src/committee_finalizer.rs

@@ -0,0 +1,148 @@
+// SPDX-License-Identifier: LGPL-3.0-only
+//
+// This file is provided WITHOUT ANY WARRANTY;
+// without even the implied warranty of MERCHANTABILITY
+// or FITNESS FOR A PARTICULAR PURPOSE.
+
+use actix::prelude::*;
+use e3_events::{
+    CommitteeFinalizeRequested, CommitteeRequested, EnclaveEvent, EventBus, Shutdown, Subscribe,
+};
+use std::collections::HashMap;
+use std::time::Duration;
+use tracing::{error, info};
+
+/// CommitteeFinalizer is an actor that listens to CommitteeRequested events and dispatches
+/// CommitteeFinalizeRequested events after the submission deadline has passed.
+pub struct CommitteeFinalizer {
+    bus: Addr<EventBus<EnclaveEvent>>,
+    pending_committees: HashMap<String, SpawnHandle>,
+}
+
+impl CommitteeFinalizer {
+    pub fn new(bus: &Addr<EventBus<EnclaveEvent>>) -> Self {
+        Self {
+            bus: bus.clone(),
+            pending_committees: HashMap::new(),
+        }
+    }
+
+    pub fn attach(bus: &Addr<EventBus<EnclaveEvent>>) -> Addr<Self> {
+        let addr = CommitteeFinalizer::new(bus).start();
+
+        bus.do_send(Subscribe::new(
+            "CommitteeRequested",
+            addr.clone().recipient(),
+        ));
+        bus.do_send(Subscribe::new("Shutdown", addr.clone().recipient()));
+
+        addr
+    }
+}
+
+impl Actor for CommitteeFinalizer {
+    type Context = Context<Self>;
+}
+
+impl Handler<EnclaveEvent> for CommitteeFinalizer {
+    type Result = ();
+    fn handle(&mut self, msg: EnclaveEvent, ctx: &mut Self::Context) -> Self::Result {
+        match msg {
+            EnclaveEvent::CommitteeRequested { data, .. } => ctx.notify(data),
+            EnclaveEvent::Shutdown { data, .. } => ctx.notify(data),
+            _ => (),
+        }
+    }
+}
+
+impl Handler<CommitteeRequested> for CommitteeFinalizer {
+    type Result = ();
+
+    fn handle(&mut self, msg: CommitteeRequested, ctx: &mut Self::Context) -> Self::Result {
+        let e3_id = msg.e3_id.clone();
+        let submission_deadline = msg.submission_deadline;
+
+        const FINALIZATION_BUFFER_SECONDS: u64 = 1;
+
+        let e3_id_for_log = e3_id.clone();
+        let fut = async move {
+            match e3_evm::helpers::get_current_timestamp().await {
+                Ok(timestamp) => Some(timestamp),
+                Err(e) => {
+                    error!(
+                        e3_id = %e3_id_for_log,
+                        error = %e,
+                        "Failed to get current timestamp from RPC"
+                    );
+                    None
+                }
+            }
+        };
+
+        let e3_id_for_async = e3_id;
+        ctx.spawn(
+            fut.into_actor(self)
+                .then(move |current_timestamp, act, ctx| {
+                    if let Some(current_timestamp) = current_timestamp {
+                        let seconds_until_deadline = if submission_deadline > current_timestamp {
+                            (submission_deadline - current_timestamp) + FINALIZATION_BUFFER_SECONDS
+                        } else {
+                            info!(
+                                e3_id = %e3_id_for_async,
+                                submission_deadline = submission_deadline,
+                                current_timestamp = current_timestamp,
+                                "Submission deadline already passed, finalizing with buffer"
+                            );
+                            FINALIZATION_BUFFER_SECONDS
+                        };
+
+                        info!(
+                            e3_id = %e3_id_for_async,
+                            submission_deadline = submission_deadline,
+                            current_timestamp = current_timestamp,
+                            seconds_to_wait = seconds_until_deadline,
+                            "Scheduling committee finalization"
+                        );
+
+                        let bus = act.bus.clone();
+                        let e3_id_clone = e3_id_for_async.clone();
+
+                        let handle = ctx.run_later(
+                            Duration::from_secs(seconds_until_deadline),
+                            move |act, _ctx| {
+                                info!(e3_id = %e3_id_clone, "Dispatching CommitteeFinalizeRequested event");
+
+                                bus.do_send(EnclaveEvent::from(CommitteeFinalizeRequested {
+                                    e3_id: e3_id_clone.clone(),
+                                }));
+
+                                act.pending_committees.remove(&e3_id_clone.to_string());
+                            },
+                        );
+
+                        act.pending_committees
+                            .insert(e3_id_for_async.to_string(), handle);
+                    } else {
+                        error!(
+                            e3_id = %e3_id_for_async,
+                            "Skipping committee finalization due to timestamp fetch failure"
+                        );
+                    }
+
+                    async {}.into_actor(act)
+                }),
+        );
+    }
+}
+
+impl Handler<Shutdown> for CommitteeFinalizer {
+    type Result = ();
+    fn handle(&mut self, _msg: Shutdown, ctx: &mut Self::Context) -> Self::Result {
+        info!("Killing CommitteeFinalizer");
+        // Cancel all pending finalization tasks
+        for (_, handle) in self.pending_committees.drain() {
+            ctx.cancel_future(handle);
+        }
+        ctx.stop();
+    }
+}

crates/aggregator/src/lib.rs

@@ -4,11 +4,13 @@
 // without even the implied warranty of MERCHANTABILITY
 // or FITNESS FOR A PARTICULAR PURPOSE.
 
+mod committee_finalizer;
 pub mod ext;
 mod plaintext_aggregator;
 mod publickey_aggregator;
 mod repo;
 mod threshold_plaintext_aggregator;
+pub use committee_finalizer::CommitteeFinalizer;
 pub use plaintext_aggregator::{
     PlaintextAggregator, PlaintextAggregatorParams, PlaintextAggregatorState,
 };

crates/aggregator/src/publickey_aggregator.rs

@@ -11,7 +11,7 @@ use e3_events::{
     Die, E3id, EnclaveEvent, EventBus, KeyshareCreated, OrderedSet, PublicKeyAggregated, Seed,
 };
 use e3_fhe::{Fhe, GetAggregatePublicKey};
-use e3_sortition::{GetNodeIndex, GetNodes, Sortition};
+use e3_sortition::{GetNodesForE3, Sortition};
 use e3_utils::ArcBytes;
 use std::sync::Arc;
 use tracing::{error, trace};
@@ -148,47 +148,32 @@ impl Handler<KeyshareCreated> for PublicKeyAggregator {
     type Result = ResponseActFuture<Self, Result<()>>;
 
     fn handle(&mut self, event: KeyshareCreated, _: &mut Self::Context) -> Self::Result {
-        let Some(PublicKeyAggregatorState::Collecting {
-            threshold_n, seed, ..
-        }) = self.state.get()
-        else {
-            error!(state=?self.state, "Aggregator has been closed for collecting keyshares.");
-            return Box::pin(fut::ready(Ok(())));
-        };
-
-        let size = threshold_n;
-        let address = event.node;
-        let chain_id = event.e3_id.chain_id();
+        let address = event.node.clone();
         let e3_id = event.e3_id.clone();
         let pubkey = event.pubkey.clone();
 
         Box::pin(
             self.sortition
-                .send(GetNodeIndex {
-                    chain_id,
-                    address,
-                    size,
-                    seed,
+                .send(GetNodesForE3 {
+                    e3_id: e3_id.clone(),
+                    chain_id: e3_id.chain_id(),
                 })
                 .into_actor(self)
                 .map(move |res, act, ctx| {
-                    // NOTE: Returning Ok(()) on errors as we probably dont need a result type here since
-                    // we will not be doing a send
-                    let maybe_found_index = res?;
-                    let Some(_) = maybe_found_index else {
-                        trace!("Node not found in committee");
+                    let nodes = res?;
+
+                    if !nodes.contains(&address) {
+                        trace!("Node {} not found in finalized committee", address);
                         return Ok(());
-                    };
+                    }
 
                     if e3_id != act.e3_id {
                         error!("Wrong e3_id sent to aggregator. This should not happen.");
                         return Ok(());
                     }
 
-                    // add the keyshare and
                     act.add_keyshare(pubkey)?;
 
-                    // Check the state and if it has changed to the computing
                     if let Some(PublicKeyAggregatorState::Computing { keyshares }) =
                         &act.state.get()
                     {
@@ -228,7 +213,8 @@ impl Handler<NotifyNetwork> for PublicKeyAggregator {
     fn handle(&mut self, msg: NotifyNetwork, _: &mut Self::Context) -> Self::Result {
         Box::pin(
             self.sortition
-                .send(GetNodes {
+                .send(GetNodesForE3 {
+                    e3_id: msg.e3_id.clone(),
                     chain_id: msg.e3_id.chain_id(),
                 })
                 .into_actor(self)

crates/aggregator/src/threshold_plaintext_aggregator.rs

@@ -14,15 +14,15 @@ use e3_events::{
     Seed,
 };
 use e3_multithread::Multithread;
-use e3_sortition::{GetNodeIndex, Sortition};
+use e3_sortition::{GetNodesForE3, Sortition};
 use e3_trbfv::{
     calculate_threshold_decryption::{
         CalculateThresholdDecryptionRequest, CalculateThresholdDecryptionResponse,
     },
     TrBFVConfig, TrBFVRequest,
 };
 use e3_utils::utility_types::ArcBytes;
-use tracing::{error, info};
+use tracing::{error, info, trace};
 
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 pub struct Collecting {
@@ -248,46 +248,23 @@ impl Handler<DecryptionshareCreated> for ThresholdPlaintextAggregator {
 
     fn handle(&mut self, event: DecryptionshareCreated, _: &mut Self::Context) -> Self::Result {
         info!(event=?event, "Processing DecryptionShareCreated...");
-        let Some(ThresholdPlaintextAggregatorState::Collecting(Collecting {
-            threshold_n,
-            seed,
-            ..
-        })) = self.state.get()
-        else {
-            error!(state=?self.state, "Aggregator has been closed for collecting.");
-            return Box::pin(fut::ready(Ok(())));
-        };
-
-        let size = threshold_n as usize;
-        let address = event.node;
+        let address = event.node.clone();
         let party_id = event.party_id;
-        let chain_id = event.e3_id.chain_id();
         let e3_id = event.e3_id.clone();
         let decryption_share = event.decryption_share.clone();
 
-        // Why do we need to get the node index when the event contains the party_id? I guess we
-        // don't trust the event. Maybe that is fine.
         Box::pin(
             self.sortition
-                .send(GetNodeIndex {
-                    chain_id,
-                    address: address.clone(),
-                    size,
-                    seed,
+                .send(GetNodesForE3 {
+                    e3_id: e3_id.clone(),
+                    chain_id: e3_id.chain_id(),
                 })
                 .into_actor(self)
                 .map(move |res, act, ctx| {
-                    let maybe_found_index = res?;
-                    let Some(party) = maybe_found_index else {
-                        error!("Attempting to aggregate share but party not found in committee");
-                        return Ok(());
-                    };
+                    let nodes = res?;
 
-                    if party != party_id {
-                        error!(
-                            "Bad aggregation state! Address {} not found at index {} instead it was found at {}",
-                            address, party_id, party
-                        );
+                    if !nodes.contains(&address) {
+                        trace!("Node {} not found in finalized committee", address);
                         return Ok(());
                     }
 
@@ -296,10 +273,10 @@ impl Handler<DecryptionshareCreated> for ThresholdPlaintextAggregator {
                         return Ok(());
                     }
 
-                    // add the keyshare and
+                    // Trust the party_id from the event - it's based on CommitteeFinalized order
+                    // which is the authoritative source of truth for party IDs
                     act.add_share(party_id, decryption_share)?;
 
-                    // Check the state and if it has changed to the computing
                     if let Some(ThresholdPlaintextAggregatorState::Computing(Computing {
                         threshold_m,
                         threshold_n,