feat: fault attribution and slashing crates [skip-line-limit]

crates/aggregator/src/committee_finalizer.rs

@@ -6,8 +6,8 @@
 
 use actix::prelude::*;
 use e3_events::{
-    prelude::*, trap, BusHandle, CommitteeFinalizeRequested, CommitteeRequested, EType,
-    EnclaveEvent, EnclaveEventData, EventType, Shutdown, TypedEvent,
+    prelude::*, trap, BusHandle, CommitteeFinalizeRequested, CommitteeRequested, E3Failed, E3Stage,
+    E3StageChanged, EType, EnclaveEvent, EnclaveEventData, EventType, Shutdown, TypedEvent,
 };
 use e3_utils::{NotifySync, MAILBOX_LIMIT};
 use std::collections::HashMap;
@@ -33,7 +33,12 @@ impl CommitteeFinalizer {
         let addr = CommitteeFinalizer::new(bus).start();
 
         bus.subscribe_all(
-            &[EventType::CommitteeRequested, EventType::Shutdown],
+            &[
+                EventType::CommitteeRequested,
+                EventType::Shutdown,
+                EventType::E3Failed,
+                EventType::E3StageChanged,
+            ],
             addr.clone().recipient(),
         );
 
@@ -57,6 +62,10 @@ impl Handler<EnclaveEvent> for CommitteeFinalizer {
                 self.notify_sync(ctx, TypedEvent::new(data, ec))
             }
             EnclaveEventData::Shutdown(data) => self.notify_sync(ctx, data),
+            EnclaveEventData::E3Failed(data) => self.notify_sync(ctx, TypedEvent::new(data, ec)),
+            EnclaveEventData::E3StageChanged(data) => {
+                self.notify_sync(ctx, TypedEvent::new(data, ec))
+            }
             _ => (),
         }
     }
@@ -166,3 +175,38 @@ impl Handler<Shutdown> for CommitteeFinalizer {
         ctx.stop();
     }
 }
+
+impl Handler<TypedEvent<E3Failed>> for CommitteeFinalizer {
+    type Result = ();
+    fn handle(&mut self, msg: TypedEvent<E3Failed>, ctx: &mut Self::Context) -> Self::Result {
+        let e3_id_str = msg.e3_id.to_string();
+        if let Some(handle) = self.pending_committees.remove(&e3_id_str) {
+            info!(
+                e3_id = %msg.e3_id,
+                reason = ?msg.reason,
+                "E3 failed — cancelling pending committee finalization timer"
+            );
+            ctx.cancel_future(handle);
+        }
+    }
+}
+
+impl Handler<TypedEvent<E3StageChanged>> for CommitteeFinalizer {
+    type Result = ();
+    fn handle(&mut self, msg: TypedEvent<E3StageChanged>, ctx: &mut Self::Context) -> Self::Result {
+        match &msg.new_stage {
+            E3Stage::Complete | E3Stage::Failed => {
+                let e3_id_str = msg.e3_id.to_string();
+                if let Some(handle) = self.pending_committees.remove(&e3_id_str) {
+                    info!(
+                        e3_id = %msg.e3_id,
+                        stage = ?msg.new_stage,
+                        "E3 reached terminal stage — cancelling pending committee finalization timer"
+                    );
+                    ctx.cancel_future(handle);
+                }
+            }
+            _ => {}
+        }
+    }
+}

crates/aggregator/src/keyshare_created_filter_buffer.rs

@@ -6,17 +6,21 @@
 
 use actix::prelude::*;
 
-use e3_events::{prelude::*, EnclaveEvent, EnclaveEventData};
+use e3_events::{prelude::*, Die, EnclaveEvent, EnclaveEventData};
 use e3_utils::MAILBOX_LIMIT;
 use std::collections::HashSet;
+use tracing::info;
 
 use crate::PublicKeyAggregator;
 
-/// Buffer KeyshareCreated events until CommitteeFinalized has been published
+/// Buffers `KeyshareCreated` events until `CommitteeFinalized` arrives.
 pub struct KeyshareCreatedFilterBuffer {
     dest: Addr<PublicKeyAggregator>,
     committee: Option<HashSet<String>>,
     buffer: Vec<EnclaveEvent>,
+    /// Nodes expelled before CommitteeFinalized arrived.
+    /// Tracked separately so they are filtered during `process_buffered_events()`.
+    expelled_before_finalization: HashSet<String>,
 }
 
 impl KeyshareCreatedFilterBuffer {
@@ -25,14 +29,17 @@ impl KeyshareCreatedFilterBuffer {
             dest,
             committee: None,
             buffer: Vec::new(),
+            expelled_before_finalization: HashSet::new(),
         }
     }
 
     fn process_buffered_events(&mut self) {
         if let Some(ref committee) = self.committee {
             for event in self.buffer.drain(..) {
                 if let EnclaveEventData::KeyshareCreated(data) = event.get_data() {
-                    if committee.contains(&data.node) {
+                    if committee.contains(&data.node)
+                        && !self.expelled_before_finalization.contains(&data.node)
+                    {
                         self.dest.do_send(event);
                     }
                 }
@@ -65,14 +72,46 @@ impl Handler<EnclaveEvent> for KeyshareCreatedFilterBuffer {
                 _ => {}
             },
             EnclaveEventData::CommitteeFinalized(data) => {
-                self.dest.do_send(msg.clone()); // forward committee first
+                self.dest.do_send(msg.clone());
                 self.committee = Some(data.committee.iter().cloned().collect());
                 self.process_buffered_events();
             }
+            EnclaveEventData::CommitteeMemberExpelled(data) => {
+                // Only process raw events from chain (party_id not yet resolved).
+                if data.party_id.is_some() {
+                    return;
+                }
+
+                let node_addr = data.node.to_string();
+
+                // Remove expelled node so we don't forward late KeyshareCreated events from them
+                if let Some(ref mut committee) = self.committee {
+                    info!(
+                        "KeyshareCreatedFilterBuffer: removing expelled node {} from committee filter (e3_id={})",
+                        node_addr, data.e3_id
+                    );
+                    committee.remove(&node_addr);
+                } else {
+                    // Committee not yet finalized — track for filtering during process_buffered_events
+                    info!(
+                        "KeyshareCreatedFilterBuffer: tracking expelled node {} before finalization (e3_id={})",
+                        node_addr, data.e3_id
+                    );
+                    self.expelled_before_finalization.insert(node_addr);
+                }
+                // Forward to PublicKeyAggregator for threshold_n adjustment
+                self.dest.do_send(msg);
+            }
             _ => {
-                // forward all other events
                 self.dest.do_send(msg);
             }
         }
     }
 }
+
+impl Handler<Die> for KeyshareCreatedFilterBuffer {
+    type Result = ();
+    fn handle(&mut self, _: Die, ctx: &mut Self::Context) -> Self::Result {
+        ctx.stop();
+    }
+}