Merge remote-tracking branch 'origin/mempool'

Cargo.toml

@@ -57,6 +57,7 @@ pkcs8 = "0.10"
 quick_cache = "0.4.1"
 quinn = { version = "0.10", default-features = false, features = ["runtime-tokio", "tls-rustls"] }
 rand = "0.8"
+rand_pcg = { version = "0.3" }
 rcgen = "0.11"
 ring = "0.16"
 rlimit = "0.10.1"

cli/Cargo.toml

@@ -47,6 +47,9 @@ tycho-network = { workspace = true }
 tycho-storage = { workspace = true }
 tycho-util = { workspace = true }
 
+[dev-dependencies]
+tycho-collator = { workspace = true, features = ["test"] }
+
 [build-dependencies]
 anyhow = { workspace = true }
 rustc_version = { workspace = true }

cli/src/node/mod.rs

@@ -12,7 +12,7 @@ use futures_util::future::BoxFuture;
 use tycho_block_util::state::{MinRefMcStateTracker, ShardStateStuff};
 use tycho_collator::collator::CollatorStdImplFactory;
 use tycho_collator::manager::CollationManager;
-use tycho_collator::mempool::MempoolAdapterStdImpl;
+use tycho_collator::mempool::MempoolAdapterFactoryStd;
 use tycho_collator::msg_queue::MessageQueueAdapterStdImpl;
 use tycho_collator::state_node::{StateNodeAdapter, StateNodeAdapterStdImpl};
 use tycho_collator::types::{CollationConfig, ValidatorNetwork};
@@ -486,14 +486,19 @@ impl Node {
             supported_block_version: 50,
             supported_capabilities: supported_capabilities(),
             max_collate_threads: 1,
-            // test_validators_keypairs: vec![],
+            #[cfg(test)]
+            test_validators_keypairs: vec![],
         };
 
         let collation_manager = CollationManager::start(
             collation_config,
             Arc::new(MessageQueueAdapterStdImpl::default()),
             |listener| StateNodeAdapterStdImpl::new(listener, self.storage.clone()),
-            MempoolAdapterStdImpl::new,
+            MempoolAdapterFactoryStd::new(
+                self.keypair.clone(),
+                self.dht_client.clone(),
+                self.overlay_service.clone(),
+            ),
             ValidatorStdImplFactory {
                 network: ValidatorNetwork {
                     overlay_service: self.overlay_service.clone(),

collator/Cargo.toml

@@ -14,6 +14,7 @@ anyhow = { workspace = true }
 async-trait = { workspace = true }
 bytesize = { workspace = true }
 futures-util = { workspace = true }
+parking_lot = { workspace = true }
 rand = { workspace = true }
 sha2 = { workspace = true }
 tl-proto = { workspace = true }
@@ -28,11 +29,11 @@ everscale-crypto = { workspace = true }
 
 # local deps
 #tycho-consensus = { workspace = true }
+tycho-block-util = { workspace = true }
+tycho-consensus = { workspace = true }
 tycho-network = { workspace = true }
 tycho-storage = { workspace = true }
 tycho-util = { workspace = true }
-tycho-block-util = { workspace = true }
-
 
 [dev-dependencies]
 tempfile = { workspace = true }

collator/src/internal_queue/session/session_state.rs

@@ -89,6 +89,8 @@ mod tests {
     use everscale_types::models::BlockIdShort;
 
     use super::*;
+    use crate::internal_queue::session::session_state;
+    use crate::internal_queue::session::session_state::*;
     use crate::internal_queue::types::ext_types_stubs::{
         EnqueuedMessage, MessageContent, MessageEnvelope,
     };
@@ -113,14 +115,17 @@ mod tests {
     #[test]
     fn test_new_session_state() {
         let base_shard = test_shard_ident();
-        let _session_state = SessionStateImpl::new(base_shard);
+        let _session_state =
+            <SessionStateImpl as SessionState<SessionStateSnapshot>>::new(base_shard);
     }
 
     #[tokio::test]
     async fn test_split_shard() {
         let base_shard = test_shard_ident();
-        let session_state = SessionStateImpl::new(base_shard);
-        let split_shard_result = session_state.split_shard(&base_shard).await;
+        let session_state =
+            <SessionStateImpl as SessionState<SessionStateSnapshot>>::new(base_shard);
+        let split_shard_result =
+            SessionState::<SessionStateSnapshot>::split_shard(&session_state, &base_shard).await;
         assert!(
             split_shard_result.is_ok(),
             "Splitting the shard should succeed."
@@ -130,7 +135,8 @@ mod tests {
     #[tokio::test]
     async fn test_apply_diff() {
         let base_shard = test_shard_ident();
-        let session_state = SessionStateImpl::new(base_shard);
+        let session_state =
+            <SessionStateImpl as SessionState<SessionStateSnapshot>>::new(base_shard);
         let block_id = BlockIdShort {
             shard: base_shard,
             seqno: 0,
@@ -140,7 +146,8 @@ mod tests {
             messages: vec![default_message()],
             processed_upto: Default::default(),
         });
-        let apply_diff_result = session_state.apply_diff(diff).await;
+        let apply_diff_result =
+            SessionState::<SessionStateSnapshot>::apply_diff(&session_state, diff).await;
         assert_eq!(
             session_state
                 .shards_flat
@@ -173,12 +180,14 @@ mod tests {
     #[tokio::test]
     async fn test_remove_diff() {
         let base_shard = test_shard_ident();
-        let session_state = SessionStateImpl::new(base_shard);
+        let session_state =
+            <SessionStateImpl as SessionState<SessionStateSnapshot>>::new(base_shard);
         let diff_id = BlockIdShort {
             shard: base_shard,
             seqno: 0,
         };
-        let remove_diff_result = session_state.remove_diff(&diff_id).await;
+        let remove_diff_result =
+            session_state::SessionState::remove_diff(&session_state, &diff_id).await;
         assert_eq!(
             session_state
                 .shards_flat
@@ -211,7 +220,8 @@ mod tests {
     #[tokio::test]
     async fn test_snapshot() {
         let base_shard = test_shard_ident();
-        let session_state = SessionStateImpl::new(base_shard);
+        let session_state =
+            <SessionStateImpl as SessionState<SessionStateSnapshot>>::new(base_shard);
         let block_id = BlockIdShort {
             shard: base_shard,
             seqno: 0,
@@ -221,9 +231,10 @@ mod tests {
             messages: vec![default_message()],
             processed_upto: Default::default(),
         });
-        let _apply_diff_result = session_state.apply_diff(diff).await;
+        let _apply_diff_result =
+            session_state::SessionState::apply_diff(&session_state, diff).await;
 
-        let snapshot = session_state.snapshot().await;
+        let snapshot = session_state::SessionState::snapshot(&session_state).await;
         assert_eq!(snapshot.flat_shards.len(), 1);
         assert_eq!(
             snapshot

collator/src/manager/mod.rs

@@ -223,7 +223,7 @@ where
             Arc::new(state_node_adapter_factory.create(arc_dispatcher.clone()));
 
         // create mempool adapter
-        let mpool_adapter = Arc::new(mpool_adapter_factory.create(arc_dispatcher.clone()));
+        let mpool_adapter = mpool_adapter_factory.create(arc_dispatcher.clone());
 
         // create validator and start its tasks queue
         let validator = validator_factory.create(ValidatorContext {