Fix/adapter2

Cargo.toml

@@ -83,6 +83,7 @@ tracing = "0.1"
 tracing-appender = "0.2.3"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 tracing-test = "0.2"
+trait-variant = "0.1.2"
 triomphe = "0.1.11"
 weedb = "0.1.1"
 x509-parser = "0.15"

cli/src/node/mod.rs

@@ -494,7 +494,7 @@ impl Node {
 
         let collation_manager = CollationManager::start(
             collation_config,
-            Arc::new(MessageQueueAdapterStdImpl::new()),
+            Arc::new(MessageQueueAdapterStdImpl::default()),
             |listener| StateNodeAdapterStdImpl::new(listener, self.storage.clone()),
             MempoolAdapterStdImpl::new,
             ValidatorStdImplFactory {
@@ -591,7 +591,7 @@ impl BlockProvider for CollatorBlockProvider {
     type GetBlockFut<'a> = BoxFuture<'a, OptionalBlockStuff>;
 
     fn get_next_block<'a>(&'a self, prev_block_id: &'a BlockId) -> Self::GetNextBlockFut<'a> {
-        self.adapter.wait_for_block(prev_block_id)
+        self.adapter.wait_for_block_next(prev_block_id)
     }
 
     fn get_block<'a>(&'a self, block_id: &'a BlockId) -> Self::GetBlockFut<'a> {

collator/Cargo.toml

@@ -22,7 +22,7 @@ tokio = { workspace = true, features = ["macros", "rt", "signal"] }
 tokio-util = { workspace = true }
 tracing = { workspace = true }
 tracing-subscriber = { workspace = true }
-
+trait-variant = { workspace = true }
 everscale-types = { workspace = true }
 everscale-crypto = { workspace = true }
 
@@ -33,6 +33,7 @@ tycho-storage = { workspace = true }
 tycho-util = { workspace = true }
 tycho-block-util = { workspace = true }
 
+
 [dev-dependencies]
 tempfile = { workspace = true }
 tokio = { version = "1", features = ["rt-multi-thread"] }

collator/src/collator/build_block.rs

@@ -4,7 +4,6 @@ use anyhow::{bail, Result};
 use everscale_types::merkle::*;
 use everscale_types::models::*;
 use everscale_types::prelude::*;
-use sha2::digest::typenum::private::IsGreaterPrivate;
 use sha2::Digest;
 use tycho_block_util::config::BlockchainConfigExt;
 use tycho_block_util::state::ShardStateStuff;
@@ -434,8 +433,6 @@ impl CollatorStdImpl {
             timer.elapsed().as_millis(),
         );
 
-        // do not need to calc out_queue_updates
-
         Ok(state_update)
     }
 }

collator/src/internal_queue/persistent/persistent_state.rs

@@ -4,7 +4,8 @@ use crate::internal_queue::types::ext_types_stubs::EnqueuedMessage;
 use everscale_types::models::BlockIdShort;
 use std::sync::Arc;
 
-pub trait PersistentState<S>
+#[trait_variant::make(PersistentState: Send)]
+pub trait LocalPersistentState<S>
 where
     S: StateSnapshot,
 {

collator/src/internal_queue/queue.rs

@@ -8,7 +8,8 @@ use std::marker::PhantomData;
 use std::sync::Arc;
 use tokio::sync::{Mutex, RwLock};
 
-pub trait Queue<SS, PS>
+#[trait_variant::make(Queue: Send)]
+pub trait LocalQueue<SS, PS>
 where
     SS: StateSnapshot + 'static,
     PS: StateSnapshot + 'static,
@@ -24,6 +25,7 @@ where
         diff_id: &BlockIdShort,
     ) -> Result<Option<Arc<QueueDiff>>, QueueError>;
 }
+
 pub struct QueueImpl<S, P, SS, PS>
 where
     S: SessionState<SS>,
@@ -39,10 +41,10 @@ where
 
 impl<S, P, SS, PS> Queue<SS, PS> for QueueImpl<S, P, SS, PS>
 where
-    S: SessionState<SS>,
-    P: PersistentState<PS>,
-    SS: StateSnapshot + 'static,
-    PS: StateSnapshot + 'static,
+    S: SessionState<SS> + Send,
+    P: PersistentState<PS> + Send + Sync,
+    SS: StateSnapshot + 'static + Send + Sync,
+    PS: StateSnapshot + 'static + Send + Sync,
 {
     fn new(base_shard: ShardIdent) -> Self {
         let session_state = Mutex::new(S::new(base_shard));
@@ -106,8 +108,8 @@ mod tests {
             PersistentStateImpl,
             SessionStateSnapshot,
             PersistentStateSnapshot,
-        > = QueueImpl::new(base_shard);
+        > = <QueueImpl<_, _, _, _> as Queue<_, _>>::new(base_shard);
 
-        queue.split_shard(&base_shard).await.unwrap();
+        Queue::split_shard(&queue, &base_shard).await.unwrap();
     }
 }

collator/src/internal_queue/session/session_state.rs

@@ -8,7 +8,8 @@ use std::collections::HashMap;
 use std::sync::Arc;
 use tokio::sync::RwLock;
 
-pub trait SessionState<S>
+#[trait_variant::make(SessionState: Send)]
+pub trait LocalSessionState<S>
 where
     S: StateSnapshot,
 {

collator/src/internal_queue/session/session_state_snapshot.rs

@@ -6,6 +6,7 @@ use everscale_types::models::ShardIdent;
 use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::Arc;
+use tracing::error;
 
 pub struct SessionStateSnapshot {
     pub flat_shards: HashMap<ShardIdent, Shard>,
@@ -39,7 +40,7 @@ impl StateSnapshot for SessionStateSnapshot {
                         }
                     }
                     Err(e) => {
-                        println!("failed to convert account to hashbytes {e:?}");
+                        error!("failed to convert account to hashbytes {e:?}");
                         return false;
                     }
                 }

collator/src/internal_queue/snapshot.rs

@@ -40,7 +40,7 @@ pub struct ShardRange {
     pub to_lt: Option<Lt>,
 }
 
-pub trait StateSnapshot {
+pub trait StateSnapshot: Send {
     fn get_outgoing_messages_by_shard(
         &self,
         shards: &mut HashMap<ShardIdent, ShardRange>,

collator/src/lib.rs

@@ -12,5 +12,3 @@ pub mod test_utils;
 
 mod tracing_targets;
 mod utils;
-
-// pub use validator::test_impl as validator_test_impl;

collator/src/manager/mod.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 
 use anyhow::{anyhow, bail, Result};
 use async_trait::async_trait;
-use everscale_types::models::{BlockId, BlockInfo, ShardIdent, ValidatorDescription, ValueFlow};
+use everscale_types::models::{BlockId, BlockInfo, ShardIdent, ValueFlow};
 use tycho_block_util::block::ValidatorSubsetInfo;
 use tycho_block_util::state::{MinRefMcStateTracker, ShardStateStuff};
 
@@ -112,7 +112,7 @@ where
     CF: CollatorFactory,
     V: Validator,
 {
-    async fn on_block_accepted(&self, block_id: &BlockId) -> Result<()> {
+    async fn on_block_accepted(&self, _block_id: &BlockId) -> Result<()> {
         //TODO: remove accepted block from cache
         //STUB: do nothing, currently we remove block from cache when it sent to state node
         Ok(())
@@ -453,7 +453,7 @@ where
 
     /// * TRUE - provided `mc_block_id` is before or equal to last processed
     /// * FALSE - provided `mc_block_id` is ahead of last processed
-    fn check_if_mc_block_not_ahead_last_processed(&self, mc_block_id: &BlockId) -> bool {
+    fn _check_if_mc_block_not_ahead_last_processed(&self, mc_block_id: &BlockId) -> bool {
         //TODO: consider block shard?
         let last_processed_mc_block_id_opt = self.last_processed_mc_block_id();
         let is_not_ahead = matches!(last_processed_mc_block_id_opt, Some(last_processed_mc_block_id)
@@ -858,8 +858,7 @@ where
             candidate_id.as_short_id(),
             candidate_chain_time,
         );
-        let _handle = self
-            .validator
+        self.validator
             .validate(candidate_id, session_info.seqno())
             .await?;
 

collator/src/msg_queue/mod.rs

@@ -1,3 +1,6 @@
+#![allow(warnings)]
+#![allow(clippy::all)]
+
 use std::sync::Arc;
 
 use anyhow::Result;
@@ -56,8 +59,8 @@ pub struct MessageQueueAdapterStdImpl {
     msg_queue: MsgQueueStdImpl,
 }
 
-impl MessageQueueAdapterStdImpl {
-    pub fn new() -> Self {
+impl Default for MessageQueueAdapterStdImpl {
+    fn default() -> Self {
         let base_shard = ShardIdent::new_full(0);
         Self {
             msg_queue: MsgQueueStdImpl::new(base_shard),

collator/src/msg_queue/state_persistent.rs

@@ -20,10 +20,10 @@ where
     ST: PersistentStateService,
     DB: StorageService,
 {
-    fn some_internal_method_for_persistent_state(&mut self) -> Result<()> {
+    fn _some_internal_method_for_persistent_state(&mut self) -> Result<()> {
         todo!()
     }
-    pub(super) fn some_module_internal_method_for_persistent_state(&mut self) -> Result<()> {
+    pub(super) fn _some_module_internal_method_for_persistent_state(&mut self) -> Result<()> {
         todo!()
     }
 }