Added Versioning middleware

massa-models/src/config/constants.rs

@@ -225,6 +225,10 @@ pub const NETWORK_NODE_EVENT_CHANNEL_SIZE: usize = 10_000;
 //
 /// Threshold to accept a new versioning
 pub const VERSIONING_THRESHOLD_TRANSITION_ACCEPTED: Amount = Amount::from_mantissa_scale(75, 0);
+/// versioning controller event channel size
+pub const VERSIONING_CONTROLLER_CHANNEL_SIZE: usize = 10_000;
+/// versioning block stats queue size
+pub const VERSIONING_COUNT_BLOCKS_CONSIDERED: usize = 1000;
 
 // Some checks at compile time that should not be ignored!
 #[allow(clippy::assertions_on_constants)]

massa-node/Cargo.toml

@@ -51,6 +51,8 @@ massa_time = { path = "../massa-time" }
 massa_wallet = { path = "../massa-wallet" }
 massa_factory_exports = { path = "../massa-factory-exports" }
 massa_factory_worker = { path = "../massa-factory-worker" }
+massa_versioning_exports = { path = "../massa-versioning-exports" }
+massa_versioning_worker = { path = "../massa-versioning-worker" }
 
 # for more information on what are the following features used for, see the cargo.toml at workspace level
 [features]

massa-node/src/main.rs

@@ -44,7 +44,8 @@ use massa_models::config::constants::{
     NETWORK_NODE_COMMAND_CHANNEL_SIZE, NETWORK_NODE_EVENT_CHANNEL_SIZE, OPERATION_VALIDITY_PERIODS,
     PERIODS_PER_CYCLE, POOL_CONTROLLER_CHANNEL_SIZE, POS_MISS_RATE_DEACTIVATION_THRESHOLD,
     POS_SAVED_CYCLES, PROTOCOL_CONTROLLER_CHANNEL_SIZE, PROTOCOL_EVENT_CHANNEL_SIZE, ROLL_PRICE,
-    T0, THREAD_COUNT, VERSION,
+    T0, THREAD_COUNT, VERSION, VERSIONING_CONTROLLER_CHANNEL_SIZE,
+    VERSIONING_COUNT_BLOCKS_CONSIDERED,
 };
 use massa_models::config::CONSENSUS_BOOTSTRAP_PART_SIZE;
 use massa_network_exports::{Establisher, NetworkConfig, NetworkManager};
@@ -60,6 +61,12 @@ use massa_protocol_exports::{
 use massa_protocol_worker::start_protocol_controller;
 use massa_storage::Storage;
 use massa_time::MassaTime;
+use massa_versioning_exports::{
+    VersioningCommand, /*VersioningCommandSender,*/ VersioningConfig, VersioningManager,
+    VersioningReceivers, VersioningSenders,
+};
+use massa_versioning_worker::start_versioning_worker;
+use massa_versioning_worker::versioning::MipStore;
 use massa_wallet::Wallet;
 use parking_lot::RwLock;
 use std::path::PathBuf;
@@ -86,6 +93,7 @@ async fn launch(
     Box<dyn SelectorManager>,
     Box<dyn PoolManager>,
     ProtocolManager,
+    VersioningManager,
     NetworkManager,
     Box<dyn FactoryManager>,
     mpsc::Receiver<()>,
@@ -382,6 +390,9 @@ async fn launch(
     let (protocol_command_sender, protocol_command_receiver) =
         mpsc::channel::<ProtocolCommand>(PROTOCOL_CONTROLLER_CHANNEL_SIZE);
 
+    let (_versioning_command_sender, versioning_command_receiver) =
+        mpsc::channel::<VersioningCommand>(VERSIONING_CONTROLLER_CHANNEL_SIZE);
+
     let consensus_config = ConsensusConfig {
         genesis_timestamp: *GENESIS_TIMESTAMP,
         end_timestamp: *END_TIMESTAMP,
@@ -432,6 +443,29 @@ async fn launch(
         shared_storage.clone(),
     );
 
+    let versioning_config = VersioningConfig {
+        count_blocks_considered: VERSIONING_COUNT_BLOCKS_CONSIDERED,
+    };
+
+    let versioning_senders = VersioningSenders {};
+
+    let versioning_receivers = VersioningReceivers {
+        versioning_command_receiver,
+    };
+
+    // Creates an empty default store
+    let mip_store = MipStore::try_from([]).unwrap();
+
+    // launch versioning manager
+    let versioning_manager = start_versioning_worker(
+        versioning_config,
+        versioning_receivers,
+        versioning_senders.clone(),
+        mip_store.clone(),
+    )
+    .await
+    .expect("could not start versioning controller");
+
     // launch protocol controller
     let protocol_config = ProtocolConfig {
         thread_count: THREAD_COUNT,
@@ -632,6 +666,7 @@ async fn launch(
         selector_manager,
         pool_manager,
         protocol_manager,
+        versioning_manager,
         network_manager,
         factory_manager,
         api_private_stop_rx,
@@ -648,6 +683,7 @@ struct Managers {
     selector_manager: Box<dyn SelectorManager>,
     pool_manager: Box<dyn PoolManager>,
     protocol_manager: ProtocolManager,
+    versioning_manager: VersioningManager,
     network_manager: NetworkManager,
     factory_manager: Box<dyn FactoryManager>,
 }
@@ -661,6 +697,7 @@ async fn stop(
         mut selector_manager,
         mut pool_manager,
         protocol_manager,
+        versioning_manager,
         network_manager,
         mut factory_manager,
     }: Managers,
@@ -706,6 +743,9 @@ async fn stop(
     // stop selector controller
     selector_manager.stop();
 
+    // stop versioning controller
+    versioning_manager.stop();
+
     // stop pool controller
     // TODO
     //let protocol_pool_event_receiver = pool_manager.stop().await.expect("pool shutdown failed");
@@ -818,6 +858,7 @@ async fn run(args: Args) -> anyhow::Result<()> {
             selector_manager,
             pool_manager,
             protocol_manager,
+            versioning_manager,
             network_manager,
             factory_manager,
             mut api_private_stop_rx,
@@ -886,6 +927,7 @@ async fn run(args: Args) -> anyhow::Result<()> {
                 selector_manager,
                 pool_manager,
                 protocol_manager,
+                versioning_manager,
                 network_manager,
                 factory_manager,
             },

massa-versioning-exports/Cargo.toml

@@ -7,7 +7,19 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+displaydoc = "0.2"
+tokio = { version = "1.23", features = ["full"] }
+serde = { version = "1.0", features = ["derive"] }
+tracing = "0.1"
+thiserror = "1.0"
+# custom modules
+massa_logging = { path = "../massa-logging" }
+massa_models = { path = "../massa-models" }
+massa_storage = { path = "../massa-storage" }
 
 [dev-dependencies]
 
 [features]
+
+
+

massa-versioning-exports/src/channels.rs

@@ -0,0 +1,13 @@
+use tokio::sync::mpsc;
+
+use crate::VersioningCommand;
+
+/// Contains channels (senders) used by the versioning worker
+/// Contains (a) channel(s) to send info to api
+#[derive(Clone)]
+pub struct VersioningSenders {}
+
+/// Contains channels(receivers) used by the protocol worker
+pub struct VersioningReceivers {
+    pub versioning_command_receiver: mpsc::Receiver<VersioningCommand>,
+}

massa-versioning-exports/src/error.rs

@@ -0,0 +1,15 @@
+// Copyright (c) 2022 MASSA LABS <info@massa.net>
+
+use displaydoc::Display;
+use massa_models::error::ModelsError;
+use thiserror::Error;
+
+/// versioning error
+#[non_exhaustive]
+#[derive(Display, Error, Debug)]
+pub enum VersioningMiddlewareError {
+    /// An error occurred during channel communication: {0}
+    ChannelError(String),
+    /// Models error: {0}
+    ModelsError(#[from] ModelsError),
+}

massa-versioning-exports/src/lib.rs

@@ -0,0 +1,14 @@
+mod channels;
+mod error;
+mod settings;
+mod versioning_controller;
+
+pub use channels::{VersioningReceivers, VersioningSenders};
+pub use error::VersioningMiddlewareError;
+pub use settings::VersioningConfig;
+pub use versioning_controller::{
+    VersioningCommand, VersioningCommandSender, VersioningManagementCommand, VersioningManager,
+};
+
+#[cfg(test)]
+pub mod tests;

massa-versioning-exports/src/settings.rs

@@ -0,0 +1,4 @@
+pub struct VersioningConfig {
+    /// Nb blocks to consider for versioning stats
+    pub count_blocks_considered: usize,
+}

massa-versioning-exports/src/versioning_controller.rs

@@ -0,0 +1,54 @@
+// Copyright (c) 2022 MASSA LABS <info@massa.net>
+
+use crate::error::VersioningMiddlewareError;
+
+use serde::Serialize;
+
+use tokio::sync::mpsc;
+
+/// Commands that protocol worker can process
+#[derive(Debug)]
+pub enum VersioningCommand {
+    /// Notify new finalized block header
+    FinalizedBlockVersion { announced_version: u32 },
+}
+
+/// protocol management commands
+#[derive(Debug, Serialize)]
+pub enum VersioningManagementCommand {}
+
+/// protocol command sender
+#[derive(Clone)]
+pub struct VersioningCommandSender(pub mpsc::Sender<VersioningCommand>);
+
+impl VersioningCommandSender {
+    pub fn send_block_version(
+        &mut self,
+        announced_version: u32,
+    ) -> Result<(), VersioningMiddlewareError> {
+        self.0
+            .blocking_send(VersioningCommand::FinalizedBlockVersion { announced_version })
+            .map_err(|_| {
+                VersioningMiddlewareError::ChannelError(
+                    "send_block_header command send error".into(),
+                )
+            })
+    }
+}
+
+/// versioning manager used to stop the protocol
+pub struct VersioningManager {
+    manager_tx: mpsc::Sender<VersioningManagementCommand>,
+}
+
+impl VersioningManager {
+    /// new versioning manager
+    pub fn new(manager_tx: mpsc::Sender<VersioningManagementCommand>) -> Self {
+        VersioningManager { manager_tx }
+    }
+
+    /// Stop the versioning controller
+    pub fn stop(self) {
+        drop(self.manager_tx);
+    }
+}

massa-versioning-worker/Cargo.toml

@@ -11,11 +11,15 @@ parking_lot = "0.12"
 thiserror = "1.0"
 num_enum = "0.5"
 nom = "7.1"
+tokio = { version = "1.23", features = ["full"] }
+tracing = "0.1"
+queues = "1.1.0"
 
 # custom module
 massa_time = { path = "../massa-time" }
 massa_models = { path = "../massa-models" }
 massa_serialization = { path = "../massa-serialization" }
+massa_versioning_exports = { path = "../massa-versioning-exports" }
 
 [dev-dependencies]
 chrono = "0.4"

massa-versioning-worker/src/lib.rs

@@ -43,7 +43,11 @@
 
 pub mod versioning;
 pub mod versioning_factory;
+pub mod versioning_middleware;
 pub mod versioning_ser_der;
+mod versioning_worker;
+
+pub use versioning_worker::start_versioning_worker;
 
 /// Test utils
 #[cfg(any(test, feature = "testing"))]