Improve statement-store gossiping performance

prdoc/pr_9912.prdoc

@@ -0,0 +1,14 @@
+title: '[WIP] Fix statement-store gossiping'
+doc:
+- audience: Node Dev
+  description: |-
+    Fixes gossiping and scalability issues in the statement-store networking:
+    reduces traffic by propagating only recent statements, skips duplicate processing,
+    and splits large batches to stay under MAX_STATEMENT_NOTIFICATION_SIZE.
+crates:
+- name: sc-network-statement
+  bump: minor
+- name: sc-statement-store
+  bump: minor
+- name: sp-statement-store
+  bump: minor

substrate/client/network/statement/src/config.rs

@@ -24,10 +24,10 @@ use std::time;
 pub(crate) const PROPAGATE_TIMEOUT: time::Duration = time::Duration::from_millis(1000);
 
 /// Maximum number of known statement hashes to keep for a peer.
-pub(crate) const MAX_KNOWN_STATEMENTS: usize = 4 * 1024 * 1024;
+pub(crate) const MAX_KNOWN_STATEMENTS: usize = 4 * 1024 * 1024; // * 32 bytes for hash = 128 MB per peer
 
 /// Maximum allowed size for a statement notification.
-pub(crate) const MAX_STATEMENT_SIZE: u64 = 256 * 1024;
+pub(crate) const MAX_STATEMENT_NOTIFICATION_SIZE: u64 = 256 * 1024;
 
 /// Maximum number of statement validation request we keep at any moment.
 pub(crate) const MAX_PENDING_STATEMENTS: usize = 2 * 1024 * 1024;

substrate/client/network/statement/src/lib.rs

@@ -30,7 +30,9 @@ use crate::config::*;
 
 use codec::{Decode, Encode};
 use futures::{channel::oneshot, prelude::*, stream::FuturesUnordered, FutureExt};
-use prometheus_endpoint::{register, Counter, PrometheusError, Registry, U64};
+use prometheus_endpoint::{
+	prometheus, register, Counter, Gauge, Histogram, HistogramOpts, PrometheusError, Registry, U64,
+};
 use sc_network::{
 	config::{NonReservedPeerMode, SetConfig},
 	error, multiaddr,
@@ -88,6 +90,10 @@ const LOG_TARGET: &str = "statement-gossip";
 
 struct Metrics {
 	propagated_statements: Counter<U64>,
+	known_statements_received: Counter<U64>,
+	skipped_oversized_statements: Counter<U64>,
+	propagated_statements_chunks: Histogram,
+	pending_statements: Gauge<U64>,
 }
 
 impl Metrics {
@@ -100,6 +106,36 @@ impl Metrics {
 				)?,
 				r,
 			)?,
+			known_statements_received: register(
+				Counter::new(
+					"substrate_sync_known_statement_received",
+					"Number of statements received via gossiping that were already in the statement store",
+				)?,
+				r,
+			)?,
+			skipped_oversized_statements: register(
+				Counter::new(
+					"substrate_sync_skipped_oversized_statements",
+					"Number of oversized statements that were skipped to be gossiped",
+				)?,
+				r,
+			)?,
+			propagated_statements_chunks: register(
+				Histogram::with_opts(
+					HistogramOpts::new(
+						"substrate_sync_propagated_statements_chunks",
+						"Distribution of chunk sizes when propagating statements",
+					).buckets(prometheus::exponential_buckets(1.0, 2.0, 14)?),
+				)?,
+				r,
+			)?,
+			pending_statements: register(
+				Gauge::new(
+					"substrate_sync_pending_statement_validations",
+					"Number of pending statement validations",
+				)?,
+				r,
+			)?,
 		})
 	}
 }
@@ -131,7 +167,7 @@ impl StatementHandlerPrototype {
 		let (config, notification_service) = Net::notification_config(
 			protocol_name.clone().into(),
 			Vec::new(),
-			MAX_STATEMENT_SIZE,
+			MAX_STATEMENT_NOTIFICATION_SIZE,
 			None,
 			SetConfig {
 				in_peers: 0,
@@ -264,6 +300,9 @@ where
 			futures::select! {
 				_ = self.propagate_timeout.next() => {
 					self.propagate_statements();
+					if let Some(ref metrics) = self.metrics {
+						metrics.pending_statements.set(self.pending_statements.len() as u64);
+					}
 				},
 				(hash, result) = self.pending_statements.select_next_some() => {
 					if let Some(peers) = self.pending_statements_peers.remove(&hash) {
@@ -388,6 +427,13 @@ where
 
 				self.network.report_peer(who, rep::ANY_STATEMENT);
 
+				if self.statement_store.has_statement(&hash) {
+					if let Some(ref metrics) = self.metrics {
+						metrics.known_statements_received.inc();
+					}
+					continue;
+				}
+
 				match self.pending_statements_peers.entry(hash) {
 					Entry::Vacant(entry) => {
 						let (completion_sender, completion_receiver) = oneshot::channel();
@@ -470,9 +516,55 @@ where
 
 			propagated_statements += to_send.len();
 
-			if !to_send.is_empty() {
-				log::trace!(target: LOG_TARGET, "Sending {} statements to {}", to_send.len(), who);
-				self.notification_service.send_sync_notification(who, to_send.encode());
+			let mut offset = 0;
+			while offset < to_send.len() {
+				// Try to send as many statements as possible in one notification
+				let chunk_size = to_send.len() - offset;
+				let mut current_end = to_send.len();
+
+				loop {
+					let chunk = &to_send[offset..current_end];
+					let encoded_size = chunk.encoded_size();
+
+					// If chunk fits, send it
+					if encoded_size <= MAX_STATEMENT_NOTIFICATION_SIZE as usize {
+						log::trace!(
+							target: LOG_TARGET,
+							"Sending {} statements ({} KB) to {}",
+							chunk.len(),
+							encoded_size / 1024,
+							who
+						);
+						self.notification_service.send_sync_notification(who, chunk.encode());
+						offset = current_end;
+						if let Some(ref metrics) = self.metrics {
+							metrics.propagated_statements_chunks.observe(chunk.len() as f64);
+						}
+						break;
+					}
+
+					// Size exceeded - split the chunk
+					let split_factor =
+						(encoded.len() / MAX_STATEMENT_NOTIFICATION_SIZE as usize) + 1;
+					let new_chunk_size = (current_end - offset) / split_factor;
+
+					// Single statement is too large
+					if new_chunk_size == 0 {
+						log::warn!(
+							target: LOG_TARGET,
+							"Statement too large ({} KB), skipping",
+							encoded.len() / 1024
+						);
+						if let Some(ref metrics) = self.metrics {
+							metrics.skipped_oversized_statements.inc();
+						}
+						offset = current_end;
+						break;
+					}
+
+					// Reduce chunk size and try again
+					current_end = offset + new_chunk_size;
+				}
 			}
 		}
 
@@ -489,7 +581,7 @@ where
 		}
 
 		log::debug!(target: LOG_TARGET, "Propagating statements");
-		if let Ok(statements) = self.statement_store.statements() {
+		if let Ok(statements) = self.statement_store.take_recent_statements() {
 			self.do_propagate_statements(&statements);
 		}
 	}

substrate/client/statement-store/src/lib.rs

@@ -155,6 +155,7 @@ impl Default for Options {
 
 #[derive(Default)]
 struct Index {
+	recent: HashSet<Hash>,
 	by_topic: HashMap<Topic, HashSet<Hash>>,
 	by_dec_key: HashMap<Option<DecryptionKey>, HashSet<Hash>>,
 	topics_and_keys: HashMap<Hash, ([Option<Topic>; MAX_TOPICS], Option<DecryptionKey>)>,
@@ -243,6 +244,7 @@ impl Index {
 		}
 		let priority = Priority(statement.priority().unwrap_or(0));
 		self.entries.insert(hash, (account, priority, statement.data_len()));
+		self.recent.insert(hash);
 		self.total_size += statement.data_len();
 		let account_info = self.accounts.entry(account).or_default();
 		account_info.data_size += statement.data_len();
@@ -324,6 +326,10 @@ impl Index {
 		purged
 	}
 
+	fn take_recent(&mut self) -> HashSet<Hash> {
+		std::mem::take(&mut self.recent)
+	}
+
 	fn make_expired(&mut self, hash: &Hash, current_time: u64) -> bool {
 		if let Some((account, priority, len)) = self.entries.remove(hash) {
 			self.total_size -= len;
@@ -347,6 +353,7 @@ impl Index {
 					}
 				}
 			}
+			let _ = self.recent.remove(hash);
 			self.expired.insert(*hash, current_time);
 			if let std::collections::hash_map::Entry::Occupied(mut account_rec) =
 				self.accounts.entry(account)
@@ -767,11 +774,27 @@ impl StatementStore for Store {
 	fn statements(&self) -> Result<Vec<(Hash, Statement)>> {
 		let index = self.index.read();
 		let mut result = Vec::with_capacity(index.entries.len());
-		for h in index.entries.keys() {
-			let encoded = self.db.get(col::STATEMENTS, h).map_err(|e| Error::Db(e.to_string()))?;
+		for hash in index.entries.keys().cloned() {
+			let encoded =
+				self.db.get(col::STATEMENTS, &hash).map_err(|e| Error::Db(e.to_string()))?;
+			if let Some(encoded) = encoded {
+				if let Ok(statement) = Statement::decode(&mut encoded.as_slice()) {
+					result.push((hash, statement));
+				}
+			}
+		}
+		Ok(result)
+	}
+
+	fn take_recent_statements(&self) -> Result<Vec<(Hash, Statement)>> {
+		let mut index = self.index.write();
+		let recent = index.take_recent();
+		let mut result = Vec::with_capacity(recent.len());
+		for hash in recent {
+			let encoded =
+				self.db.get(col::STATEMENTS, &hash).map_err(|e| Error::Db(e.to_string()))?;
 			if let Some(encoded) = encoded {
 				if let Ok(statement) = Statement::decode(&mut encoded.as_slice()) {
-					let hash = statement.hash();
 					result.push((hash, statement));
 				}
 			}
@@ -810,6 +833,10 @@ impl StatementStore for Store {
 		)
 	}
 
+	fn has_statement(&self, hash: &Hash) -> bool {
+		self.index.read().entries.contains_key(hash)
+	}
+
 	/// Return the data of all known statements which include all topics and have no `DecryptionKey`
 	/// field.
 	fn broadcasts(&self, match_all_topics: &[Topic]) -> Result<Vec<Vec<u8>>> {
@@ -1214,6 +1241,40 @@ mod tests {
 		assert_eq!(store.statement(&statement1.hash()).unwrap(), Some(statement1));
 	}
 
+	#[test]
+	fn take_recent_statements_clears_index() {
+		let (store, _temp) = test_store();
+		let statement0 = signed_statement(0);
+		let statement1 = signed_statement(1);
+		let statement2 = signed_statement(2);
+		let statement3 = signed_statement(3);
+
+		let _ = store.submit(statement0.clone(), StatementSource::Local);
+		let _ = store.submit(statement1.clone(), StatementSource::Local);
+		let _ = store.submit(statement2.clone(), StatementSource::Local);
+
+		let recent1 = store.take_recent_statements().unwrap();
+		let (recent1_hashes, recent1_statements): (Vec<_>, Vec<_>) = recent1.into_iter().unzip();
+		let expected1 = vec![statement0, statement1, statement2];
+		assert_eq!(recent1_hashes, expected1.iter().map(|s| s.hash()).collect::<Vec<_>>());
+		assert_eq!(recent1_statements, expected1);
+
+		// Recent statements are cleared.
+		let recent2 = store.take_recent_statements().unwrap();
+		assert_eq!(recent2.len(), 0);
+
+		store.submit(statement3.clone(), StatementSource::Network);
+
+		let recent3 = store.take_recent_statements().unwrap();
+		let (recent3_hashes, recent3_statements): (Vec<_>, Vec<_>) = recent3.into_iter().unzip();
+		let expected3 = vec![statement3];
+		assert_eq!(recent3_hashes, expected3.iter().map(|s| s.hash()).collect::<Vec<_>>());
+		assert_eq!(recent3_statements, expected3);
+
+		// Recent statements are cleared, but statements remain in the store.
+		assert_eq!(store.statements().unwrap().len(), 4);
+	}
+
 	#[test]
 	fn search_by_topic_and_key() {
 		let (store, _temp) = test_store();

substrate/primitives/statement-store/src/store_api.rs

@@ -66,9 +66,20 @@ pub trait StatementStore: Send + Sync {
 	/// Return all statements.
 	fn statements(&self) -> Result<Vec<(Hash, Statement)>>;
 
+	/// Return recent statements and clear the internal index.
+	///
+	/// This consumes and clears the recently received statements,
+	/// allowing new statements to be collected from this point forward.
+	fn take_recent_statements(&self) -> Result<Vec<(Hash, Statement)>>;
+
 	/// Get statement by hash.
 	fn statement(&self, hash: &Hash) -> Result<Option<Statement>>;
 
+	/// Check if statement exists in the store
+	///
+	/// Fast index check without accessing the DB.
+	fn has_statement(&self, hash: &Hash) -> bool;
+
 	/// Return the data of all known statements which include all topics and have no `DecryptionKey`
 	/// field.
 	fn broadcasts(&self, match_all_topics: &[Topic]) -> Result<Vec<Vec<u8>>>;