feat: add a mempool proxy support for concurrent calls

Cargo.toml

@@ -20,9 +20,11 @@ as_conversions = "deny"
 
 [workspace.dependencies]
 assert_matches = "1.5.0"
+async-trait = "0.1.79"
 axum = "0.6.12"
 clap = "4.3.10"
 derive_more = "0.99"
+const_format = "0.2.30"
 hyper = "1.2.0"
 papyrus_config = "0.3.0"
 pretty_assertions = "1.4.0"

crates/mempool_node/Cargo.toml

@@ -9,5 +9,19 @@ license.workspace = true
 workspace = true
 
 [dependencies]
+async-trait.workspace = true
+clap.workspace = true
+const_format.workspace = true
+starknet_gateway = { path = "../gateway", version = "0.0" }
+hyper.workspace = true
+serde.workspace = true
+serde_json.workspace = true
+starknet_api.workspace = true
+thiserror.workspace = true
+papyrus_config.workspace = true
 tokio.workspace = true
-async-trait = "0.1.79"
+validator.workspace = true
+
+[dev-dependencies]
+assert_matches.workspace = true
+pretty_assertions.workspace = true

crates/mempool_node/src/config/config_test.rs

@@ -0,0 +1,67 @@
+#![allow(unused_imports)]
+use crate::config::node_command;
+use crate::config::{ComponentConfig, ComponentExecutionConfig, MempoolNodeConfig};
+use assert_matches::assert_matches;
+use papyrus_config::dumping::SerializeConfig;
+use papyrus_config::loading::load_and_process_config;
+use papyrus_config::presentation::get_config_presentation;
+use papyrus_config::validators::ParsedValidationErrors;
+use papyrus_config::{SerializationType, SerializedContent, SerializedParam};
+use starknet_gateway::GatewayConfig;
+use std::env::{self, args};
+use std::fs::File;
+use std::ops::IndexMut;
+use std::path::{Path, PathBuf};
+use validator::Validate;
+
+const TEST_FILES_FOLDER: &str = "./src/test_files";
+const CONFIG_FILE: &str = "mempool_node_config.json";
+
+fn get_config_file(file_name: &str) -> Result<MempoolNodeConfig, papyrus_config::ConfigError> {
+    let config_file = File::open(Path::new(TEST_FILES_FOLDER).join(file_name)).unwrap();
+    load_and_process_config::<MempoolNodeConfig>(config_file, node_command(), vec![])
+}
+
+#[test]
+fn test_valid_config() {
+    // Read the valid config file and validate its content.
+    let expected_config = MempoolNodeConfig {
+        components: ComponentConfig {
+            gateway_component: ComponentExecutionConfig { execute: true },
+            mempool_component: ComponentExecutionConfig { execute: false },
+        },
+        gateway_config: GatewayConfig {
+            bind_address: String::from("0.0.0.0:8080"),
+        },
+    };
+    let loaded_config = get_config_file(CONFIG_FILE).unwrap();
+
+    assert!(loaded_config.validate().is_ok());
+    assert_eq!(loaded_config, expected_config);
+}
+
+#[test]
+fn test_components_config() {
+    // Read the valid config file and check that the validator finds no errors.
+    let mut config = get_config_file(CONFIG_FILE).unwrap();
+    assert!(config.validate().is_ok());
+
+    // Invalidate the gateway component and check that the validator finds an error.
+    config.components.gateway_component.execute = false;
+
+    assert_matches!(config.validate(), Err(e) => {
+        let parse_err = ParsedValidationErrors::from(e);
+        let mut error_msg = String::new();
+        for error in parse_err.0 {
+            if error.param_path == "components.__all__" {
+                error_msg.push_str(&error.code);
+                break;
+            }
+        }
+        assert_eq!(error_msg, "Invalid components configuration.");
+    });
+
+    // Validate the mempool component and check that the validator finds no errors.
+    config.components.mempool_component.execute = true;
+    assert!(config.validate().is_ok());
+}

crates/mempool_node/src/config/mod.rs

@@ -0,0 +1,128 @@
+#[cfg(test)]
+mod config_test;
+
+use std::collections::BTreeMap;
+use std::fs::File;
+use std::path::Path;
+
+use clap::Command;
+use papyrus_config::dumping::{append_sub_config_name, ser_param, SerializeConfig};
+use papyrus_config::loading::load_and_process_config;
+use papyrus_config::ParamPrivacyInput;
+use papyrus_config::{ConfigError, ParamPath, SerializedParam};
+use serde::{Deserialize, Serialize};
+use starknet_gateway::GatewayConfig;
+use validator::{Validate, ValidationError};
+
+use crate::version::VERSION_FULL;
+
+// The path of the default configuration file, provided as part of the crate.
+pub const DEFAULT_CONFIG_PATH: &str = "config/default_config.json";
+
+/// The single crate configuration.
+#[derive(Clone, Debug, Serialize, Deserialize, Validate, PartialEq)]
+pub struct ComponentExecutionConfig {
+    pub execute: bool,
+}
+
+impl SerializeConfig for ComponentExecutionConfig {
+    fn dump(&self) -> BTreeMap<ParamPath, SerializedParam> {
+        BTreeMap::from_iter([ser_param(
+            "execute",
+            &self.execute,
+            "The component execution flag.",
+            ParamPrivacyInput::Public,
+        )])
+    }
+}
+
+impl Default for ComponentExecutionConfig {
+    fn default() -> Self {
+        Self { execute: true }
+    }
+}
+
+/// The components configuration.
+#[derive(Clone, Debug, Serialize, Deserialize, Validate, PartialEq, Default)]
+#[validate(schema(function = "validate_components_config"))]
+pub struct ComponentConfig {
+    pub gateway_component: ComponentExecutionConfig,
+    pub mempool_component: ComponentExecutionConfig,
+}
+
+impl SerializeConfig for ComponentConfig {
+    fn dump(&self) -> BTreeMap<ParamPath, SerializedParam> {
+        #[allow(unused_mut)]
+        let mut sub_configs = vec![
+            append_sub_config_name(self.gateway_component.dump(), "gateway_component"),
+            append_sub_config_name(self.mempool_component.dump(), "mempool_component"),
+        ];
+
+        sub_configs.into_iter().flatten().collect()
+    }
+}
+
+pub fn validate_components_config(components: &ComponentConfig) -> Result<(), ValidationError> {
+    if components.gateway_component.execute || components.mempool_component.execute {
+        return Ok(());
+    }
+
+    let mut error = ValidationError::new("Invalid components configuration.");
+    error.message = Some("At least one component should be allowed to execute.".into());
+    Err(error)
+}
+
+/// The configurations of the various components of the node.
+#[derive(Debug, Deserialize, Serialize, Clone, PartialEq, Validate, Default)]
+pub struct MempoolNodeConfig {
+    #[validate]
+    pub components: ComponentConfig,
+    #[validate]
+    pub gateway_config: GatewayConfig,
+}
+
+impl SerializeConfig for MempoolNodeConfig {
+    fn dump(&self) -> BTreeMap<ParamPath, SerializedParam> {
+        #[allow(unused_mut)]
+        let mut sub_configs = vec![
+            append_sub_config_name(self.components.dump(), "components"),
+            append_sub_config_name(self.gateway_config.dump(), "gateway_config"),
+        ];
+
+        sub_configs.into_iter().flatten().collect()
+    }
+}
+
+impl MempoolNodeConfig {
+    /// Creates a config object. Selects the values from the default file and from resources with
+    /// higher priority.
+    fn load_and_process_config_file(
+        args: Vec<String>,
+        config_file_name: Option<&str>,
+    ) -> Result<Self, ConfigError> {
+        let config_file_name = match config_file_name {
+            Some(file_name) => file_name,
+            None => DEFAULT_CONFIG_PATH,
+        };
+
+        let default_config_file = File::open(Path::new(config_file_name))?;
+        load_and_process_config(default_config_file, node_command(), args)
+    }
+
+    pub fn load_and_process(args: Vec<String>) -> Result<Self, ConfigError> {
+        Self::load_and_process_config_file(args, None)
+    }
+    pub fn load_and_process_file(
+        args: Vec<String>,
+        config_file_name: &str,
+    ) -> Result<Self, ConfigError> {
+        Self::load_and_process_config_file(args, Some(config_file_name))
+    }
+}
+
+/// The command line interface of this node.
+pub fn node_command() -> Command {
+    Command::new("Mempool")
+        .version(VERSION_FULL)
+        .about("Mempool is a StarkNet mempool node written in Rust.")
+}

crates/mempool_node/src/lib.rs

@@ -0,0 +1,2 @@
+pub mod config;
+pub mod version;

crates/mempool_node/src/mempool.rs

@@ -1,30 +1,24 @@
 use async_trait::async_trait;
+use tokio::sync::Mutex;
 
 pub type AddTransactionCallType = u32;
 pub type AddTransactionReturnType = usize;
 
 #[async_trait]
 pub trait MempoolTrait {
-    async fn add_transaction(&mut self, tx: AddTransactionCallType) -> AddTransactionReturnType;
+    async fn add_transaction(&self, tx: AddTransactionCallType) -> AddTransactionReturnType;
 }
 
 #[derive(Default)]
 pub struct Mempool {
-    transactions: Vec<u32>,
-}
-
-impl Mempool {
-    pub fn new() -> Self {
-        Self {
-            transactions: vec![],
-        }
-    }
+    transactions: Mutex<Vec<u32>>,
 }
 
 #[async_trait]
 impl MempoolTrait for Mempool {
-    async fn add_transaction(&mut self, tx: AddTransactionCallType) -> AddTransactionReturnType {
-        self.transactions.push(tx);
-        self.transactions.len()
+    async fn add_transaction(&self, tx: AddTransactionCallType) -> AddTransactionReturnType {
+        let mut guarded_transactions = self.transactions.lock().await;
+        guarded_transactions.push(tx);
+        guarded_transactions.len()
     }
 }