From f69fc1d2f3bca0c5ae76dd5dae7d088f55323636 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 14:40:58 +0000 Subject: [PATCH 01/10] core: fix predicate for dropping grandpa round messages --- core/finality-grandpa/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/core/finality-grandpa/src/lib.rs b/core/finality-grandpa/src/lib.rs index fd2828b9bbd7d..708caf58f96b8 100644 --- a/core/finality-grandpa/src/lib.rs +++ b/core/finality-grandpa/src/lib.rs @@ -285,7 +285,7 @@ impl, H: ExHashT fn drop_messages(&self, round: u64, set_id: u64) { let topic = message_topic::(round, set_id); - self.service.consensus_gossip().write().collect_garbage(|t| t == &topic); + self.service.consensus_gossip().write().collect_garbage(|t| t != &topic); } fn commit_messages(&self, set_id: u64) -> Self::In { From af0bf6cadf8dd727bcbb13d351ad990b9738f660 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 14:58:57 +0000 Subject: [PATCH 02/10] core: grandpa: drop commits topic on authority set change --- core/finality-grandpa/src/communication.rs | 32 +++++++++++++++++----- core/finality-grandpa/src/lib.rs | 12 ++++++-- 2 files changed, 35 insertions(+), 9 deletions(-) diff --git a/core/finality-grandpa/src/communication.rs b/core/finality-grandpa/src/communication.rs index fe6e2a0477a6f..6282708169af3 100644 --- a/core/finality-grandpa/src/communication.rs +++ b/core/finality-grandpa/src/communication.rs @@ -18,6 +18,8 @@ //! that sign or re-shape. use std::collections::HashMap; +use std::sync::Arc; + use grandpa::VoterSet; use futures::prelude::*; use futures::sync::mpsc; @@ -27,8 +29,6 @@ use runtime_primitives::traits::Block as BlockT; use tokio::timer::Interval; use {Error, Network, Message, SignedMessage, Commit, CompactCommit}; -use std::sync::Arc; - fn localized_payload(round: u64, set_id: u64, message: &E) -> Vec { (message, round, set_id).encode() } @@ -42,6 +42,8 @@ enum Broadcast { Announcement(u64, u64, Block::Hash), // set_id, round being dropped. DropRound(u64, u64), + // set_id being dropped. + DropSet(u64), } impl Broadcast { @@ -51,6 +53,7 @@ impl Broadcast { Broadcast::Message(s, _, _) => s, Broadcast::Announcement(s, _, _) => s, Broadcast::DropRound(s, _) => s, + Broadcast::DropSet(s) => s, } } } @@ -179,7 +182,12 @@ impl> Future for BroadcastWorker { Broadcast::DropRound(set_id, round) => { // stop making announcements for any dead rounds. self.announcements.retain(|_, &mut r| r > round); - self.network.drop_messages(round, set_id); + self.network.drop_round_messages(round, set_id); + } + Broadcast::DropSet(set_id) => { + // stop making announcements for any dead rounds. + self.announcements.retain(|_, &mut s| s > set_id); + self.network.drop_set_messages(set_id); } } } @@ -199,10 +207,14 @@ impl> Network for BroadcastHandle { let _ = self.relay.unbounded_send(Broadcast::Message(set_id, round, message)); } - fn drop_messages(&self, round: u64, set_id: u64) { + fn drop_round_messages(&self, round: u64, set_id: u64) { let _ = self.relay.unbounded_send(Broadcast::DropRound(set_id, round)); } + fn drop_set_messages(&self, set_id: u64) { + let _ = self.relay.unbounded_send(Broadcast::DropSet(set_id)); + } + fn commit_messages(&self, set_id: u64) -> Self::In { self.network.commit_messages(set_id) } @@ -322,7 +334,7 @@ impl> Sink for OutgoingMessages impl> Drop for OutgoingMessages { fn drop(&mut self) { - self.network.drop_messages(self.round, self.set_id); + self.network.drop_round_messages(self.round, self.set_id); } } @@ -429,14 +441,14 @@ pub(crate) fn checked_commit_stream( } /// An output sink for commit messages. -pub(crate) struct CommitsOut { +pub(crate) struct CommitsOut> { network: N, set_id: u64, _marker: ::std::marker::PhantomData, is_voter: bool, } -impl CommitsOut { +impl> CommitsOut { /// Create a new commit output stream. pub(crate) fn new(network: N, set_id: u64, is_voter: bool) -> Self { CommitsOut { @@ -477,3 +489,9 @@ impl> Sink for CommitsOut { fn close(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) } fn poll_complete(&mut self) -> Poll<(), Error> { Ok(Async::Ready(())) } } + +impl> Drop for CommitsOut { + fn drop(&mut self) { + self.network.drop_set_messages(self.set_id); + } +} diff --git a/core/finality-grandpa/src/lib.rs b/core/finality-grandpa/src/lib.rs index 708caf58f96b8..186e28f9b15d1 100644 --- a/core/finality-grandpa/src/lib.rs +++ b/core/finality-grandpa/src/lib.rs @@ -231,7 +231,10 @@ pub trait Network: Clone { fn send_message(&self, round: u64, set_id: u64, message: Vec); /// Clean up messages for a round. - fn drop_messages(&self, round: u64, set_id: u64); + fn drop_round_messages(&self, round: u64, set_id: u64); + + /// Clean up messages for a given authority set id (e.g. commit messages). + fn drop_set_messages(&self, set_id: u64); /// Get a stream of commit messages for a specific set-id. This stream /// should never logically conclude. @@ -283,11 +286,16 @@ impl, H: ExHashT self.service.gossip_consensus_message(topic, message, false); } - fn drop_messages(&self, round: u64, set_id: u64) { + fn drop_round_messages(&self, round: u64, set_id: u64) { let topic = message_topic::(round, set_id); self.service.consensus_gossip().write().collect_garbage(|t| t != &topic); } + fn drop_set_messages(&self, set_id: u64) { + let topic = commit_topic::(set_id); + self.service.consensus_gossip().write().collect_garbage(|t| t != &topic); + } + fn commit_messages(&self, set_id: u64) -> Self::In { self.service.consensus_gossip().write().messages_for(commit_topic::(set_id)) } From c0b6a055572ec719d976e3a364156bc022fc0d8f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 15:00:58 +0000 Subject: [PATCH 03/10] core: gossip: only drop known messages based on expiration time --- core/network/src/consensus_gossip.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/network/src/consensus_gossip.rs b/core/network/src/consensus_gossip.rs index 186919e509fcb..171ae89a51fcc 100644 --- a/core/network/src/consensus_gossip.rs +++ b/core/network/src/consensus_gossip.rs @@ -188,11 +188,11 @@ impl ConsensusGossip { known_messages.retain(|(topic, message_hash)| { message_times.get(&(*topic, *message_hash)) - .map(|instant| *instant + (2 * MESSAGE_LIFETIME) >= now && predicate(topic)) + .map(|instant| *instant + (2 * MESSAGE_LIFETIME) >= now) .unwrap_or(false) }); - trace!(target:"gossip", "Cleaned up {} stale messages, {} left ({} known)", + trace!(target: "gossip", "Cleaned up {} stale messages, {} left ({} known)", before - self.messages.len(), self.messages.len(), known_messages.len(), From 9b3319dc2301ec1a5e819a8bc5382ee75bcaf177 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 15:03:45 +0000 Subject: [PATCH 04/10] core: grandpa: don't broadcast commit messages --- core/finality-grandpa/src/lib.rs | 2 +- core/finality-grandpa/src/tests.rs | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/core/finality-grandpa/src/lib.rs b/core/finality-grandpa/src/lib.rs index 186e28f9b15d1..2c12b398e1997 100644 --- a/core/finality-grandpa/src/lib.rs +++ b/core/finality-grandpa/src/lib.rs @@ -302,7 +302,7 @@ impl, H: ExHashT fn send_commit(&self, _round: u64, set_id: u64, message: Vec) { let topic = commit_topic::(set_id); - self.service.gossip_consensus_message(topic, message, true); + self.service.gossip_consensus_message(topic, message, false); } fn announce(&self, round: u64, _set_id: u64, block: B::Hash) { diff --git a/core/finality-grandpa/src/tests.rs b/core/finality-grandpa/src/tests.rs index 1e2b25ad4b3c5..f69e5891be1da 100644 --- a/core/finality-grandpa/src/tests.rs +++ b/core/finality-grandpa/src/tests.rs @@ -226,7 +226,7 @@ impl Network for MessageRouting { fn send_commit(&self, _round: u64, set_id: u64, message: Vec) { let mut inner = self.inner.lock(); - inner.peer(self.peer_id).gossip_message(make_commit_topic(set_id), message, true); + inner.peer(self.peer_id).gossip_message(make_commit_topic(set_id), message, false); inner.route_until_complete(); } From b4c835698736327cb9f9ea1774ade1039ca97d46 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 15:15:10 +0000 Subject: [PATCH 05/10] core: gossip: don't assume topics are header hashes --- core/network/src/consensus_gossip.rs | 17 +---------------- 1 file changed, 1 insertion(+), 16 deletions(-) diff --git a/core/network/src/consensus_gossip.rs b/core/network/src/consensus_gossip.rs index 171ae89a51fcc..99243290140cb 100644 --- a/core/network/src/consensus_gossip.rs +++ b/core/network/src/consensus_gossip.rs @@ -22,7 +22,7 @@ use futures::sync::mpsc; use std::time::{Instant, Duration}; use rand::{self, seq::SliceRandom}; use network_libp2p::NodeIndex; -use runtime_primitives::traits::{Block as BlockT, Header as HeaderT, Hash, HashFor}; +use runtime_primitives::traits::{Block as BlockT, Hash, HashFor}; use runtime_primitives::generic::BlockId; pub use message::generic::{Message, ConsensusMessage}; use protocol::Context; @@ -235,21 +235,6 @@ impl ConsensusGossip { return None; } - match (protocol.client().info(), protocol.client().header(&BlockId::Hash(topic))) { - (_, Err(e)) | (Err(e), _) => { - debug!(target:"gossip", "Error reading blockchain: {:?}", e); - return None; - }, - (Ok(info), Ok(Some(header))) => { - if header.number() < &info.chain.best_number { - trace!(target:"gossip", "Ignored ancient message from {}, hash={}", who, topic); - return None; - } - }, - (Ok(_), Ok(None)) => {}, - } - - if let Some(ref mut peer) = self.peers.get_mut(&who) { use std::collections::hash_map::Entry; peer.known_messages.insert((topic, message_hash)); From 1188367c1557519a9d52fcd514f4695555b4c792 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 15:17:24 +0000 Subject: [PATCH 06/10] core: gossip: expire messages more agressively --- core/network/src/consensus_gossip.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/core/network/src/consensus_gossip.rs b/core/network/src/consensus_gossip.rs index 99243290140cb..8d8a4879ac1cd 100644 --- a/core/network/src/consensus_gossip.rs +++ b/core/network/src/consensus_gossip.rs @@ -29,7 +29,7 @@ use protocol::Context; use config::Roles; // FIXME: Add additional spam/DoS attack protection: https://github.com/paritytech/substrate/issues/1115 -const MESSAGE_LIFETIME: Duration = Duration::from_secs(600); +const MESSAGE_LIFETIME: Duration = Duration::from_secs(120); struct PeerConsensus { known_messages: HashSet, @@ -188,7 +188,7 @@ impl ConsensusGossip { known_messages.retain(|(topic, message_hash)| { message_times.get(&(*topic, *message_hash)) - .map(|instant| *instant + (2 * MESSAGE_LIFETIME) >= now) + .map(|instant| *instant + (5 * MESSAGE_LIFETIME) >= now) .unwrap_or(false) }); From 6fcae0d52a39e96a294e44bddddf154fcb617efe Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 15:31:41 +0000 Subject: [PATCH 07/10] core: grandpa: fix test environment --- core/finality-grandpa/src/tests.rs | 23 ++++++++++++++++------- 1 file changed, 16 insertions(+), 7 deletions(-) diff --git a/core/finality-grandpa/src/tests.rs b/core/finality-grandpa/src/tests.rs index f69e5891be1da..4895a54e866ab 100644 --- a/core/finality-grandpa/src/tests.rs +++ b/core/finality-grandpa/src/tests.rs @@ -145,6 +145,15 @@ impl MessageRouting { peer_id, } } + + fn drop_messages(&self, topic: Hash) { + let inner = self.inner.lock(); + let peer = inner.peer(self.peer_id); + let mut gossip = peer.consensus_gossip().write(); + peer.with_spec(move |_, _| { + gossip.collect_garbage(|t| t != &topic) + }); + } } fn make_topic(round: u64, set_id: u64) -> Hash { @@ -199,14 +208,14 @@ impl Network for MessageRouting { inner.route_until_complete(); } - fn drop_messages(&self, round: u64, set_id: u64) { + fn drop_round_messages(&self, round: u64, set_id: u64) { let topic = make_topic(round, set_id); - let inner = self.inner.lock(); - let peer = inner.peer(self.peer_id); - let mut gossip = peer.consensus_gossip().write(); - peer.with_spec(move |_, _| { - gossip.collect_garbage(|t| t == &topic) - }); + self.drop_messages(topic); + } + + fn drop_set_messages(&self, set_id: u64) { + let topic = make_commit_topic(set_id); + self.drop_messages(topic); } fn commit_messages(&self, set_id: u64) -> Self::In { From fb2c97d60a70f3e9bec2031605e406be05f40a2c Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 15:44:30 +0000 Subject: [PATCH 08/10] core: gossip: fix tests --- core/network/src/consensus_gossip.rs | 23 ++++++++++++++--------- 1 file changed, 14 insertions(+), 9 deletions(-) diff --git a/core/network/src/consensus_gossip.rs b/core/network/src/consensus_gossip.rs index 8d8a4879ac1cd..781ca492107ae 100644 --- a/core/network/src/consensus_gossip.rs +++ b/core/network/src/consensus_gossip.rs @@ -314,13 +314,15 @@ mod tests { macro_rules! push_msg { ($topic:expr, $hash: expr, $now: expr, $m:expr) => { - consensus.messages.push(MessageEntry { - topic: $topic, - message_hash: $hash, - message: $m, - broadcast: false, - }); - consensus.message_times.insert(($topic, $hash), $now); + if consensus.known_messages.insert(($topic, $hash)) { + consensus.messages.push(MessageEntry { + topic: $topic, + message_hash: $hash, + message: $m, + broadcast: false, + }); + consensus.message_times.insert(($topic, $hash), $now); + } } } @@ -342,18 +344,21 @@ mod tests { // topic that was used in one message. consensus.collect_garbage(|topic| topic != &prev_hash); assert_eq!(consensus.messages.len(), 1); - assert_eq!(consensus.known_messages.len(), 1); + // known messages are only pruned based on expiration time + assert_eq!(consensus.known_messages.len(), 2); assert!(consensus.known_messages.contains(&(best_hash, m2_hash))); // make timestamp expired, but the message is still kept as known consensus.messages.clear(); + consensus.known_messages.clear(); push_msg!(best_hash, m2_hash, now - MESSAGE_LIFETIME, m2.clone()); consensus.collect_garbage(|_topic| true); assert!(consensus.messages.is_empty()); assert_eq!(consensus.known_messages.len(), 1); // make timestamp expired past the known message lifetime - push_msg!(best_hash, m2_hash, now - (2 * MESSAGE_LIFETIME), m2); + consensus.known_messages.clear(); + push_msg!(best_hash, m2_hash, now - (5 * MESSAGE_LIFETIME), m2); consensus.collect_garbage(|_topic| true); assert!(consensus.messages.is_empty()); assert!(consensus.known_messages.is_empty()); From 011d14cbeb0a6a36ba05ccc0386fe2a1b391e720 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 16:25:44 +0000 Subject: [PATCH 09/10] core: gossip: track dead topics (and ignore messages) --- Cargo.lock | 1 + core/finality-grandpa/src/lib.rs | 4 ++-- core/finality-grandpa/src/tests.rs | 2 +- core/network/Cargo.toml | 1 + core/network/src/consensus_gossip.rs | 28 +++++++++++++++++++++++----- core/network/src/lib.rs | 1 + 6 files changed, 29 insertions(+), 8 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 6771530327c8f..7caa98e8dc7d7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -3708,6 +3708,7 @@ dependencies = [ "futures 0.1.25 (registry+https://github.com/rust-lang/crates.io-index)", "linked-hash-map 0.5.1 (registry+https://github.com/rust-lang/crates.io-index)", "log 0.4.6 (registry+https://github.com/rust-lang/crates.io-index)", + "lru-cache 0.1.1 (registry+https://github.com/rust-lang/crates.io-index)", "parity-codec 3.0.0 (registry+https://github.com/rust-lang/crates.io-index)", "parity-codec-derive 3.0.0 (registry+https://github.com/rust-lang/crates.io-index)", "parking_lot 0.7.1 (registry+https://github.com/rust-lang/crates.io-index)", diff --git a/core/finality-grandpa/src/lib.rs b/core/finality-grandpa/src/lib.rs index 2c12b398e1997..08e95a2955de0 100644 --- a/core/finality-grandpa/src/lib.rs +++ b/core/finality-grandpa/src/lib.rs @@ -288,12 +288,12 @@ impl, H: ExHashT fn drop_round_messages(&self, round: u64, set_id: u64) { let topic = message_topic::(round, set_id); - self.service.consensus_gossip().write().collect_garbage(|t| t != &topic); + self.service.consensus_gossip().write().collect_garbage_for_topic(topic); } fn drop_set_messages(&self, set_id: u64) { let topic = commit_topic::(set_id); - self.service.consensus_gossip().write().collect_garbage(|t| t != &topic); + self.service.consensus_gossip().write().collect_garbage_for_topic(topic); } fn commit_messages(&self, set_id: u64) -> Self::In { diff --git a/core/finality-grandpa/src/tests.rs b/core/finality-grandpa/src/tests.rs index 4895a54e866ab..a44bd3f32e53c 100644 --- a/core/finality-grandpa/src/tests.rs +++ b/core/finality-grandpa/src/tests.rs @@ -151,7 +151,7 @@ impl MessageRouting { let peer = inner.peer(self.peer_id); let mut gossip = peer.consensus_gossip().write(); peer.with_spec(move |_, _| { - gossip.collect_garbage(|t| t != &topic) + gossip.collect_garbage_for_topic(topic); }); } } diff --git a/core/network/Cargo.toml b/core/network/Cargo.toml index 87052b7e4111b..1a3a230778fed 100644 --- a/core/network/Cargo.toml +++ b/core/network/Cargo.toml @@ -14,6 +14,7 @@ error-chain = "0.12" bitflags = "1.0" futures = "0.1.17" linked-hash-map = "0.5" +lru-cache = "0.1.1" rustc-hex = "2.0" rand = "0.6" substrate-primitives = { path = "../../core/primitives" } diff --git a/core/network/src/consensus_gossip.rs b/core/network/src/consensus_gossip.rs index 781ca492107ae..99f9d4566a38d 100644 --- a/core/network/src/consensus_gossip.rs +++ b/core/network/src/consensus_gossip.rs @@ -18,9 +18,10 @@ //! Handles chain-specific and standard BFT messages. use std::collections::{HashMap, HashSet}; -use futures::sync::mpsc; use std::time::{Instant, Duration}; +use futures::sync::mpsc; use rand::{self, seq::SliceRandom}; +use lru_cache::LruCache; use network_libp2p::NodeIndex; use runtime_primitives::traits::{Block as BlockT, Hash, HashFor}; use runtime_primitives::generic::BlockId; @@ -30,6 +31,7 @@ use config::Roles; // FIXME: Add additional spam/DoS attack protection: https://github.com/paritytech/substrate/issues/1115 const MESSAGE_LIFETIME: Duration = Duration::from_secs(120); +const DEAD_TOPICS_CACHE_SIZE: usize = 4096; struct PeerConsensus { known_messages: HashSet, @@ -49,6 +51,7 @@ pub struct ConsensusGossip { live_message_sinks: HashMap>>, messages: Vec>, known_messages: HashSet<(B::Hash, B::Hash)>, + known_dead_topics: LruCache, message_times: HashMap<(B::Hash, B::Hash), Instant>, session_start: Option, } @@ -61,6 +64,7 @@ impl ConsensusGossip { live_message_sinks: HashMap::new(), messages: Default::default(), known_messages: Default::default(), + known_dead_topics: LruCache::new(DEAD_TOPICS_CACHE_SIZE), message_times: Default::default(), session_start: None } @@ -150,7 +154,9 @@ impl ConsensusGossip { fn register_message(&mut self, message_hash: B::Hash, topic: B::Hash, broadcast: bool, get_message: F) where F: Fn() -> ConsensusMessage { - if self.known_messages.insert((topic, message_hash)) { + if !self.known_dead_topics.contains_key(&topic) && + self.known_messages.insert((topic, message_hash)) + { self.messages.push(MessageEntry { topic, message_hash, @@ -167,6 +173,11 @@ impl ConsensusGossip { self.peers.remove(&who); } + pub fn collect_garbage_for_topic(&mut self, topic: B::Hash) { + self.known_dead_topics.insert(topic, ()); + self.collect_garbage(|_| true); + } + /// Prune old or no longer relevant consensus messages. Provide a predicate /// for pruning, which returns `false` when the items with a given topic should be pruned. pub fn collect_garbage bool>(&mut self, predicate: P) { @@ -177,13 +188,15 @@ impl ConsensusGossip { let message_times = &mut self.message_times; let known_messages = &mut self.known_messages; + let known_dead_topics = &mut self.known_dead_topics; let before = self.messages.len(); let now = Instant::now(); self.messages.retain(|entry| { - message_times.get(&(entry.topic, entry.message_hash)) - .map(|instant| *instant + MESSAGE_LIFETIME >= now && predicate(&entry.topic)) - .unwrap_or(false) + !known_dead_topics.contains_key(&entry.topic) && + message_times.get(&(entry.topic, entry.message_hash)) + .map(|instant| *instant + MESSAGE_LIFETIME >= now && predicate(&entry.topic)) + .unwrap_or(false) }); known_messages.retain(|(topic, message_hash)| { @@ -230,6 +243,11 @@ impl ConsensusGossip { ) -> Option<(B::Hash, ConsensusMessage)> { let message_hash = HashFor::::hash(&message[..]); + if self.known_dead_topics.contains_key(&topic) { + trace!(target:"gossip", "Ignored message from {} in dead topic {}", who, topic); + return None; + } + if self.known_messages.contains(&(topic, message_hash)) { trace!(target:"gossip", "Ignored already known message from {} in {}", who, topic); return None; diff --git a/core/network/src/lib.rs b/core/network/src/lib.rs index 95b6d68113d4a..d2868f1a5c248 100644 --- a/core/network/src/lib.rs +++ b/core/network/src/lib.rs @@ -21,6 +21,7 @@ //! Allows attachment of an optional subprotocol for chain-specific requests. extern crate linked_hash_map; +extern crate lru_cache; extern crate parking_lot; extern crate substrate_primitives as primitives; extern crate substrate_client as client; From 480f1276abf0519f9e6d01db2246af545c8bb149 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Silva?= Date: Fri, 1 Feb 2019 17:09:16 +0000 Subject: [PATCH 10/10] core: gossip: test dead topic pruning --- core/network/src/consensus_gossip.rs | 62 ++++++++++++++++++++-------- 1 file changed, 44 insertions(+), 18 deletions(-) diff --git a/core/network/src/consensus_gossip.rs b/core/network/src/consensus_gossip.rs index 99f9d4566a38d..a5a67b9507eb5 100644 --- a/core/network/src/consensus_gossip.rs +++ b/core/network/src/consensus_gossip.rs @@ -319,6 +319,20 @@ mod tests { type Block = RawBlock>; + macro_rules! push_msg { + ($consensus:expr, $topic:expr, $hash: expr, $now: expr, $m:expr) => { + if $consensus.known_messages.insert(($topic, $hash)) { + $consensus.messages.push(MessageEntry { + topic: $topic, + message_hash: $hash, + message: $m, + broadcast: false, + }); + $consensus.message_times.insert(($topic, $hash), $now); + } + } + } + #[test] fn collects_garbage() { let prev_hash = H256::random(); @@ -330,22 +344,8 @@ mod tests { let m1 = vec![1, 2, 3]; let m2 = vec![4, 5, 6]; - macro_rules! push_msg { - ($topic:expr, $hash: expr, $now: expr, $m:expr) => { - if consensus.known_messages.insert(($topic, $hash)) { - consensus.messages.push(MessageEntry { - topic: $topic, - message_hash: $hash, - message: $m, - broadcast: false, - }); - consensus.message_times.insert(($topic, $hash), $now); - } - } - } - - push_msg!(prev_hash, m1_hash, now, m1); - push_msg!(best_hash, m2_hash, now, m2.clone()); + push_msg!(consensus, prev_hash, m1_hash, now, m1); + push_msg!(consensus, best_hash, m2_hash, now, m2.clone()); consensus.known_messages.insert((prev_hash, m1_hash)); consensus.known_messages.insert((best_hash, m2_hash)); @@ -369,19 +369,45 @@ mod tests { // make timestamp expired, but the message is still kept as known consensus.messages.clear(); consensus.known_messages.clear(); - push_msg!(best_hash, m2_hash, now - MESSAGE_LIFETIME, m2.clone()); + push_msg!(consensus, best_hash, m2_hash, now - MESSAGE_LIFETIME, m2.clone()); consensus.collect_garbage(|_topic| true); assert!(consensus.messages.is_empty()); assert_eq!(consensus.known_messages.len(), 1); // make timestamp expired past the known message lifetime consensus.known_messages.clear(); - push_msg!(best_hash, m2_hash, now - (5 * MESSAGE_LIFETIME), m2); + push_msg!(consensus, best_hash, m2_hash, now - (5 * MESSAGE_LIFETIME), m2); consensus.collect_garbage(|_topic| true); assert!(consensus.messages.is_empty()); assert!(consensus.known_messages.is_empty()); } + #[test] + fn collects_garbage_for_topic() { + let topic = H256::random(); + let dead_topic = H256::random(); + let message = Vec::new(); + let now = Instant::now(); + let mut consensus = ConsensusGossip::::new(); + + let message_hash = H256::random(); + push_msg!(consensus, topic, message_hash, now, message.clone()); + push_msg!(consensus, dead_topic, message_hash, now, message.clone()); + assert_eq!(consensus.messages.len(), 2); + + consensus.collect_garbage_for_topic(topic); + + // removes all messages for the topic and marks the topic as dead + assert_eq!(consensus.messages.len(), 1); + assert_eq!(consensus.known_messages.len(), 2); + assert_eq!(consensus.known_dead_topics.len(), 1); + + // new messages for dead topics are ignored + consensus.register_message(HashFor::::hash(&message), topic, false, || message.clone()); + assert_eq!(consensus.messages.len(), 1); + assert_eq!(consensus.known_messages.len(), 2); + } + #[test] fn message_stream_include_those_sent_before_asking_for_stream() { use futures::Stream;