Feature/ping timings (#603)

* WIP

* WIP for time measurement

* In theory working sender and listener

* Further additions + main.rs for local testing

* Further improvements + config builder

* Initial integration into mixnode

* Verifying mixnode version

* 1.52+ clipy warning

* 1.54 nightly clippy fixes

* Changed HTTP Api to bind to the same ip as used for mix packets

* Changed measurements to instead write to shared object

* Required dependencies

* Connecting with the http api

* Updated mixnode common rand dependency

mixnode/src/node/http/mod.rs

@@ -1 +1,8 @@
+use rocket::Request;
+
 pub(crate) mod verloc;
+
+#[catch(404)]
+pub(crate) fn not_found(req: &Request) -> String {
+    format!("I couldn't find '{}'. Try something else?", req.uri())
+}

mixnode/src/node/http/verloc.rs

@@ -1,28 +1,23 @@
+use mixnode_common::rtt_measurement::{AtomicVerlocResult, Verloc};
+use rocket::State;
 use rocket_contrib::json::Json;
-use serde::Serialize;
+
+pub(crate) struct VerlocState {
+    shared: AtomicVerlocResult,
+}
+
+impl VerlocState {
+    pub fn new(atomic_verloc_result: AtomicVerlocResult) -> Self {
+        VerlocState {
+            shared: atomic_verloc_result,
+        }
+    }
+}
 
 /// Provides verifiable location (verloc) measurements for this mixnode - a list of the
 /// round-trip times, in milliseconds, for all other mixnodes that this node knows about.
 #[get("/verloc")]
-pub(crate) fn verloc() -> Json<Vec<Foomp>> {
-    // replace the foomps with a reference to the measurements vec :)
-    let foomp1 = Foomp {
-        ip: "1.2.3.4".to_string(),
-        port: "1789".to_string(),
-        identity_key: "abc".to_string(),
-    };
-    let foomp2 = Foomp {
-        ip: "2.3.4.5".to_string(),
-        port: "1789".to_string(),
-        identity_key: "def".to_string(),
-    };
-    let foomps = vec![foomp1, foomp2];
-    Json(foomps)
-}
-
-#[derive(Serialize)]
-pub(crate) struct Foomp {
-    ip: String,
-    port: String,
-    identity_key: String,
+pub(crate) async fn verloc(state: State<'_, VerlocState>) -> Json<Vec<Verloc>> {
+    // since it's impossible to get a mutable reference to the state, we can't cache any results outside the lock : (
+    Json(state.shared.clone_data().await)
 }

mixnode/src/node/mod.rs

@@ -2,16 +2,21 @@
 // SPDX-License-Identifier: Apache-2.0
 
 use crate::config::Config;
-use crate::node::http::verloc::verloc;
+use crate::node::http::{
+    not_found,
+    verloc::{verloc, VerlocState},
+};
 use crate::node::listener::connection_handler::packet_processing::PacketProcessor;
 use crate::node::listener::connection_handler::ConnectionHandler;
 use crate::node::listener::Listener;
 use crate::node::packet_delayforwarder::{DelayForwarder, PacketDelayForwardSender};
 use crypto::asymmetric::{encryption, identity};
 use log::{error, info, warn};
+use mixnode_common::rtt_measurement::{self, AtomicVerlocResult, RttMeasurer};
 use std::process;
 use std::sync::Arc;
 use tokio::runtime::Runtime;
+use version_checker::parse_version;
 
 pub(crate) mod http;
 mod listener;
@@ -38,9 +43,24 @@ impl MixNode {
         }
     }
 
-    fn start_http_api(&self) {
+    fn start_http_api(&self, atomic_verloc_result: AtomicVerlocResult) {
         info!("Starting HTTP API on port 8000...");
-        tokio::spawn(async move { rocket::build().mount("/", routes![verloc]).launch().await });
+
+        let mut config = rocket::config::Config::release_default();
+        // bind to the same address as we are using for mixnodes
+        config.address = self.config.get_listening_address().ip();
+
+        let state = VerlocState::new(atomic_verloc_result);
+
+        tokio::spawn(async move {
+            rocket::build()
+                .configure(config)
+                .mount("/", routes![verloc])
+                .register("/", catchers![not_found])
+                .manage(state)
+                .launch()
+                .await
+        });
     }
 
     fn start_metrics_reporter(&self) -> metrics::MetricsReporter {
@@ -93,6 +113,44 @@ impl MixNode {
         packet_sender
     }
 
+    fn start_rtt_measurer(&self) -> AtomicVerlocResult {
+        info!("Starting the round-trip-time measurer...");
+
+        // this is a sanity check to make sure we didn't mess up with the minimum version at some point
+        // and whether the user has run update if they're using old config
+        // if this code exists in the node, it MUST BE compatible
+        let config_version =
+            parse_version(self.config.get_version()).expect("malformed version in the config file");
+        let minimum_version = parse_version(rtt_measurement::MINIMUM_NODE_VERSION).unwrap();
+        if config_version < minimum_version {
+            error!("You seem to have not updated your mixnode configuration file - please run `upgrade` before attempting again");
+            process::exit(1)
+        }
+
+        // use the same binding address with the HARDCODED port for time being (I don't like that approach personally)
+
+        let listening_address = rtt_measurement::replace_port(
+            self.config.get_listening_address(),
+            rtt_measurement::DEFAULT_MEASUREMENT_PORT,
+        );
+        let config = rtt_measurement::ConfigBuilder::new()
+            .listening_address(listening_address)
+            .packets_per_node(self.config.get_measurement_packets_per_node())
+            .packet_timeout(self.config.get_measurement_packet_timeout())
+            .delay_between_packets(self.config.get_measurement_delay_between_packets())
+            .tested_nodes_batch_size(self.config.get_measurement_tested_nodes_batch_size())
+            .testing_interval(self.config.get_measurement_testing_interval())
+            .retry_timeout(self.config.get_measurement_retry_timeout())
+            .validator_urls(self.config.get_validator_rest_endpoints())
+            .mixnet_contract_address(self.config.get_validator_mixnet_contract_address())
+            .build();
+
+        let mut rtt_measurer = RttMeasurer::new(config, Arc::clone(&self.identity_keypair));
+        let atomic_verloc_results = rtt_measurer.get_verloc_results_pointer();
+        tokio::spawn(async move { rtt_measurer.run().await });
+        atomic_verloc_results
+    }
+
     // TODO: ask DH whether this function still makes sense in ^0.10
     async fn check_if_same_ip_node_exists(&mut self) -> Option<String> {
         let validator_client_config = validator_client_rest::Config::new(
@@ -148,7 +206,9 @@ impl MixNode {
             let metrics_reporter = self.start_metrics_reporter();
             let delay_forwarding_channel = self.start_packet_delay_forwarder(metrics_reporter.clone());
             self.start_socket_listener(metrics_reporter, delay_forwarding_channel);
-            self.start_http_api();
+
+            let atomic_verloc_results= self.start_rtt_measurer();
+            self.start_http_api(atomic_verloc_results);
 
             info!("Finished nym mixnode startup procedure - it should now be able to receive mix traffic!");
             self.wait_for_interrupt().await