feature(collator): queue diffs tail

Cargo.toml

@@ -136,6 +136,7 @@ tycho-util = { path = "./util", version = "0.1.4" }
 
 [patch.crates-io]
 weedb = { version = "0.3.8", git = "https://github.com/broxus/weedb.git", branch = "next-rocksdb" }
+everscale-types = { git = "https://github.com/broxus/everscale-types.git", rev = "8ea6b16e81f1ffd88006263327004b891b5ef47c"}
 
 [workspace.lints.rust]
 future_incompatible = "warn"
@@ -241,4 +242,4 @@ opt-level = 3
 [profile.dev.package.hashbrown]
 opt-level = 3
 [profile.dev.package."*"]
-opt-level = 1
+opt-level = 1

block-util/src/block/block_proof_stuff.rs

@@ -43,6 +43,7 @@ impl BlockProofStuff {
             state_update: Lazy::new(&MerkleUpdate::default()).unwrap(),
             out_msg_queue_updates: OutMsgQueueUpdates {
                 diff_hash: Default::default(),
+                tail_len: 0,
             },
             extra: Lazy::new(&BlockExtra::default()).unwrap(),
         };

block-util/src/block/block_stuff.rs

@@ -55,6 +55,7 @@ impl BlockStuff {
             state_update: Lazy::new(&MerkleUpdate::default()).unwrap(),
             out_msg_queue_updates: OutMsgQueueUpdates {
                 diff_hash: Default::default(),
+                tail_len: 0,
             },
             extra: Lazy::new(&BlockExtra::default()).unwrap(),
         };

block-util/src/lib.rs

@@ -4,5 +4,4 @@ pub mod config;
 pub mod dict;
 pub mod queue;
 pub mod state;
-
 pub mod tl;

block-util/src/queue/proto.rs

@@ -22,7 +22,8 @@ pub struct QueueDiff {
     /// Seqno of the corresponding block.
     pub seqno: u32,
     /// collator boundaries.
-    pub processed_upto: BTreeMap<ShardIdent, QueueKey>,
+    // TODO: should rename field in `proto.tl` on network reset
+    pub processed_to: BTreeMap<ShardIdent, QueueKey>,
     /// Min message queue key.
     pub min_message: QueueKey,
     /// Max message queue key.
@@ -55,7 +56,7 @@ impl TlWrite for QueueDiff {
         4 + tl::hash_bytes::SIZE_HINT
             + tl::shard_ident::SIZE_HINT
             + 4
-            + processed_upto_map::size_hint(&self.processed_upto)
+            + processed_to_map::size_hint(&self.processed_to)
             + 2 * QueueKey::SIZE_HINT
             + messages_list::size_hint(&self.messages)
     }
@@ -68,7 +69,7 @@ impl TlWrite for QueueDiff {
         tl::hash_bytes::write(&self.prev_hash, packet);
         tl::shard_ident::write(&self.shard_ident, packet);
         packet.write_u32(self.seqno);
-        processed_upto_map::write(&self.processed_upto, packet);
+        processed_to_map::write(&self.processed_to, packet);
         self.min_message.write_to(packet);
         self.max_message.write_to(packet);
         messages_list::write(&self.messages, packet);
@@ -90,7 +91,7 @@ impl<'tl> TlRead<'tl> for QueueDiff {
             prev_hash: tl::hash_bytes::read(data, offset)?,
             shard_ident: tl::shard_ident::read(data, offset)?,
             seqno: u32::read_from(data, offset)?,
-            processed_upto: processed_upto_map::read(data, offset)?,
+            processed_to: processed_to_map::read(data, offset)?,
             min_message: QueueKey::read_from(data, offset)?,
             max_message: QueueKey::read_from(data, offset)?,
             messages: messages_list::read(data, offset)?,
@@ -205,7 +206,7 @@ impl std::fmt::Display for QueueKey {
     }
 }
 
-mod processed_upto_map {
+mod processed_to_map {
     use tl_proto::{TlPacket, TlResult};
 
     use super::*;
@@ -222,9 +223,9 @@ mod processed_upto_map {
     pub fn write<P: TlPacket>(items: &BTreeMap<ShardIdent, QueueKey>, packet: &mut P) {
         packet.write_u32(items.len() as u32);
 
-        for (shard_ident, processed_upto) in items {
+        for (shard_ident, processed_to) in items {
             tl::shard_ident::write(shard_ident, packet);
-            processed_upto.write_to(packet);
+            processed_to.write_to(packet);
         }
     }
 
@@ -445,7 +446,7 @@ mod tests {
             prev_hash: HashBytes::from([0x33; 32]),
             shard_ident: ShardIdent::MASTERCHAIN,
             seqno: 123,
-            processed_upto: BTreeMap::from([
+            processed_to: BTreeMap::from([
                 (ShardIdent::MASTERCHAIN, QueueKey {
                     lt: 1,
                     hash: HashBytes::from([0x11; 32]),
@@ -491,7 +492,7 @@ mod tests {
                 prev_hash,
                 shard_ident: ShardIdent::MASTERCHAIN,
                 seqno,
-                processed_upto: BTreeMap::from([
+                processed_to: BTreeMap::from([
                     (ShardIdent::MASTERCHAIN, QueueKey {
                         lt: 10 * seqno as u64,
                         hash: HashBytes::from([seqno as u8; 32]),

block-util/src/queue/queue_diff.rs

@@ -27,11 +27,11 @@ impl QueueDiffStuffBuilder {
     }
 
     // TODO: Use iterator of `(ShardIdent, QueueKey)`?
-    pub fn with_processed_upto<'a, I>(mut self, processed_upto: I) -> Self
+    pub fn with_processed_to<'a, I>(mut self, processed_to: I) -> Self
     where
         I: IntoIterator<Item = (ShardIdent, u64, &'a HashBytes)>,
     {
-        self.inner_mut().diff.processed_upto = processed_upto
+        self.inner_mut().diff.processed_to = processed_to
             .into_iter()
             .map(|(shard_ident, lt, hash)| (shard_ident, QueueKey { lt, hash: *hash }))
             .collect();
@@ -79,6 +79,14 @@ impl SerializedQueueDiff {
         &self.inner.diff.hash
     }
 
+    pub fn processed_to(&self) -> impl Iterator<Item = (ShardIdent, &QueueKey)> {
+        self.inner
+            .diff
+            .processed_to
+            .iter()
+            .map(|(shard_ident, key)| (*shard_ident, key))
+    }
+
     fn inner_mut(&mut self) -> &mut Inner {
         Arc::get_mut(&mut self.inner).expect("inner is not shared")
     }
@@ -102,7 +110,7 @@ impl QueueDiffStuff {
                     prev_hash: HashBytes::ZERO,
                     shard_ident: block_id.shard,
                     seqno: block_id.seqno,
-                    processed_upto: BTreeMap::from([(block_id.shard, QueueKey::MIN)]),
+                    processed_to: BTreeMap::from([(block_id.shard, QueueKey::MIN)]),
                     min_message: QueueKey::MIN,
                     max_message: QueueKey::MIN,
                     messages: Vec::new(),
@@ -134,7 +142,7 @@ impl QueueDiffStuff {
                     prev_hash: *prev_hash,
                     shard_ident,
                     seqno,
-                    processed_upto: Default::default(),
+                    processed_to: Default::default(),
                     min_message: Default::default(),
                     max_message: Default::default(),
                     messages: Default::default(),
@@ -361,7 +369,7 @@ mod tests {
                     prev_hash: HashBytes::ZERO,
                     shard_ident: ShardIdent::BASECHAIN,
                     seqno: 1,
-                    processed_upto: Default::default(),
+                    processed_to: Default::default(),
                     min_message: QueueKey {
                         lt: 0,
                         hash: message_hashes[0],

cli/src/node/mod.rs

@@ -11,8 +11,8 @@ use tycho_block_util::block::BlockIdRelation;
 use tycho_block_util::state::MinRefMcStateTracker;
 use tycho_collator::collator::CollatorStdImplFactory;
 use tycho_collator::internal_queue::queue::{QueueConfig, QueueFactory, QueueFactoryStdImpl};
-use tycho_collator::internal_queue::state::persistent_state::PersistentStateImplFactory;
-use tycho_collator::internal_queue::state::session_state::SessionStateImplFactory;
+use tycho_collator::internal_queue::state::commited_state::CommittedStateImplFactory;
+use tycho_collator::internal_queue::state::uncommitted_state::UncommittedStateImplFactory;
 use tycho_collator::manager::CollationManager;
 use tycho_collator::mempool::MempoolAdapterStdImpl;
 use tycho_collator::queue_adapter::MessageQueueAdapterStdImpl;
@@ -316,12 +316,12 @@ impl Node {
         // Create collator
         tracing::info!("starting collator");
 
-        let session_state_factory = SessionStateImplFactory::new(self.storage.clone());
-        let persistent_state_factory = PersistentStateImplFactory::new(self.storage.clone());
+        let session_state_factory = UncommittedStateImplFactory::new(self.storage.clone());
+        let persistent_state_factory = CommittedStateImplFactory::new(self.storage.clone());
 
         let queue_factory = QueueFactoryStdImpl {
-            session_state_factory,
-            persistent_state_factory,
+            uncommitted_state_factory: session_state_factory,
+            committed_state_factory: persistent_state_factory,
             config: self.internal_queue_config,
         };
         let queue = queue_factory.create();