refactor(consensus): get rid of injection of notifications into the Affair socket

Author: bidzyyys

Cargo.lock

@@ -4533,6 +4533,7 @@ dependencies = [
  "lightning-application",
  "lightning-indexer",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-rep-collector",
  "lightning-signer",
  "lightning-test-utils",
@@ -4692,6 +4693,7 @@ dependencies = [
  "infusion",
  "lightning-application",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-signer",
  "lightning-test-utils",
  "queue-file",
@@ -4839,6 +4841,7 @@ dependencies = [
  "infusion",
  "lightning-application",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-rep-collector",
  "lightning-signer",
  "lightning-test-utils",
@@ -4980,6 +4983,7 @@ dependencies = [
  "lightning-blockstore",
  "lightning-indexer",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-origin-http",
  "lightning-origin-ipfs",
  "lightning-signer",
@@ -5009,6 +5013,7 @@ dependencies = [
  "lightning-blockstore",
  "lightning-indexer",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-signer",
  "lightning-test-utils",
  "reqwest",
@@ -5034,6 +5039,7 @@ dependencies = [
  "lightning-blockstore",
  "lightning-indexer",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-signer",
  "lightning-test-utils",
  "multihash 0.19.1",
@@ -5151,6 +5157,7 @@ dependencies = [
  "lightning-application",
  "lightning-broadcast",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-pool",
  "lightning-rep-collector",
  "lightning-signer",
@@ -5271,6 +5278,7 @@ dependencies = [
  "infusion",
  "lightning-application",
  "lightning-interfaces",
+ "lightning-notifier",
  "lightning-test-utils",
  "lightning-utils",
  "resolved-pathbuf",

core/blockstore/Cargo.toml

@@ -31,3 +31,4 @@ lightning-signer = { path="../signer" }
 lightning-application = { path="../application", features = ["test"] }
 lightning-rep-collector = { path="../rep-collector" }
 lightning-indexer = { path = "../indexer" }
+lightning-notifier = { path="../notifier" }

core/blockstore/src/lib.rs

@@ -25,10 +25,12 @@ mod tests {
         ConsensusInterface,
         IncrementalPutInterface,
         IndexerInterface,
+        NotifierInterface,
         SignerInterface,
         SyncQueryRunnerInterface,
         WithStartAndShutdown,
     };
+    use lightning_notifier::Notifier;
     use lightning_signer::{Config as SignerConfig, Signer};
     use lightning_test_utils::consensus::{Config as ConsensusConfig, MockConsensus};
     use tokio::test;
@@ -58,6 +60,7 @@ mod tests {
         SignerInterface = Signer<Self>;
         ConsensusInterface = MockConsensus<Self>;
         IndexerInterface = Indexer<Self>;
+        NotifierInterface = Notifier<Self>;
     });
 
     fn create_content() -> Vec<u8> {
@@ -169,6 +172,8 @@ mod tests {
 
         let mut signer = Signer::<TestBinding>::init(signer_config, query_runner.clone()).unwrap();
 
+        let notifier = Notifier::<TestBinding>::init(&app);
+
         let consensus_config = ConsensusConfig {
             min_ordering_time: 0,
             max_ordering_time: 1,
@@ -183,6 +188,7 @@ mod tests {
             query_runner.clone(),
             infusion::Blank::default(),
             None,
+            &notifier,
         )
         .unwrap();
 

core/consensus/src/consensus.rs

@@ -16,7 +16,9 @@ use lightning_interfaces::types::{Epoch, EpochInfo, UpdateMethod};
 use lightning_interfaces::{
     ApplicationInterface,
     BroadcastInterface,
+    EpochNotifierEmitter,
     IndexSocket,
+    NotifierInterface,
     PubSub,
     SyncQueryRunnerInterface,
 };
@@ -52,6 +54,7 @@ pub struct Consensus<C: Collection> {
             EpochState<
                 c![C::ApplicationInterface::SyncExecutor],
                 c![C::BroadcastInterface::PubSub<PubSubMsg>],
+                c![C::NotifierInterface::EpochEmitter],
             >,
         >,
     >,
@@ -71,7 +74,11 @@ pub struct Consensus<C: Collection> {
 }
 
 /// This struct contains mutable state only for the current epoch.
-struct EpochState<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static> {
+struct EpochState<
+    Q: SyncQueryRunnerInterface,
+    P: PubSub<PubSubMsg> + 'static,
+    EN: EpochNotifierEmitter,
+> {
     /// The node public key of the node.
     node_public_key: NodePublicKey,
     /// The consensus public key of the node.
@@ -85,7 +92,7 @@ struct EpochState<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static> {
     /// Path to the database used by the narwhal implementation
     pub store_path: ResolvedPathBuf,
     /// Narwhal execution state.
-    execution_state: Arc<Execution<Q>>,
+    execution_state: Arc<Execution<Q, EN>>,
     /// 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
@@ -100,14 +107,16 @@ struct EpochState<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static> {
 }
 
 #[allow(clippy::too_many_arguments)]
-impl<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static> EpochState<Q, P> {
+impl<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static, EN: EpochNotifierEmitter>
+    EpochState<Q, P, EN>
+{
     fn new(
         node_public_key: NodePublicKey,
         consensus_public_key: ConsensusPublicKey,
         query_runner: Q,
         narwhal_args: NarwhalArgs,
         store_path: ResolvedPathBuf,
-        execution_state: Arc<Execution<Q>>,
+        execution_state: Arc<Execution<Q, EN>>,
         txn_socket: SubmitTxSocket,
         pub_sub: P,
         rx_narwhal_batches: mpsc::Receiver<(AuthenticStampedParcel, bool)>,
@@ -384,6 +393,7 @@ impl<C: Collection> ConsensusInterface<C> for Consensus<C> {
         query_runner: c!(C::ApplicationInterface::SyncExecutor),
         pubsub: c!(C::BroadcastInterface::PubSub<Self::Certificate>),
         indexer_socket: Option<IndexSocket>,
+        notifier: &c!(C::NotifierInterface),
     ) -> anyhow::Result<Self> {
         // Spawn the registry for narwhal
         let registry = Registry::new();
@@ -413,6 +423,7 @@ impl<C: Collection> ConsensusInterface<C> for Consensus<C> {
             tx_narwhal_batches,
             query_runner.clone(),
             indexer_socket,
+            notifier.epoch_emitter(),
         ));
 
         let shutdown_notify = Arc::new(Notify::new());

core/consensus/src/edge_node/consensus.rs

@@ -4,7 +4,13 @@ use std::time::Duration;
 
 use fleek_crypto::NodePublicKey;
 use lightning_interfaces::types::Epoch;
-use lightning_interfaces::{BroadcastEventInterface, PubSub, SyncQueryRunnerInterface, ToDigest};
+use lightning_interfaces::{
+    BroadcastEventInterface,
+    EpochNotifierEmitter,
+    PubSub,
+    SyncQueryRunnerInterface,
+    ToDigest,
+};
 use lightning_utils::application::QueryRunnerExt;
 use quick_cache::unsync::Cache;
 use tokio::pin;
@@ -24,9 +30,13 @@ pub struct EdgeConsensus {
 }
 
 impl EdgeConsensus {
-    pub fn spawn<P: PubSub<PubSubMsg> + 'static, Q: SyncQueryRunnerInterface>(
+    pub fn spawn<
+        P: PubSub<PubSubMsg> + 'static,
+        Q: SyncQueryRunnerInterface,
+        EN: EpochNotifierEmitter,
+    >(
         pub_sub: P,
-        execution: Arc<Execution<Q>>,
+        execution: Arc<Execution<Q, EN>>,
         query_runner: Q,
         node_public_key: NodePublicKey,
         rx_narwhal_batches: mpsc::Receiver<(AuthenticStampedParcel, bool)>,
@@ -62,10 +72,14 @@ 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>(
+async fn message_receiver_worker<
+    P: PubSub<PubSubMsg>,
+    Q: SyncQueryRunnerInterface,
+    EN: EpochNotifierEmitter,
+>(
     mut pub_sub: P,
     shutdown_notify: Arc<Notify>,
-    execution: Arc<Execution<Q>>,
+    execution: Arc<Execution<Q, EN>>,
     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::SyncQueryRunnerInterface;
+use lightning_interfaces::{EpochNotifierEmitter, SyncQueryRunnerInterface};
 use lightning_utils::application::QueryRunnerExt;
 
 use super::ring_buffer::RingBuffer;
@@ -107,12 +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>(
+    pub async fn try_execute<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter>(
         &mut self,
         digest: Digest,
         threshold: usize,
         query_runner: &Q,
-        execution: &Arc<Execution<Q>>,
+        execution: &Arc<Execution<Q, EN>>,
     ) -> Result<bool, NotExecuted> {
         // get the current chain head
         let head = query_runner.get_last_block();
@@ -141,11 +141,11 @@ impl TransactionStore {
         Ok(epoch_changed)
     }
 
-    async fn try_execute_internal<Q: SyncQueryRunnerInterface>(
+    async fn try_execute_internal<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter>(
         &mut self,
         digest: Digest,
         threshold: usize,
-        execution: &Arc<Execution<Q>>,
+        execution: &Arc<Execution<Q, EN>>,
         head: Digest,
     ) -> Result<bool, NotExecuted> {
         if self.executed.contains(&digest) {
@@ -162,10 +162,10 @@ impl TransactionStore {
         Err(NotExecuted::MissingAttestations(digest))
     }
 
-    async fn try_execute_chain<Q: SyncQueryRunnerInterface>(
+    async fn try_execute_chain<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter>(
         &mut self,
         digest: Digest,
-        execution: &Arc<Execution<Q>>,
+        execution: &Arc<Execution<Q, EN>>,
         head: Digest,
     ) -> Result<bool, NotExecuted> {
         let mut txn_chain = VecDeque::new();

core/consensus/src/execution.rs

@@ -5,6 +5,7 @@ use fastcrypto::hash::HashFunction;
 use fleek_blake3 as blake3;
 use lightning_interfaces::types::{Block, Epoch, IndexRequest, NodeIndex, TransactionRequest};
 use lightning_interfaces::{
+    EpochNotifierEmitter,
     ExecutionEngineSocket,
     IndexSocket,
     SyncQueryRunnerInterface,
@@ -58,7 +59,7 @@ pub struct CommitteeAttestation {
     pub epoch: Epoch,
 }
 
-pub struct Execution<Q: SyncQueryRunnerInterface> {
+pub struct Execution<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter> {
     /// Managing certificates generated by narwhal.
     executor: ExecutionEngineSocket,
     /// Used to signal internal consensus processes that it is time to reconfigure for a new epoch
@@ -73,16 +74,18 @@ pub struct Execution<Q: SyncQueryRunnerInterface> {
     /// 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>,
+    epoch_change_notifier: EN,
 }
 
-impl<Q: SyncQueryRunnerInterface> Execution<Q> {
+impl<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter> Execution<Q, EN> {
     pub fn new(
         executor: ExecutionEngineSocket,
         reconfigure_notify: Arc<Notify>,
         new_block_notify: Arc<Notify>,
         tx_narwhal_batches: mpsc::Sender<(AuthenticStampedParcel, bool)>,
         query_runner: Q,
         index_socket: Option<IndexSocket>,
+        epoch_change_notifier: EN,
     ) -> Self {
         Self {
             executor,
@@ -91,6 +94,7 @@ impl<Q: SyncQueryRunnerInterface> Execution<Q> {
             tx_narwhal_batches,
             query_runner,
             index_socket,
+            epoch_change_notifier,
         }
     }
 
@@ -124,6 +128,7 @@ impl<Q: SyncQueryRunnerInterface> Execution<Q> {
 
         if results.change_epoch {
             change_epoch = true;
+            self.epoch_change_notifier.epoch_changed();
         }
 
         // If we have the archive socket that means our node is in archive node and we should send
@@ -147,7 +152,7 @@ impl<Q: SyncQueryRunnerInterface> Execution<Q> {
 }
 
 #[async_trait]
-impl<Q: SyncQueryRunnerInterface> ExecutionState for Execution<Q> {
+impl<Q: SyncQueryRunnerInterface, EN: EpochNotifierEmitter> ExecutionState for Execution<Q, EN> {
     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/dack-aggregator/Cargo.toml

@@ -6,7 +6,7 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-lightning-interfaces = {path="../interfaces"}
+lightning-interfaces = { path="../interfaces" }
 anyhow.workspace = true
 serde.workspace = true
 bincode.workspace = true
@@ -17,8 +17,9 @@ resolved-pathbuf.workspace = true
 queue-file = "1.4.10"
 
 [dev-dependencies]
-lightning-test-utils = {path="../test-utils"}
-lightning-signer = {path="../signer"}
-lightning-application = {path="../application", features = ["test"]}
+lightning-test-utils = { path="../test-utils" }
+lightning-signer = { path = "../signer" }
+lightning-application = { path = "../application", features = ["test"] }
+lightning-notifier = { path = "../notifier" }
 infusion.workspace = true
 fleek-crypto.workspace = true