chore(application): take ExecutionEngineSocket instead of cloning it

Author: bidzyyys

core/application/src/app.rs

@@ -1,4 +1,5 @@
 use std::marker::PhantomData;
+use std::sync::{Arc, Mutex};
 
 use affair::{Executor, TokioSpawn};
 use anyhow::{anyhow, Result};
@@ -15,7 +16,7 @@ use crate::config::{Config, StorageConfig};
 use crate::env::{Env, UpdateWorker};
 use crate::query_runner::QueryRunner;
 pub struct Application<C: Collection> {
-    update_socket: ExecutionEngineSocket,
+    update_socket: Arc<Mutex<Option<ExecutionEngineSocket>>>,
     query_runner: QueryRunner,
     collection: PhantomData<C>,
 }
@@ -63,7 +64,9 @@ impl<C: Collection> ApplicationInterface<C> for Application<C> {
 
         Ok(Self {
             query_runner: env.query_runner(),
-            update_socket: TokioSpawn::spawn_async(UpdateWorker::<C>::new(env, blockstore)),
+            update_socket: Arc::new(Mutex::new(Some(TokioSpawn::spawn_async(
+                UpdateWorker::<C>::new(env, blockstore),
+            )))),
             collection: PhantomData,
         })
     }
@@ -74,8 +77,8 @@ impl<C: Collection> ApplicationInterface<C> for Application<C> {
     /// # Safety
     ///
     /// See the safety document for the [`ExecutionEngineSocket`].
-    fn transaction_executor(&self) -> ExecutionEngineSocket {
-        self.update_socket.clone()
+    fn transaction_executor(&self) -> Option<ExecutionEngineSocket> {
+        self.update_socket.lock().unwrap().take()
     }
 
     /// Returns the instance of a sync query runner which can be used to run queries without

core/application/src/tests.rs

@@ -570,7 +570,7 @@ fn init_app(config: Option<Config>) -> (ExecutionEngineSocket, QueryRunner) {
 fn do_init_app(config: Config) -> (ExecutionEngineSocket, QueryRunner) {
     let app = Application::<TestBinding>::init(config, Default::default()).unwrap();
 
-    (app.transaction_executor(), app.sync_query())
+    (app.transaction_executor().unwrap(), app.sync_query())
 }
 
 /// Initialize application with provided committee.

core/blockstore/src/lib.rs

@@ -179,7 +179,7 @@ mod tests {
         let consensus = MockConsensus::<TestBinding>::init(
             consensus_config,
             &signer,
-            update_socket,
+            update_socket.unwrap(),
             query_runner.clone(),
             infusion::Blank::default(),
             None,

core/consensus/src/consensus.rs

@@ -44,7 +44,7 @@ use crate::narwhal::{NarwhalArgs, NarwhalService};
 
 pub struct Consensus<C: Collection> {
     /// Inner state of the consensus
-    /// todo(dalton): We can probably get a little more effecient then a mutex here
+    /// todo(dalton): We can probably get a little more efficient then a mutex here
     /// maybe a once box
     #[allow(clippy::type_complexity)]
     epoch_state: Mutex<
@@ -55,13 +55,13 @@ pub struct Consensus<C: Collection> {
             >,
         >,
     >,
-    /// This socket recieves signed transactions and forwards them to an active committee member to
+    /// This socket receives signed transactions and forwards them to an active committee member to
     /// be ordered
     mempool_socket: MempoolSocket,
     /// Timestamp of the narwhal certificate that caused an epoch change
     /// is sent through this channel to notify that epoch chould change.
     reconfigure_notify: Arc<Notify>,
-    /// A notifier that is notified everytime a new block is proccessed
+    /// A notifier that is notified every time a new block is processed
     new_block_notify: Arc<Notify>,
     /// Called from the shutdown function to notify the start event loop to
     /// exit.
@@ -93,7 +93,7 @@ struct EpochState<Q: SyncQueryRunnerInterface, P: PubSub<PubSubMsg> + 'static> {
     txn_socket: SubmitTxSocket,
     /// Interface for sending messages through the gossip layer
     pub_sub: P,
-    /// Narhwal sends payloads ready for broadcast to this reciever
+    /// Narhwal sends payloads ready for broadcast to this receiver
     rx_narwhal_batches: Option<mpsc::Receiver<(AuthenticStampedParcel, bool)>>,
     /// To notify when consensus is shutting down
     shutdown_notify: Arc<Notify>,

core/consensus/src/execution.rs

@@ -47,12 +47,12 @@ impl ToDigest for AuthenticStampedParcel {
 }
 
 /// A message an authority sends out attest that an Authentic stamp parcel is accurate. When an edge
-/// node gets 2f+1 of these it commmits the transactions in the parcel
+/// node gets 2f+1 of these it commits the transactions in the parcel
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct CommitteeAttestation {
     /// The digest we are attesting is correct
     pub digest: Digest,
-    /// We send random bytes with this messsage so it gives it a unique hash and differentiates it
+    /// We send random bytes with this message so it gives it a unique hash and differentiates it
     /// from the other committee members attestation broadcasts
     pub node_index: NodeIndex,
     pub epoch: Epoch,
@@ -61,9 +61,9 @@ pub struct CommitteeAttestation {
 pub struct Execution<Q: SyncQueryRunnerInterface> {
     /// Managing certificates generated by narwhal.
     executor: ExecutionEngineSocket,
-    /// Used to signal internal consensus proccesses that it is time to reconfigure for a new epoch
+    /// Used to signal internal consensus processes that it is time to reconfigure for a new epoch
     reconfigure_notify: Arc<Notify>,
-    /// Notifier that notifies everytime a block is executed on application state
+    /// Notifier that notifies every time a block is executed on application state
     new_block_notify: Arc<Notify>,
     /// Used to send payloads to the edge node consensus to broadcast out to other nodes
     tx_narwhal_batches: mpsc::Sender<(AuthenticStampedParcel, bool)>,
@@ -127,7 +127,7 @@ impl<Q: SyncQueryRunnerInterface> Execution<Q> {
         }
 
         // If we have the archive socket that means our node is in archive node and we should send
-        // the block and the reciept to be indexed
+        // the block and the receipt to be indexed
         if let (Some(block), Some(socket)) = (archive_block, &self.index_socket) {
             if let Err(e) = socket
                 .run(IndexRequest {
@@ -189,7 +189,7 @@ impl<Q: SyncQueryRunnerInterface> ExecutionState for Execution<Q> {
                 let epoch_changed = self.submit_batch(batch_payload, parcel.to_digest()).await;
 
                 if let Err(e) = self.tx_narwhal_batches.send((parcel, epoch_changed)).await {
-                    // This shouldnt ever happen. But if it does there is no critical tasks
+                    // This shouldn't ever happen. But if it does there is no critical tasks
                     // happening on the other end of this that would require a
                     // panic
                     error!("Narwhal failed to send batch payload to edge consensus: {e:?}");

core/dack-aggregator/src/tests.rs

@@ -100,7 +100,7 @@ async fn init_aggregator(path: PathBuf) -> Node<TestBinding> {
     let consensus = MockConsensus::<TestBinding>::init(
         consensus_config,
         &signer,
-        update_socket.clone(),
+        update_socket.unwrap(),
         query_runner.clone(),
         infusion::Blank::default(),
         None,

core/fetcher/src/tests.rs

@@ -157,9 +157,10 @@ async fn get_fetchers(
     .unwrap();
     app.start().await;
 
+    let update_socket = app.transaction_executor().unwrap();
     let mut peers = Vec::new();
     for (i, signer_config) in signers_configs.into_iter().enumerate() {
-        let (update_socket, query_runner) = (app.transaction_executor(), app.sync_query());
+        let query_runner = app.sync_query();
         let mut signer = Signer::<TestBinding>::init(signer_config, query_runner.clone()).unwrap();
         let topology = Topology::<TestBinding>::init(
             TopologyConfig::default(),

core/indexer/src/tests.rs

@@ -121,7 +121,7 @@ async fn test_submission() {
     let consensus = MockConsensus::<TestBinding>::init(
         consensus_config,
         &signer,
-        update_socket,
+        update_socket.unwrap(),
         query_runner.clone(),
         infusion::Blank::default(),
         None,

core/interfaces/src/application.rs

@@ -66,7 +66,7 @@ pub trait ApplicationInterface<C: Collection>:
     /// # Safety
     ///
     /// See the safety document for the [`ExecutionEngineSocket`].
-    fn transaction_executor(&self) -> ExecutionEngineSocket;
+    fn transaction_executor(&self) -> Option<ExecutionEngineSocket>;
 
     /// Returns the instance of a sync query runner which can be used to run queries without
     /// blocking or awaiting. A naive (& blocking) implementation can achieve this by simply
@@ -111,7 +111,7 @@ pub trait SyncQueryRunnerInterface: Clone + Send + Sync + 'static {
     /// Query Pub Key to Node Index Table
     fn pubkey_to_index(&self, pub_key: &NodePublicKey) -> Option<NodeIndex>;
 
-    /// Query Committe Table
+    /// Query Committee Table
     fn get_committe_info<V>(
         &self,
         epoch: &Epoch,

core/interfaces/src/consensus.rs

@@ -17,6 +17,7 @@ use crate::{
     BroadcastInterface,
     ConfigProviderInterface,
     IndexSocket,
+    NotifierInterface,
 };
 
 /// A socket that gives services and other sub-systems the required functionality to
@@ -39,8 +40,17 @@ pub trait ConsensusInterface<C: Collection>:
         app: ::ApplicationInterface,
         broadcast: ::BroadcastInterface,
         archive: ::ArchiveInterface,
+        notifier: ::NotifierInterface,
     ) {
-        let executor = app.transaction_executor();
+        let executor = app
+            .transaction_executor()
+            .expect("ConsensusInferface::_init - Update Socket should be available");
+        let epoch_change_notifier = notifier.emitters();
+        executor.inject(move |res| {
+            if res.change_epoch {
+                epoch_change_notifier.notify_waiters();
+            }
+        });
         let sqr = app.sync_query();
         let pubsub = broadcast.get_pubsub(crate::types::Topic::Consensus);
         Self::init(
@@ -71,7 +81,7 @@ pub trait ConsensusInterface<C: Collection>:
     #[blank = Socket::raw_bounded(64).0]
     fn mempool(&self) -> MempoolSocket;
 
-    /// Returns a tokio Notifier that notifies everytime a new block is finished being processed
+    /// Returns a tokio Notifier that notifies every time a new block is finished being processed
     #[blank = Default::default()]
     fn new_block_notifier(&self) -> Arc<Notify>;
 }