Merge branch 'refactor/mempool'

cli/Cargo.toml

@@ -49,6 +49,7 @@ tycho-util = { workspace = true }
 
 [dev-dependencies]
 tycho-collator = { workspace = true, features = ["test"] }
+tycho-storage = { workspace = true, features = ["test"] }
 
 [build-dependencies]
 anyhow = { workspace = true }

collator/src/mempool/mempool_adapter.rs

@@ -1,18 +1,19 @@
-use std::collections::{BTreeMap, HashMap};
+use std::collections::BTreeMap;
 use std::sync::Arc;
 
 use anyhow::Result;
 use async_trait::async_trait;
 use everscale_crypto::ed25519::KeyPair;
 use everscale_types::boc::Boc;
-use everscale_types::cell::HashBytes;
 use everscale_types::models::ExtInMsgInfo;
 use everscale_types::prelude::Load;
 use parking_lot::RwLock;
+use tokio::sync::mpsc;
 use tokio::sync::mpsc::UnboundedReceiver;
 use tycho_block_util::state::ShardStateStuff;
-use tycho_consensus::Point;
+use tycho_consensus::{InputBufferImpl, Point};
 use tycho_network::{DhtClient, OverlayService};
+use tycho_util::FastHashSet;
 
 use crate::mempool::types::ExternalMessage;
 use crate::mempool::{MempoolAnchor, MempoolAnchorId};
@@ -127,9 +128,16 @@ impl MempoolAdapterStdImpl {
         let (sender, receiver) =
             tokio::sync::mpsc::unbounded_channel::<(Arc<Point>, Vec<Arc<Point>>)>();
 
+        // TODO receive from outside
+        let (_externals_tx, externals_rx) = mpsc::unbounded_channel();
+        let mut engine = tycho_consensus::Engine::new(
+            key_pair,
+            &dht_client,
+            &overlay_service,
+            sender,
+            InputBufferImpl::new(externals_rx),
+        );
         tokio::spawn(async move {
-            let mut engine =
-                tycho_consensus::Engine::new(key_pair, &dht_client, &overlay_service, sender);
             // TODO replace with some sensible init before run
             engine.init_with_genesis(&[]).await;
             engine.run().await;
@@ -156,7 +164,8 @@ pub async fn parse_points(
     mut rx: UnboundedReceiver<(Arc<Point>, Vec<Arc<Point>>)>,
 ) {
     while let Some((anchor, points)) = rx.recv().await {
-        let mut external_messages = HashMap::<HashBytes, ExternalMessage>::new();
+        let mut repr_hashes = FastHashSet::default();
+        let mut messages = Vec::new();
 
         for point in points {
             'message: for message in &point.body.payload {
@@ -184,16 +193,12 @@ pub async fn parse_points(
                     }
                 };
 
-                let external_message = ExternalMessage::new(cell.clone(), ext_in_message);
-                external_messages.insert(*cell.repr_hash(), external_message);
+                if repr_hashes.insert(*cell.repr_hash()) {
+                    messages.push(Arc::new(ExternalMessage::new(cell.clone(), ext_in_message)));
+                }
             }
         }
 
-        let messages = external_messages
-            .into_iter()
-            .map(|m| Arc::new(m.1))
-            .collect::<Vec<_>>();
-
         let anchor = Arc::new(MempoolAnchor::new(
             anchor.body.location.round.0,
             anchor.body.time.as_u64(),

consensus/Cargo.toml

@@ -15,6 +15,7 @@ path = "examples/consensus_node.rs"
 [dependencies]
 ahash = { workspace = true }
 anyhow = { workspace = true }
+async-trait = { workspace = true }
 bincode = { workspace = true }
 bytes = { workspace = true, features = ["serde"] }
 dashmap = { workspace = true }
@@ -39,7 +40,7 @@ tracing-appender = { workspace = true }
 # local deps
 tycho-network = { workspace = true }
 tycho-storage = { workspace = true }
-tycho-util = { workspace = true, features = ["test"] }
+tycho-util = { workspace = true }
 
 [dev-dependencies]
 parking_lot = { workspace = true, features = ["deadlock_detection"] }
@@ -50,5 +51,7 @@ tikv-jemallocator = { workspace = true, features = [
     "background_threads",
 ]}
 
+tycho-util = { workspace = true, features = ["test"] }
+
 [lints]
 workspace = true

consensus/examples/consensus_node.rs

@@ -16,7 +16,7 @@ use tokio::sync::mpsc;
 use tracing_subscriber::layer::SubscriberExt;
 use tracing_subscriber::{fmt, EnvFilter, Layer};
 use tycho_consensus::test_utils::drain_anchors;
-use tycho_consensus::Engine;
+use tycho_consensus::{Engine, InputBufferStub};
 use tycho_network::{DhtConfig, NetworkConfig, PeerId, PeerInfo};
 use tycho_util::time::now_sec;
 
@@ -129,7 +129,13 @@ impl CmdRun {
         }
 
         let (committed_tx, committed_rx) = mpsc::unbounded_channel();
-        let mut engine = Engine::new(key_pair.clone(), &dht_client, &overlay, committed_tx);
+        let mut engine = Engine::new(
+            key_pair.clone(),
+            &dht_client,
+            &overlay,
+            committed_tx,
+            InputBufferStub::new(100, 5),
+        );
         engine.init_with_genesis(all_peers.as_slice()).await;
         tokio::spawn(drain_anchors(committed_rx));
 

consensus/src/dag/anchor_stage.rs

@@ -31,6 +31,8 @@ impl AnchorStage {
         else {
             panic!("selecting a leader from an empty validator set")
         };
+        // the leader cannot produce three points in a row, so we have an undefined leader,
+        // rather than an intentional leaderless support round - all represented by `None`
         if !current_peers.contains_key(leader) {
             return None;
         };
@@ -39,11 +41,11 @@ impl AnchorStage {
             // 1 is an anchor candidate (surprisingly, nothing special about this point)
             0 | 1 => None,
             2 => Some(AnchorStage::Proof {
-                leader: leader.clone(),
+                leader: *leader,
                 is_used: AtomicBool::new(false),
             }),
             3 => Some(AnchorStage::Trigger {
-                leader: leader.clone(),
+                leader: *leader,
                 is_used: AtomicBool::new(false),
             }),
             _ => unreachable!(),

consensus/src/dag/dag.rs

@@ -12,7 +12,7 @@ use crate::dag::anchor_stage::AnchorStage;
 use crate::dag::DagRound;
 use crate::engine::MempoolConfig;
 use crate::intercom::PeerSchedule;
-use crate::models::{Point, Round, Ugly, ValidPoint};
+use crate::models::{LinkField, Point, Round, Ugly, ValidPoint};
 
 #[derive(Clone)]
 pub struct Dag {
@@ -30,10 +30,10 @@ impl Dag {
     pub fn init(&self, dag_round: DagRound) {
         let mut rounds = self.rounds.lock();
         assert!(rounds.is_empty(), "DAG already initialized");
-        rounds.insert(dag_round.round().clone(), dag_round);
+        rounds.insert(dag_round.round(), dag_round);
     }
 
-    pub fn top(&self, round: &Round, peer_schedule: &PeerSchedule) -> DagRound {
+    pub fn top(&self, round: Round, peer_schedule: &PeerSchedule) -> DagRound {
         let mut rounds = self.rounds.lock();
         let mut top = match rounds.last_key_value() {
             None => unreachable!("DAG cannot be empty if properly initialized?"),
@@ -85,8 +85,8 @@ impl Dag {
         }
     }
 
-    async fn latest_trigger(next_round: &DagRound) -> Option<ValidPoint> {
-        let mut next_dag_round = next_round.clone();
+    async fn latest_trigger(next_dag_round: &DagRound) -> Option<ValidPoint> {
+        let mut next_dag_round = next_dag_round.clone();
         let mut latest_trigger = None;
         while let Some(current_dag_round) = next_dag_round.prev().get() {
             if let Some(AnchorStage::Trigger {
@@ -125,72 +125,71 @@ impl Dag {
     ) -> Vec<(ValidPoint, DagRound)> {
         assert_eq!(
             last_trigger.point.prev_id(),
-            Some(last_trigger.point.anchor_proof_id()),
+            Some(last_trigger.point.anchor_id(LinkField::Proof)),
             "invalid anchor proof link, trigger point must have been invalidated"
         );
         let mut anchor_stack = Vec::new();
         let Some(mut proof) = future_round.vertex_by_proof(last_trigger).await else {
             panic!("anchor proof round not in DAG")
         };
         loop {
-            let Some(proof_round) = future_round.scan(&proof.point.body.location.round) else {
+            let Some(proof_round) = future_round.scan(proof.point.body.location.round) else {
                 panic!("anchor proof round not in DAG while a point from it was received")
             };
-            if proof_round.round() == &MempoolConfig::GENESIS_ROUND {
+            if proof_round.round() == MempoolConfig::GENESIS_ROUND {
                 break;
             }
-            match proof_round.anchor_stage() {
-                Some(AnchorStage::Proof {
-                    ref leader,
-                    ref is_used,
-                }) => {
-                    assert_eq!(
-                        proof.point.body.location.round,
-                        *proof_round.round(),
-                        "anchor proof round does not match"
-                    );
-                    assert_eq!(
-                        proof.point.body.location.author, leader,
-                        "anchor proof author does not match prescribed by round"
-                    );
-                    let Some(anchor_round) = proof_round.prev().get() else {
-                        break;
-                    };
-                    if is_used.load(Ordering::Relaxed) {
-                        break;
-                    };
-                    let mut proofs = FuturesUnordered::new();
-                    proof_round.view(leader, |loc| {
-                        for (_, version) in loc.versions() {
-                            proofs.push(version.clone())
-                        }
-                    });
-                    let mut anchor = None;
-                    'v: while let Some((proof, _)) = proofs.next().await {
-                        if let Some(valid) = proof.into_valid() {
-                            let Some(valid) = proof_round.vertex_by_proof(&valid).await else {
-                                panic!("anchor proof is not linked to anchor, validation broken")
-                            };
-                            _ = anchor.insert(valid);
-                            is_used.store(true, Ordering::Relaxed);
-                            break 'v;
-                        }
-                    }
-                    let anchor = anchor
-                        .expect("any anchor proof points to anchor point, validation is broken");
-                    anchor_stack.push((anchor.clone(), anchor_round.clone()));
-
-                    let Some(next_proof) = proof_round
-                        .valid_point(&anchor.point.anchor_proof_id())
-                        .await
-                    else {
-                        break;
+            let Some(AnchorStage::Proof {
+                ref leader,
+                ref is_used,
+            }) = proof_round.anchor_stage()
+            else {
+                panic!("anchor proof round is not expected, validation is broken")
+            };
+            assert_eq!(
+                proof.point.body.location.round,
+                proof_round.round(),
+                "anchor proof round does not match"
+            );
+            assert_eq!(
+                proof.point.body.location.author, leader,
+                "anchor proof author does not match prescribed by round"
+            );
+            let Some(anchor_round) = proof_round.prev().get() else {
+                break;
+            };
+            if is_used.load(Ordering::Relaxed) {
+                break;
+            };
+            let mut proofs = FuturesUnordered::new();
+            proof_round.view(leader, |loc| {
+                for (_, version) in loc.versions() {
+                    proofs.push(version.clone())
+                }
+            });
+            let mut anchor = None;
+            'v: while let Some((proof, _)) = proofs.next().await {
+                if let Some(valid) = proof.into_valid() {
+                    let Some(valid) = proof_round.vertex_by_proof(&valid).await else {
+                        panic!("anchor proof is not linked to anchor, validation broken")
                     };
-                    proof = next_proof;
-                    future_round = anchor_round;
+                    _ = anchor.insert(valid);
+                    is_used.store(true, Ordering::Relaxed);
+                    break 'v;
                 }
-                _ => panic!("anchor proof round is not expected, validation is broken"),
             }
+            let anchor =
+                anchor.expect("any anchor proof points to anchor point, validation is broken");
+            anchor_stack.push((anchor.clone(), anchor_round.clone()));
+
+            let Some(next_proof) = proof_round
+                .valid_point(&anchor.point.anchor_id(LinkField::Proof))
+                .await
+            else {
+                break;
+            };
+            proof = next_proof;
+            future_round = anchor_round;
         }
         anchor_stack
     }
@@ -210,7 +209,7 @@ impl Dag {
         anchor_round: &DagRound, // r+1
     ) -> Vec<Arc<Point>> {
         assert_eq!(
-            *anchor_round.round(),
+            anchor_round.round(),
             anchor.body.location.round,
             "passed anchor round does not match anchor point's round"
         );

consensus/src/dag/dag_location.rs

@@ -109,7 +109,7 @@ impl InclusionState {
             None => assert!(false, "Coding error: own point is not trusted"),
             Some(valid) => {
                 _ = signed.set(Ok(Signed {
-                    at: valid.point.body.location.round.clone(),
+                    at: valid.point.body.location.round,
                     with: valid.point.signature.clone(),
                 }))
             }
@@ -132,9 +132,9 @@ impl InclusionState {
     pub fn signed(&self) -> Option<&'_ Result<Signed, ()>> {
         self.0.get()?.signed.get()
     }
-    pub fn signed_point(&self, at: &Round) -> Option<&'_ ValidPoint> {
+    pub fn signed_point(&self, at: Round) -> Option<&'_ ValidPoint> {
         let signable = self.0.get()?;
-        if &signable.signed.get()?.as_ref().ok()?.at == at {
+        if signable.signed.get()?.as_ref().ok()?.at == at {
             signable.first_completed.valid()
         } else {
             None
@@ -168,7 +168,7 @@ pub struct Signed {
 impl Signable {
     pub fn sign(
         &self,
-        at: &Round,
+        at: Round,
         key_pair: Option<&KeyPair>, // same round for own point and next round for other's
         time_range: RangeInclusive<UnixTime>,
     ) -> bool {
@@ -178,7 +178,7 @@ impl Signable {
                 _ = self.signed.get_or_init(|| {
                     this_call_signed = true;
                     Ok(Signed {
-                        at: at.clone(),
+                        at,
                         with: Signature::new(key_pair, &valid.point.digest),
                     })
                 });