Remp queue test1

CHANGELOG.md

@@ -4,6 +4,7 @@ All notable changes to this project will be documented in this file.
 
 ## Version 0.58.4
 
+- Load cells from storing cells cache by id
 - Added support for due payment fix
 - Bump block version
 

Cargo.toml

@@ -44,10 +44,12 @@ name = 'zerostate'
 path = 'bin/zerostate.rs'
 
 [dependencies]
+adnl = { features = [ 'client', 'node', 'server' ], git = 'https://github.com/everx-labs/ever-adnl.git', tag = '0.10.24' }
 arc-swap = '0.3.11'
 async-recursion = '0.3.2'
 async-trait = '0.1.22'
 bitflags = '1.2.1'
+catchain = { path = 'catchain' }
 chrono = '=0.4.19'
 clap = '2.33'
 colored = '1.9.3'
@@ -58,12 +60,18 @@ deflate = '1.0.0'
 dirs = '2.0.2'
 enum-as-inner = '=0.5.1'
 env_logger = '0.7.1'
+ever_abi = { git = 'https://github.com/everx-labs/ever-abi.git', tag = '2.5.3' }
+ever_block = { git = 'https://github.com/everx-labs/ever-block.git', tag = '1.10.2' }
+ever_block_json = { git = 'https://github.com/everx-labs/ever-block-json.git', tag = '0.8.6' }
+ever_executor = { git = 'https://github.com/everx-labs/ever-executor.git', tag = '1.17.6' }
+ever_vm = { git = 'https://github.com/everx-labs/ever-vm.git', tag = '2.1.5' }
 failure = '0.1'
 futures = '0.3.1'
 futures-timer = '3.0.1'
 hex = '0.4'
 inflate = '0.4.5'
 lazy_static = '1.4.0'
+lockfree = { git = 'https://github.com/everx-labs/lockfree.git' }
 log = '0.4'
 log4rs = '1.2'
 log4rs-rolling-file = '0.2.0'
@@ -82,19 +90,11 @@ serde_json = '1.0.64'
 shell-words = '1.0'
 spin = '0.7.1'
 statsd = { optional = true, version = '0.15' }
+storage = { path = 'storage' }
 stream-cancel = '0.8.0'
 string-builder = '^0.2.0'
 tokio = { features = [ 'rt-multi-thread' ], version = '1.5' }
 tokio-util = '0.7'
-adnl = { features = [ 'client', 'node', 'server' ], git = 'https://github.com/everx-labs/ever-adnl.git', tag = '0.10.24' }
-catchain = { path = 'catchain' }
-ever_abi = { git = 'https://github.com/everx-labs/ever-abi.git', tag = '2.5.3' }
-ever_block = { git = 'https://github.com/everx-labs/ever-block.git', tag = '1.10.2' }
-ever_block_json = { git = 'https://github.com/everx-labs/ever-block-json.git', tag = '0.8.6' }
-ever_executor = { git = 'https://github.com/everx-labs/ever-executor.git', tag = '1.17.6' }
-ever_vm = { git = 'https://github.com/everx-labs/ever-vm.git', tag = '2.1.5' }
-lockfree = { git = 'https://github.com/everx-labs/lockfree.git' }
-storage = { path = 'storage' }
 ton_api = { git = 'https://github.com/everx-labs/ever-tl.git', package = 'ton_api', tag = '0.3.79' }
 validator_session = { path = 'validator-session' }
 

catchain/Cargo.toml

@@ -5,8 +5,10 @@ name = 'catchain'
 version = '0.1.0'
 
 [dependencies]
+adnl = { features = [ 'node' ], git = 'https://github.com/everx-labs/ever-adnl.git', tag = '0.10.24' }
 chrono = '0.4.10'
 crossbeam = '0.7.3'
+ever_block = { git = 'https://github.com/everx-labs/ever-block.git', tag = '1.10.2' }
 failure = '0.1'
 futures = '0.3.4'
 hex = '0.4'
@@ -18,10 +20,8 @@ metrics-util = '0.15.0'
 quanta = '0.11.1'
 rand = '0.8'
 regex = '1.3.1'
-tokio = { features = [ 'rt-multi-thread' ], version = '1.5' }
-adnl = { features = [ 'node' ], git = 'https://github.com/everx-labs/ever-adnl.git', tag = '0.10.24' }
-ever_block = { git = 'https://github.com/everx-labs/ever-block.git', tag = '1.10.2' }
 storage = { path = '../storage' }
+tokio = { features = [ 'rt-multi-thread' ], version = '1.5' }
 ton_api = { git = 'https://github.com/everx-labs/ever-tl.git', package = 'ton_api', tag = '0.3.79' }
 
 [dev-dependencies]

src/collator_test_bundle.rs

@@ -39,7 +39,7 @@ use storage::StorageTelemetry;
 use ever_block::{
     BlockIdExt, Message, ShardIdent, Serializable, MerkleUpdate, Deserializable, 
     ValidatorBaseInfo, BlockSignaturesPure, BlockSignatures, HashmapAugType, 
-    TopBlockDescrSet, GlobalCapabilities, OutMsgQueue,
+    TopBlockDescrSet, OutMsgQueue,
 };
 use ever_block::{ShardStateUnsplit, TopBlockDescr};
 use ever_block::{UInt256, fail, error, Result, CellType, read_boc, read_single_root_boc};
@@ -1208,19 +1208,6 @@ impl CollatorTestBundle {
 
     pub fn candidate(&self) -> Option<&BlockCandidate> { self.candidate.as_ref() }
     pub fn set_notes(&mut self, notes: String) { self.index.notes = notes }
-
-    fn get_messages(&self, remp: bool) -> Result<Vec<(Arc<Message>, UInt256)>> {
-        let remp_enabled = self.states
-            .get(&self.index.last_mc_state)
-            .ok_or_else(|| error!("Can't load last ms block to read config"))?
-            .config_params()?.has_capability(GlobalCapabilities::CapRemp);
-
-        if remp_enabled == remp {
-            Ok(self.external_messages.clone())
-        } else {
-            Ok(vec!())
-        }
-    }
 }
 
 // Is used instead full node's engine for run tests
@@ -1346,20 +1333,6 @@ impl EngineOperations for CollatorTestBundle {
         &self.allocated
     }
 
-    fn get_remp_messages(&self, _shard: &ShardIdent) -> Result<Vec<(Arc<Message>, UInt256)>> {
-        self.get_messages(true)
-    }
-
-    fn finalize_remp_messages(
-        &self,
-        _block: BlockIdExt,
-        _accepted: Vec<UInt256>,
-        _rejected: Vec<(UInt256, String)>,
-        _ignored: Vec<UInt256>,
-    ) -> Result<()> {
-        Ok(())
-    }
-
     async fn check_remp_duplicate(&self, _message_id: &UInt256) -> Result<RempDuplicateStatus> {
         Ok(RempDuplicateStatus::Fresh(UInt256::default()))
     }

src/config.rs

@@ -166,8 +166,7 @@ pub struct TonNodeConfig {
     port: Option<u16>,
     #[serde(skip)]
     file_name: String,
-    #[serde(default = "RempConfig::default")]
-    remp: RempConfig,
+    remp: Option<RempConfig>,
     #[serde(default)]
     restore_db: bool,
     #[serde(default)]
@@ -599,8 +598,11 @@ impl TonNodeConfig {
     pub fn extensions(&self) -> &NodeExtensions {
         &self.extensions
     }
-    pub fn remp_config(&self) -> &RempConfig {
-        &self.remp
+    pub fn remp_config(&self) -> RempConfig {
+        match &self.remp {
+            Some(x) => x.clone(),
+            None => RempConfig::default()
+        }
     }
     pub fn restore_db(&self) -> bool {
         self.restore_db

src/engine_operations.rs

@@ -990,18 +990,6 @@ impl EngineOperations for Engine {
     fn new_remp_message(&self, id: UInt256, message: Arc<Message>) -> Result<()> {
         self.remp_messages()?.new_message(id, message)
     }
-    fn get_remp_messages(&self, shard: &ShardIdent) -> Result<Vec<(Arc<Message>, UInt256)>> {
-        self.remp_messages()?.get_messages(shard)
-    }
-    fn finalize_remp_messages(
-        &self,
-        block: BlockIdExt,
-        accepted: Vec<UInt256>,
-        rejected: Vec<(UInt256, String)>,
-        ignored: Vec<UInt256>,
-    ) -> Result<()> {
-        self.remp_messages()?.finalize_messages(block, accepted, rejected, ignored)
-    }
     fn finalize_remp_messages_as_ignored(&self, block_id: &BlockIdExt)
     -> Result<()> {
         self.remp_messages()?.finalize_remp_messages_as_ignored(block_id)
@@ -1013,7 +1001,6 @@ impl EngineOperations for Engine {
     async fn check_remp_duplicate(&self, message_id: &UInt256) -> Result<RempDuplicateStatus> {
         self.remp_service()
             .ok_or_else(|| error!("Can't get message status because remp service was not set"))?
-            .remp_core_interface()?
             .check_remp_duplicate(message_id)
     }
 

src/engine_traits.rs

@@ -628,18 +628,6 @@ pub trait EngineOperations : Sync + Send {
     fn new_remp_message(&self, id: UInt256, message: Arc<Message>) -> Result<()> {
         unimplemented!()
     }
-    fn get_remp_messages(&self, shard: &ShardIdent) -> Result<Vec<(Arc<Message>, UInt256)>> {
-        unimplemented!()
-    }
-    fn finalize_remp_messages(
-        &self,
-        block: BlockIdExt,
-        accepted: Vec<UInt256>,
-        rejected: Vec<(UInt256, String)>,
-        ignored: Vec<UInt256>,
-    ) -> Result<()> {
-        unimplemented!()
-    }
     fn finalize_remp_messages_as_ignored(&self, block_id: &BlockIdExt) -> Result<()> {
         unimplemented!()
     }
@@ -968,6 +956,16 @@ pub enum RempDuplicateStatus {
 
 #[async_trait::async_trait]
 pub trait RempCoreInterface: Sync + Send {
-    async fn process_incoming_message(&self, message_id: UInt256, message: Message, source: Arc<KeyId>) -> Result<()>;
+    async fn process_incoming_message(&self, message: &RempMessage, source: Arc<KeyId>) -> Result<()>;
     fn check_remp_duplicate(&self, message_id: &UInt256) -> Result<RempDuplicateStatus>;
 }
+
+#[async_trait::async_trait]
+pub trait RempQueueCollatorInterface : Send + Sync {
+    async fn init_queue(
+        &self,
+        master_block_id: &BlockIdExt,
+        prev_blocks_ids: &[&BlockIdExt]
+    ) -> Result<()>;
+    async fn get_next_message_for_collation(&self) -> Result<Option<(Arc<Message>, UInt256)>>;
+}

src/ext_messages.rs

@@ -27,7 +27,7 @@ const MESSAGE_LIFETIME: u32 = 600; // seconds
 const MESSAGE_MAX_GENERATIONS: u8 = 3;
 
 const MAX_EXTERNAL_MESSAGE_DEPTH: u16 = 512;
-const MAX_EXTERNAL_MESSAGE_SIZE: usize = 65535;
+pub const MAX_EXTERNAL_MESSAGE_SIZE: usize = 65535;
 
 pub const EXT_MESSAGES_TRACE_TARGET: &str = "ext_messages";
 
@@ -460,6 +460,7 @@ pub fn is_finally_accepted(status: &RempMessageStatus) -> bool {
     }
 }
 
+// To be deleted
 pub struct RempMessagesPool {
     messages: Map<UInt256, Arc<Message>>,
     statuses_queue: lockfree::queue::Queue<(UInt256, Arc<Message>, RempMessageStatus)>,
@@ -483,7 +484,7 @@ impl RempMessagesPool {
 
     // Important! If call get_messages with same shard two times in row (without finalize_messages between)
     // the messages returned first call will return second time too.
-    pub fn get_messages(&self, shard: &ShardIdent) -> Result<Vec<(Arc<Message>, UInt256)>> {
+    /*pub fn get_messages(&self, shard: &ShardIdent) -> Result<Vec<(Arc<Message>, UInt256)>> {
         let mut result = vec!();
         let mut ids = String::new();
         for guard in self.messages.iter() {
@@ -506,7 +507,7 @@ impl RempMessagesPool {
         );
 
         Ok(result)
-    }
+    }*/
 
     pub fn finalize_messages(
         &self,

src/tests/test_helper.rs

@@ -697,11 +697,12 @@ impl TestEngine {
             extra.created_by().clone(),
             self.clone(),
             Some(extra.rand_seed().clone()),
+            None,
             CollatorSettings::default(),
         )?;
 
-        let (block_candidate, new_state) = collator.collate().await?;   
-            
+        let (block_candidate, new_state) = collator.collate().await?;
+
         if let Some(res_path) = &self.res_path {
 
             let new_block = Block::construct_from_bytes(&block_candidate.data)?;