refactor(notifier): apply CR changes

Author: bidzyyys

core/consensus/src/consensus.rs

@@ -15,9 +15,8 @@ use lightning_interfaces::signer::{SignerInterface, SubmitTxSocket};
 use lightning_interfaces::types::{Epoch, EpochInfo, UpdateMethod};
 use lightning_interfaces::{
     ApplicationInterface,
-    BlockNotifierEmitter,
     BroadcastInterface,
-    EpochNotifierEmitter,
+    Emitter,
     IndexSocket,
     NotifierInterface,
     PubSub,
@@ -55,8 +54,7 @@ pub struct Consensus<C: Collection> {
             EpochState<
                 c![C::ApplicationInterface::SyncExecutor],
                 c![C::BroadcastInterface::PubSub<PubSubMsg>],
-                c![C::NotifierInterface::EpochEmitter],
-                c![C::NotifierInterface::BlockEmitter],
+                c![C::NotifierInterface::Emitter],
             >,
         >,
     >,
@@ -74,12 +72,7 @@ pub struct Consensus<C: Collection> {
 }
 
 /// This struct contains mutable state only for the current epoch.
-struct EpochState<
-    Q: SyncQueryRunnerInterface,
-    P: PubSub<PubSubMsg> + 'static,
-    EN: EpochNotifierEmitter,
-    BN: BlockNotifierEmitter,
-> {
+struct EpochState<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static, NE: Emitter> {
     /// The node public key of the node.
     node_public_key: NodePublicKey,
     /// The consensus public key of the node.
@@ -93,7 +86,7 @@ struct EpochState<
     /// Path to the database used by the narwhal implementation
     pub store_path: ResolvedPathBuf,
     /// Narwhal execution state.
-    execution_state: Arc<Execution<Q, EN, BN>>,
+    execution_state: Arc<Execution<Q, NE>>,
     /// Used to send transactions to consensus
     /// We still use this socket on consensus struct because a node is not always on the committee,
     /// so its not always sending     a transaction to its own mempool. The signer interface
@@ -108,20 +101,16 @@ struct EpochState<
 }
 
 #[allow(clippy::too_many_arguments)]
-impl<
-    Q: SyncQueryRunnerInterface,
-    P: PubSub<PubSubMsg> + 'static,
-    EN: EpochNotifierEmitter,
-    BN: BlockNotifierEmitter,
-> EpochState<Q, P, EN, BN>
+impl<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static, NE: Emitter>
+    EpochState<Q, P, NE>
 {
     fn new(
         node_public_key: NodePublicKey,
         consensus_public_key: ConsensusPublicKey,
         query_runner: Q,
         narwhal_args: NarwhalArgs,
         store_path: ResolvedPathBuf,
-        execution_state: Arc<Execution<Q, EN, BN>>,
+        execution_state: Arc<Execution<Q, NE>>,
         txn_socket: SubmitTxSocket,
         pub_sub: P,
         rx_narwhal_batches: mpsc::Receiver<(AuthenticStampedParcel, bool)>,
@@ -426,8 +415,7 @@ impl<C: Collection> ConsensusInterface<C> for Consensus<C> {
             tx_narwhal_batches,
             query_runner.clone(),
             indexer_socket,
-            notifier.new_epoch_emitter(),
-            notifier.new_block_emitter(),
+            notifier.get_emitter(),
         ));
 
         let shutdown_notify = Arc::new(Notify::new());

core/consensus/src/edge_node/consensus.rs

@@ -5,9 +5,8 @@ use std::time::Duration;
 use fleek_crypto::NodePublicKey;
 use lightning_interfaces::types::Epoch;
 use lightning_interfaces::{
-    BlockNotifierEmitter,
     BroadcastEventInterface,
-    EpochNotifierEmitter,
+    Emitter,
     PubSub,
     SyncQueryRunnerInterface,
     ToDigest,
@@ -31,14 +30,9 @@ pub struct EdgeConsensus {
 }
 
 impl EdgeConsensus {
-    pub fn spawn<
-        P: PubSub<PubSubMsg> + 'static,
-        Q: SyncQueryRunnerInterface,
-        EN: EpochNotifierEmitter,
-        BN: BlockNotifierEmitter,
-    >(
+    pub fn spawn<P: PubSub<PubSubMsg> + 'static, Q: SyncQueryRunnerInterface, NE: Emitter>(
         pub_sub: P,
-        execution: Arc<Execution<Q, EN, BN>>,
+        execution: Arc<Execution<Q, NE>>,
         query_runner: Q,
         node_public_key: NodePublicKey,
         rx_narwhal_batches: mpsc::Receiver<(AuthenticStampedParcel, bool)>,
@@ -74,15 +68,10 @@ impl EdgeConsensus {
 
 /// Creates and event loop which consumes messages from pubsub and sends them to the
 /// right destination.
-async fn message_receiver_worker<
-    P: PubSub<PubSubMsg>,
-    Q: SyncQueryRunnerInterface,
-    EN: EpochNotifierEmitter,
-    BN: BlockNotifierEmitter,
->(
+async fn message_receiver_worker<P: PubSub<PubSubMsg>, Q: SyncQueryRunnerInterface, NE: Emitter>(
     mut pub_sub: P,
     shutdown_notify: Arc<Notify>,
-    execution: Arc<Execution<Q, EN, BN>>,
+    execution: Arc<Execution<Q, NE>>,
     query_runner: Q,
     node_public_key: NodePublicKey,
     mut rx_narwhal_batch: mpsc::Receiver<(AuthenticStampedParcel, bool)>,

core/consensus/src/edge_node/transaction_store.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use std::time::{Duration, SystemTime};
 
 use lightning_interfaces::types::{Digest as BroadcastDigest, NodeIndex};
-use lightning_interfaces::{BlockNotifierEmitter, EpochNotifierEmitter, SyncQueryRunnerInterface};
+use lightning_interfaces::{Emitter, SyncQueryRunnerInterface};
 use lightning_utils::application::QueryRunnerExt;
 
 use super::ring_buffer::RingBuffer;
@@ -107,16 +107,12 @@ impl TransactionStore {
 
     // Threshold should be 2f + 1 of the committee
     // Returns true if the epoch has changed
-    pub async fn try_execute<
-        Q: SyncQueryRunnerInterface,
-        EN: EpochNotifierEmitter,
-        BN: BlockNotifierEmitter,
-    >(
+    pub async fn try_execute<Q: SyncQueryRunnerInterface, NE: Emitter>(
         &mut self,
         digest: Digest,
         threshold: usize,
         query_runner: &Q,
-        execution: &Arc<Execution<Q, EN, BN>>,
+        execution: &Arc<Execution<Q, NE>>,
     ) -> Result<bool, NotExecuted> {
         // get the current chain head
         let head = query_runner.get_last_block();
@@ -145,15 +141,11 @@ impl TransactionStore {
         Ok(epoch_changed)
     }
 
-    async fn try_execute_internal<
-        Q: SyncQueryRunnerInterface,
-        EN: EpochNotifierEmitter,
-        BN: BlockNotifierEmitter,
-    >(
+    async fn try_execute_internal<Q: SyncQueryRunnerInterface, NE: Emitter>(
         &mut self,
         digest: Digest,
         threshold: usize,
-        execution: &Arc<Execution<Q, EN, BN>>,
+        execution: &Arc<Execution<Q, NE>>,
         head: Digest,
     ) -> Result<bool, NotExecuted> {
         if self.executed.contains(&digest) {
@@ -170,14 +162,10 @@ impl TransactionStore {
         Err(NotExecuted::MissingAttestations(digest))
     }
 
-    async fn try_execute_chain<
-        Q: SyncQueryRunnerInterface,
-        EN: EpochNotifierEmitter,
-        BN: BlockNotifierEmitter,
-    >(
+    async fn try_execute_chain<Q: SyncQueryRunnerInterface, NE: Emitter>(
         &mut self,
         digest: Digest,
-        execution: &Arc<Execution<Q, EN, BN>>,
+        execution: &Arc<Execution<Q, NE>>,
         head: Digest,
     ) -> Result<bool, NotExecuted> {
         let mut txn_chain = VecDeque::new();

core/consensus/src/execution.rs

@@ -5,8 +5,7 @@ use fastcrypto::hash::HashFunction;
 use fleek_blake3 as blake3;
 use lightning_interfaces::types::{Block, Epoch, IndexRequest, NodeIndex, TransactionRequest};
 use lightning_interfaces::{
-    BlockNotifierEmitter,
-    EpochNotifierEmitter,
+    Emitter,
     ExecutionEngineSocket,
     IndexSocket,
     SyncQueryRunnerInterface,
@@ -60,11 +59,7 @@ pub struct CommitteeAttestation {
     pub epoch: Epoch,
 }
 
-pub struct Execution<
-    Q: SyncQueryRunnerInterface,
-    EN: EpochNotifierEmitter,
-    BN: BlockNotifierEmitter,
-> {
+pub struct Execution<Q: SyncQueryRunnerInterface, NE: Emitter> {
     /// Managing certificates generated by narwhal.
     executor: ExecutionEngineSocket,
     /// Used to signal internal consensus processes that it is time to reconfigure for a new epoch
@@ -77,32 +72,26 @@ pub struct Execution<
     /// If this socket is present it means the node is in archive node and should send all blocks
     /// and transactions it executes to the archiver to be indexed
     index_socket: Option<IndexSocket>,
-    /// Notifier that notifies every time an epoch is changed
-    epoch_change_notifier: EN,
-    /// Notifier that notifies every time a block is executed on application state
-    new_block_notifier: BN,
+    /// Notifications emitter
+    notifier: NE,
 }
 
-impl<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter, BN: BlockNotifierEmitter>
-    Execution<Q, EN, BN>
-{
+impl<Q: SyncQueryRunnerInterface, NE: Emitter> Execution<Q, NE> {
     pub fn new(
         executor: ExecutionEngineSocket,
         reconfigure_notify: Arc<Notify>,
         tx_narwhal_batches: mpsc::Sender<(AuthenticStampedParcel, bool)>,
         query_runner: Q,
         index_socket: Option<IndexSocket>,
-        epoch_change_notifier: EN,
-        new_block_notifier: BN,
+        notifier: NE,
     ) -> Self {
         Self {
             executor,
             reconfigure_notify,
             tx_narwhal_batches,
             query_runner,
             index_socket,
-            epoch_change_notifier,
-            new_block_notifier,
+            notifier,
         }
     }
 
@@ -136,7 +125,7 @@ impl<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter, BN: BlockNotifierEmi
 
         if results.change_epoch {
             change_epoch = true;
-            self.epoch_change_notifier.epoch_changed();
+            self.notifier.epoch_changed();
         }
 
         // If we have the archive socket that means our node is in archive node and we should send
@@ -152,16 +141,14 @@ impl<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter, BN: BlockNotifierEmi
                 error!("We could not send a message to the archiver: {e}");
             }
         }
-        self.new_block_notifier.new_block();
+        self.notifier.new_block();
 
         change_epoch
     }
 }
 
 #[async_trait]
-impl<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter, BN: BlockNotifierEmitter> ExecutionState
-    for Execution<Q, EN, BN>
-{
+impl<Q: SyncQueryRunnerInterface, NE: Emitter> ExecutionState for Execution<Q, NE> {
     async fn handle_consensus_output(&self, consensus_output: ConsensusOutput) {
         for (cert, batches) in consensus_output.batches {
             let current_epoch = self.query_runner.get_current_epoch();

core/interfaces/src/notifier.rs

@@ -15,38 +15,27 @@ pub enum Notification {
 
 #[infusion::service]
 pub trait NotifierInterface<C: Collection>: Sync + Send + Clone {
-    type EpochEmitter: EpochNotifierEmitter;
-    type BlockEmitter: BlockNotifierEmitter;
+    type Emitter: Emitter;
 
     fn _init(app: ::ApplicationInterface) {
         ok!(Self::init(app))
     }
-
     fn init(app: &c!(C::ApplicationInterface)) -> Self;
-
-    /// Returns a reference to the emitter end of this notifier. Should only be used if we are
-    /// interested (and responsible) for triggering a notification around new block.
-    fn new_block_emitter(&self) -> Self::BlockEmitter;
-
     /// Returns a reference to the emitter end of this notifier. Should only be used if we are
     /// interested (and responsible) for triggering a notification around new epoch.
-    fn new_epoch_emitter(&self) -> Self::EpochEmitter;
+    fn get_emitter(&self) -> Self::Emitter;
 
     fn notify_on_new_block(&self, tx: mpsc::Sender<Notification>);
 
     fn notify_on_new_epoch(&self, tx: mpsc::Sender<Notification>);
 
     fn notify_before_epoch_change(&self, duration: Duration, tx: mpsc::Sender<Notification>);
 }
-
 #[infusion::blank]
-pub trait EpochNotifierEmitter: Clone + Send + Sync + 'static {
+pub trait Emitter: Clone + Send + Sync + 'static {
     /// Notify the waiters about epoch change.
     fn epoch_changed(&self);
-}
 
-#[infusion::blank]
-pub trait BlockNotifierEmitter: Clone + Send + Sync + 'static {
     /// Notify the waiters about new block.
     fn new_block(&self);
 }

core/mock/src/consensus.rs

@@ -12,10 +12,10 @@ use lightning_interfaces::infu_collection::{c, Collection};
 use lightning_interfaces::types::{Block, TransactionRequest};
 use lightning_interfaces::{
     ApplicationInterface,
-    BlockNotifierEmitter,
     BroadcastInterface,
     ConfigConsumer,
     ConsensusInterface,
+    Emitter,
     ExecutionEngineSocket,
     IndexSocket,
     MempoolSocket,
@@ -160,11 +160,7 @@ impl<C: Collection> ConsensusInterface<C> for MockConsensus<C> {
         notifier: &c!(C::NotifierInterface),
     ) -> anyhow::Result<Self> {
         let (tx, rx) = mpsc::channel(128);
-        tokio::spawn(run_consensus::<C>(
-            rx,
-            executor,
-            notifier.new_block_emitter(),
-        ));
+        tokio::spawn(run_consensus::<C>(rx, executor, notifier.get_emitter()));
 
         let mempool = TokioSpawn::spawn_async(MempoolSocketWorker {
             sender: tx.clone(),
@@ -199,7 +195,7 @@ impl<C: Collection> ConfigConsumer for MockConsensus<C> {
 async fn run_consensus<C: Collection>(
     mut rx: mpsc::Receiver<TransactionRequest>,
     exec: ExecutionEngineSocket,
-    new_block_notifier: c!(C::NotifierInterface::BlockEmitter),
+    notifier: c!(C::NotifierInterface::Emitter),
 ) {
     let mut ticker = tokio::time::interval(Duration::from_millis(1));
     let mut buffer = Vec::<TransactionRequest>::with_capacity(32);
@@ -219,7 +215,7 @@ async fn run_consensus<C: Collection>(
 
                 let _ = exec.run(block).await;
 
-                new_block_notifier.new_block();
+                notifier.new_block();
             }
             tmp = rx.recv() => {
                 if let Some(tx) = tmp {