#185908936 - Connect router to ring manager

crates/core/Cargo.toml

@@ -31,6 +31,7 @@ config = { version = "0.13.0", features = [ "toml" ] }
 crossbeam = "0.8.2"
 ctrlc = { version = "3.4", features = ["termination"] }
 dashmap = "^5.5"
+delegate = "0.10"
 directories = "5"
 either = { workspace = true , features = ["serde"] }
 futures = "0.3.21"

crates/core/src/config.rs

@@ -86,6 +86,7 @@ pub struct ConfigPaths {
     secrets_dir: PathBuf,
     db_dir: PathBuf,
     app_data_dir: PathBuf,
+    event_log: PathBuf,
 }
 
 impl ConfigPaths {
@@ -122,12 +123,21 @@ impl ConfigPaths {
             fs::create_dir_all(db_dir.join("local"))?;
         }
 
+        let event_log = app_data_dir.join("_EVENT_LOG");
+        if !event_log.exists() {
+            fs::write(&event_log, [])?;
+            let mut local_file = event_log.clone();
+            local_file.set_file_name("_EVENT_LOG_LOCAL");
+            fs::write(local_file, [])?;
+        }
+
         Ok(Self {
             contracts_dir,
             delegates_dir,
             secrets_dir,
             db_dir,
             app_data_dir,
+            event_log,
         })
     }
 }
@@ -139,7 +149,7 @@ impl Config {
 
     pub fn set_op_mode(mode: OperationMode) {
         let local_mode = matches!(mode, OperationMode::Local);
-        Self::get_static_conf()
+        Self::conf()
             .local_mode
             .store(local_mode, std::sync::atomic::Ordering::SeqCst);
     }
@@ -176,8 +186,24 @@ impl Config {
         }
     }
 
-    pub fn get_static_conf() -> &'static Config {
-        CONFIG.get_or_init(|| Config::load_conf().expect("Failed to load configuration"))
+    pub fn event_log(&self) -> PathBuf {
+        if self.local_mode.load(std::sync::atomic::Ordering::SeqCst) {
+            let mut local_file = self.config_paths.event_log.clone();
+            local_file.set_file_name("_EVENT_LOG_LOCAL");
+            local_file
+        } else {
+            self.config_paths.event_log.to_owned()
+        }
+    }
+
+    pub fn conf() -> &'static Config {
+        CONFIG.get_or_init(|| match Config::load_conf() {
+            Ok(config) => config,
+            Err(err) => {
+                tracing::error!("failed while loading configuration: {err}");
+                panic!("Failed while loading configuration")
+            }
+        })
     }
 
     fn load_conf() -> std::io::Result<Config> {

crates/core/src/contract/executor.rs

@@ -31,7 +31,7 @@ use crate::runtime::{
 };
 use crate::{
     client_events::{ClientId, HostResult},
-    node::{NodeConfig, P2pBridge},
+    node::NodeConfig,
     operations::{self, op_trait::Operation},
     DynError,
 };
@@ -136,7 +136,7 @@ impl ExecutorToEventLoopChannel<ExecutorHalve> {
     async fn send_to_event_loop<Op, T>(&mut self, message: T) -> Result<(), DynError>
     where
         T: ComposeNetworkMessage<Op>,
-        Op: Operation<P2pBridge> + Send + 'static,
+        Op: Operation + Send + 'static,
     {
         let op = message.initiate_op(&self.op_manager);
         self.end.sender.send(*op.id()).await?;
@@ -178,7 +178,7 @@ mod sealed {
 trait ComposeNetworkMessage<Op>
 where
     Self: Sized,
-    Op: Operation<P2pBridge> + Send + 'static,
+    Op: Operation + Send + 'static,
 {
     fn initiate_op(self, op_manager: &OpManager) -> Op {
         todo!()
@@ -205,7 +205,7 @@ impl ComposeNetworkMessage<operations::get::GetOp> for GetContract {
         op: operations::get::GetOp,
         op_manager: &OpManager,
     ) -> Result<Transaction, OpError> {
-        let id = *<operations::get::GetOp as Operation<P2pBridge>>::id(&op);
+        let id = *op.id();
         operations::get::request_get(op_manager, op, None).await?;
         Ok(id)
     }
@@ -321,7 +321,7 @@ impl Executor<Runtime> {
     // dependencies to be resolved
     async fn op_request<Op, M>(&mut self, request: M) -> Result<Op, DynError>
     where
-        Op: Operation<P2pBridge> + Send + 'static,
+        Op: Operation + Send + 'static,
         M: ComposeNetworkMessage<Op>,
     {
         debug_assert!(self.event_loop_channel.is_some());
@@ -342,7 +342,7 @@ impl Executor<Runtime> {
     pub async fn from_config(config: NodeConfig) -> Result<Self, DynError> {
         const MAX_SIZE: i64 = 10 * 1024 * 1024;
         const MAX_MEM_CACHE: u32 = 10_000_000;
-        let static_conf = crate::config::Config::get_static_conf();
+        let static_conf = crate::config::Config::conf();
 
         let state_store = StateStore::new(Storage::new().await?, MAX_MEM_CACHE).unwrap();
 

crates/core/src/contract/in_memory.rs

@@ -46,7 +46,7 @@ where
             _kv_store: StateStore::new(kv_store, 10_000_000).unwrap(),
             _runtime: MockRuntime {
                 contract_store: ContractStore::new(
-                    Config::get_static_conf().contracts_dir(),
+                    Config::conf().contracts_dir(),
                     Self::MAX_MEM_CACHE,
                 )
                 .unwrap(),

crates/core/src/contract/storages/rocks_db.rs

@@ -9,7 +9,7 @@ pub struct RocksDb(DB);
 impl RocksDb {
     #[cfg_attr(feature = "sqlite", allow(unused))]
     pub async fn new() -> Result<Self, rocksdb::Error> {
-        let path = Config::get_static_conf().db_dir().join("freenet.db");
+        let path = Config::conf().db_dir().join("freenet.db");
         tracing::info!("loading contract store from {path:?}");
 
         let mut opts = Options::default();

crates/core/src/contract/storages/sqlite.rs

@@ -18,7 +18,7 @@ static POOL: Lazy<SqlitePool> = Lazy::new(|| {
     let opts = if cfg!(test) {
         SqliteConnectOptions::from_str("sqlite::memory:").unwrap()
     } else {
-        let conn_str = Config::get_static_conf().db_dir().join("freenet.db");
+        let conn_str = Config::conf().db_dir().join("freenet.db");
         tracing::info!("loading contract store from {conn_str:?}");
         SqliteConnectOptions::new()
             .create_if_missing(true)

crates/core/src/message.rs

@@ -1,9 +1,6 @@
 //! Main message type which encapsulated all the messaging between nodes.
 
-use std::{
-    fmt::Display,
-    time::{Duration, SystemTime},
-};
+use std::{fmt::Display, time::Duration};
 
 use serde::{Deserialize, Serialize};
 use uuid::{
@@ -45,13 +42,7 @@ static UUID_CONTEXT: Context = Context::new(14);
 impl Transaction {
     pub fn new(ty: TransactionTypeId, initial_peer: &PeerKey) -> Transaction {
         // using v1 UUID to keep to keep track of the creation ts
-        let now = SystemTime::now()
-            .duration_since(SystemTime::UNIX_EPOCH)
-            .expect("infallible");
-        let now_secs = now.as_secs();
-        let now_nanos = now.as_nanos();
-        let now_nanos = now_nanos - (now_secs as u128 * 1_000_000_000);
-        let ts = Timestamp::from_unix(&UUID_CONTEXT, now_secs, now_nanos as u32);
+        let ts: Timestamp = uuid::timestamp::Timestamp::now(&UUID_CONTEXT);
 
         // event in the net this UUID should be unique since peer keys are unique
         // however some id collision may be theoretically possible if two transactions
@@ -61,8 +52,8 @@ impl Transaction {
         let b = &mut [0; 6];
         b.copy_from_slice(&initial_peer.to_bytes()[0..6]);
         let id = Uuid::new_v1(ts, b);
-        // 2 word size for 64-bits platforms most likely since msg type
-        // probably will be aligned to 64 bytes
+
+        // 3 word size for 64-bits platforms
         Self { id, ty }
     }
 
@@ -231,6 +222,11 @@ impl Message {
             Canceled(_) => true,
         }
     }
+
+    pub fn track_stats(&self) -> bool {
+        use Message::*;
+        !matches!(self, JoinRing(_) | Subscribe(_) | Canceled(_))
+    }
 }
 
 impl Display for Message {