feat(new-RPC): connection healthcheck implementation for peers

mm2src/mm2_core/src/mm_ctx.rs

@@ -1,15 +1,19 @@
 #[cfg(feature = "track-ctx-pointer")]
 use common::executor::Timer;
-use common::executor::{abortable_queue::{AbortableQueue, WeakSpawner},
-                       graceful_shutdown, AbortSettings, AbortableSystem, SpawnAbortable, SpawnFuture};
 use common::log::{self, LogLevel, LogOnError, LogState};
 use common::{cfg_native, cfg_wasm32, small_rng};
+use common::{executor::{abortable_queue::{AbortableQueue, WeakSpawner},
+                        graceful_shutdown, AbortSettings, AbortableSystem, SpawnAbortable, SpawnFuture},
+             expirable_map::ExpirableMap};
+use futures::channel::oneshot;
+use futures::lock::Mutex as AsyncMutex;
 use gstuff::{try_s, Constructible, ERR, ERRL};
 use lazy_static::lazy_static;
 use mm2_event_stream::{controller::Controller, Event, EventStreamConfiguration};
 use mm2_metrics::{MetricsArc, MetricsOps};
 use primitives::hash::H160;
 use rand::Rng;
+use serde::Deserialize;
 use serde_json::{self as json, Value as Json};
 use shared_ref_counter::{SharedRc, WeakRc};
 use std::any::Any;
@@ -30,7 +34,6 @@ cfg_wasm32! {
 cfg_native! {
     use db_common::async_sql_conn::AsyncConnection;
     use db_common::sqlite::rusqlite::Connection;
-    use futures::lock::Mutex as AsyncMutex;
     use rustls::ServerName;
     use mm2_metrics::prometheus;
     use mm2_metrics::MmMetricsError;
@@ -42,6 +45,36 @@ cfg_native! {
 /// Default interval to export and record metrics to log.
 const EXPORT_METRICS_INTERVAL: f64 = 5. * 60.;
 
+pub struct HealthChecker {
+    /// Links the RPC context to the P2P context to handle health check responses.
+    pub response_handler: AsyncMutex<ExpirableMap<String, oneshot::Sender<()>>>,
+    /// This is used to record healthcheck sender peers in an expirable manner to prevent DDoS attacks.
+    pub ddos_shield: AsyncMutex<ExpirableMap<String, ()>>,
+    pub config: HealthcheckConfig,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(default)]
+pub struct HealthcheckConfig {
+    /// Required time (millisecond) to wait before processing another healthcheck request from the same peer.
+    pub blocking_ms_for_per_address: u64,
+    /// Lifetime of the message.
+    /// Do not change this unless you know what you are doing.
+    pub message_expiration_secs: u64,
+    /// Maximum time (milliseconds) to wait for healthcheck response.
+    pub timeout_secs: u64,
+}
+
+impl Default for HealthcheckConfig {
+    fn default() -> Self {
+        Self {
+            blocking_ms_for_per_address: 750,
+            message_expiration_secs: 10,
+            timeout_secs: 10,
+        }
+    }
+}
+
 /// MarketMaker state, shared between the various MarketMaker threads.
 ///
 /// Every MarketMaker has one and only one instance of `MmCtx`.
@@ -142,6 +175,7 @@ pub struct MmCtx {
     /// asynchronous handle for rusqlite connection.
     #[cfg(not(target_arch = "wasm32"))]
     pub async_sqlite_connection: Constructible<Arc<AsyncMutex<AsyncConnection>>>,
+    pub health_checker: HealthChecker,
 }
 
 impl MmCtx {
@@ -191,6 +225,11 @@ impl MmCtx {
             nft_ctx: Mutex::new(None),
             #[cfg(not(target_arch = "wasm32"))]
             async_sqlite_connection: Constructible::default(),
+            health_checker: HealthChecker {
+                response_handler: AsyncMutex::new(ExpirableMap::default()),
+                ddos_shield: AsyncMutex::new(ExpirableMap::default()),
+                config: HealthcheckConfig::default(),
+            },
         }
     }
 
@@ -776,6 +815,13 @@ impl MmCtxBuilder {
                         .expect("Invalid json value in 'event_stream_configuration'.");
                 ctx.event_stream_configuration = Some(event_stream_configuration);
             }
+
+            let healthcheck_config = &ctx.conf["healthcheck_config"];
+            if !healthcheck_config.is_null() {
+                let healthcheck_config: HealthcheckConfig =
+                    json::from_value(healthcheck_config.clone()).expect("Invalid json value in 'healthcheck_config'.");
+                ctx.health_checker.config = healthcheck_config;
+            }
         }
 
         #[cfg(target_arch = "wasm32")]

mm2src/mm2_main/Cargo.toml

@@ -32,6 +32,7 @@ bitcrypto = { path = "../mm2_bitcoin/crypto" }
 blake2 = "0.10.6"
 bytes = "0.4"
 chain = { path = "../mm2_bitcoin/chain" }
+chrono = "0.4"
 cfg-if = "1.0"
 coins = { path = "../coins" }
 coins_activation = { path = "../coins_activation" }