Implement persistent reputation database for collator protocol (#7751)

.github/zombienet-tests/zombienet_polkadot_tests.yml

@@ -252,3 +252,8 @@
   test-filter: "functional::shared_core_idle_parachain::shared_core_idle_parachain_test"
   runner-type: "default"
   use-zombienet-sdk: true
+
+- job-name: "zombienet-polkadot-collators-basic-reputation-persistence"
+  test-filter: "functional::collators_reputation_persistence"
+  runner-type: "default"
+  use-zombienet-sdk: true

cumulus/client/relay-chain-inprocess-interface/src/lib.rs

@@ -421,6 +421,7 @@ fn build_polkadot_full_node(
 		invulnerable_ah_collators: HashSet::new(),
 		collator_protocol_hold_off: None,
 		experimental_collator_protocol: false,
+		collator_reputation_persist_interval: None,
 	};
 
 	let (relay_chain_full_node, paranode_req_receiver) = match config.network.network_backend {

polkadot/cli/src/cli.rs

@@ -170,6 +170,13 @@ pub struct RunCmd {
 	/// Enable experimental collator protocol. TESTING ONLY! Don't use on production
 	#[arg(long, hide = true, default_value = "false")]
 	pub experimental_collator_protocol: bool,
+
+	/// Collator reputation persistence interval in seconds.
+	/// If not specified, defaults to 600 seconds (10 minutes).
+	/// This should be used only with experimental_collator_protocol
+	/// and only on validators.
+	#[arg(long, requires = "experimental_collator_protocol", requires = "validator")]
+	pub collator_reputation_persist_interval: Option<u64>,
 }
 
 #[allow(missing_docs)]

polkadot/cli/src/command.rs

@@ -294,6 +294,10 @@ where
 				invulnerable_ah_collators,
 				collator_protocol_hold_off,
 				experimental_collator_protocol,
+				collator_reputation_persist_interval: cli
+					.run
+					.collator_reputation_persist_interval
+					.map(std::time::Duration::from_secs),
 			},
 		)
 		.map(|full| full.task_manager)?;

polkadot/node/network/collator-protocol/Cargo.toml

@@ -23,17 +23,20 @@ sp-core = { workspace = true, default-features = true }
 sp-keystore = { workspace = true, default-features = true }
 sp-runtime = { workspace = true, default-features = true }
 
+codec.workspace = true
 fatality = { workspace = true }
 polkadot-node-network-protocol = { workspace = true, default-features = true }
 polkadot-node-primitives = { workspace = true, default-features = true }
 polkadot-node-subsystem = { workspace = true, default-features = true }
 polkadot-node-subsystem-util = { workspace = true, default-features = true }
 polkadot-primitives = { workspace = true, default-features = true }
 thiserror = { workspace = true }
+tokio.workspace = true
 tokio-util = { workspace = true }
 
 [dev-dependencies]
 assert_matches = { workspace = true }
+kvdb-memorydb = { workspace = true }
 rstest = { workspace = true }
 sc-network-types = { workspace = true, default-features = true }
 sp-tracing = { workspace = true }

polkadot/node/network/collator-protocol/src/lib.rs

@@ -23,6 +23,7 @@
 
 use std::{
 	collections::HashSet,
+	sync::Arc,
 	time::{Duration, Instant},
 };
 
@@ -31,16 +32,16 @@ use futures::{
 	FutureExt, TryFutureExt,
 };
 
-use polkadot_node_subsystem_util::reputation::ReputationAggregator;
+use polkadot_node_subsystem_util::{database::Database, reputation::ReputationAggregator};
 use sp_keystore::KeystorePtr;
 
 use polkadot_node_network_protocol::{
 	request_response::{v2 as protocol_v2, IncomingRequestReceiver},
 	PeerId, UnifiedReputationChange as Rep,
 };
-use polkadot_primitives::CollatorPair;
-
 use polkadot_node_subsystem::{errors::SubsystemError, overseer, DummySubsystem, SpawnedSubsystem};
+use polkadot_primitives::CollatorPair;
+pub use validator_side_experimental::ReputationConfig;
 
 mod collator_side;
 mod validator_side;
@@ -91,6 +92,10 @@ pub enum ProtocolSide {
 		keystore: KeystorePtr,
 		/// Prometheus metrics for validators.
 		metrics: validator_side_experimental::Metrics,
+		/// Database used for reputation house keeping.
+		db: Arc<dyn Database>,
+		/// Reputation configuration (column number).
+		reputation_config: validator_side_experimental::ReputationConfig,
 	},
 	/// Collators operate on a parachain.
 	Collator {
@@ -148,8 +153,8 @@ impl<Context> CollatorProtocolSubsystem {
 				.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
 				.boxed()
 			},
-			ProtocolSide::ValidatorExperimental { keystore, metrics } => {
-				validator_side_experimental::run(ctx, keystore, metrics)
+			ProtocolSide::ValidatorExperimental { keystore, metrics, db, reputation_config } => {
+				validator_side_experimental::run(ctx, keystore, metrics, db, reputation_config)
 					.map_err(|e| SubsystemError::with_origin("collator-protocol", e))
 					.boxed()
 			},

polkadot/node/network/collator-protocol/src/validator_side_experimental/common.rs

@@ -14,6 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
+use codec::{Decode, Encode};
 use polkadot_node_network_protocol::{
 	peer_set::CollationVersion,
 	request_response::{outgoing::RequestError, v2 as request_v2},
@@ -84,7 +85,7 @@ pub const MAX_FETCH_DELAY: Duration = Duration::from_millis(300);
 pub const MIN_FETCH_TIMER_DELAY: Duration = Duration::from_millis(150);
 
 /// Reputation score type.
-#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone, Copy, Default)]
+#[derive(PartialEq, Eq, PartialOrd, Ord, Debug, Clone, Copy, Default, Encode, Decode)]
 pub struct Score(u16);
 
 impl Score {

polkadot/node/network/collator-protocol/src/validator_side_experimental/error.rs

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-use crate::LOG_TARGET;
+use crate::{validator_side_experimental::peer_manager::PersistenceError, LOG_TARGET};
 use fatality::Nested;
 use polkadot_node_subsystem::{ChainApiError, SubsystemError};
 use polkadot_node_subsystem_util::{backing_implicit_view, runtime};
@@ -44,6 +44,12 @@ pub enum Error {
 	#[fatal]
 	#[error("Receiving message from overseer failed: {0}")]
 	SubsystemReceive(#[source] SubsystemError),
+	#[fatal]
+	#[error("Failed to initialize reputation database: {0}")]
+	ReputationDbInit(PersistenceError),
+	#[fatal]
+	#[error("Failed to spawn background task: {0}")]
+	SpawnTask(String),
 	#[error("Unable to retrieve block number for {0:?} from implicit view")]
 	BlockNumberNotFoundInImplicitView(Hash),
 	#[fatal(forward)]

polkadot/node/network/collator-protocol/src/validator_side_experimental/mod.rs

@@ -22,7 +22,10 @@ mod state;
 #[cfg(test)]
 mod tests;
 
-use crate::{validator_side_experimental::common::MIN_FETCH_TIMER_DELAY, LOG_TARGET};
+use crate::{
+	validator_side_experimental::{common::MIN_FETCH_TIMER_DELAY, peer_manager::PersistentDb},
+	LOG_TARGET,
+};
 use collation_manager::CollationManager;
 use common::{ProspectiveCandidate, MAX_STORED_SCORES_PER_PARA};
 use error::{log_error, FatalError, FatalResult, Result};
@@ -35,25 +38,50 @@ use polkadot_node_subsystem::{
 	messages::{CollatorProtocolMessage, NetworkBridgeEvent},
 	overseer, ActivatedLeaf, CollatorProtocolSenderTrait, FromOrchestra, OverseerSignal,
 };
+use polkadot_node_subsystem_util::database::Database;
 use sp_keystore::KeystorePtr;
-use std::{future, future::Future, pin::Pin, time::Duration};
+use std::{future, future::Future, pin::Pin, sync::Arc, time::Duration};
 
-use peer_manager::{Db, PeerManager};
+#[cfg(test)]
+use peer_manager::Db;
+use peer_manager::PeerManager;
 
 use state::State;
 
 pub use crate::validator_side_metrics::Metrics;
 
+/// Default interval for persisting the reputation database to disk (in seconds).
+const DEFAULT_PERSIST_INTERVAL_SECS: u64 = 600;
+
+/// Configuration for the reputation db.
+#[derive(Debug, Clone, Copy)]
+pub struct ReputationConfig {
+	/// The data column in the store to use for reputation data.
+	pub col_reputation_data: u32,
+	/// How often to persist the reputation database to disk.
+	/// If None, defaults to DEFAULT_PERSIST_INTERVAL_SECS seconds.
+	pub persist_interval: Option<Duration>,
+}
+
 /// The main run loop.
 #[overseer::contextbounds(CollatorProtocol, prefix = self::overseer)]
 pub(crate) async fn run<Context>(
 	mut ctx: Context,
 	keystore: KeystorePtr,
 	metrics: Metrics,
+	db: Arc<dyn Database>,
+	reputation_config: ReputationConfig,
 ) -> FatalResult<()> {
-	gum::info!(LOG_TARGET, "Running experimental collator protocol");
-	if let Some(state) = initialize(&mut ctx, keystore, metrics).await? {
-		run_inner(ctx, state).await?;
+	let persist_interval = reputation_config
+		.persist_interval
+		.unwrap_or(Duration::from_secs(DEFAULT_PERSIST_INTERVAL_SECS));
+	gum::info!(
+		LOG_TARGET,
+		persist_interval_secs = persist_interval.as_secs(),
+		"Running experimental collator protocol"
+	);
+	if let Some(state) = initialize(&mut ctx, keystore, metrics, db, reputation_config).await? {
+		run_inner(ctx, state, persist_interval).await?;
 	}
 
 	Ok(())
@@ -64,7 +92,9 @@ async fn initialize<Context>(
 	ctx: &mut Context,
 	keystore: KeystorePtr,
 	metrics: Metrics,
-) -> FatalResult<Option<State<Db>>> {
+	db: Arc<dyn Database>,
+	reputation_config: ReputationConfig,
+) -> FatalResult<Option<State<PersistentDb>>> {
 	loop {
 		let first_leaf = match wait_for_first_leaf(ctx).await? {
 			Some(activated_leaf) => {
@@ -84,7 +114,30 @@ async fn initialize<Context>(
 
 		let scheduled_paras = collation_manager.assignments();
 
-		let backend = Db::new(MAX_STORED_SCORES_PER_PARA).await;
+		// Create PersistentDb with disk persistence
+		let (backend, task) = match PersistentDb::new(
+			db.clone(),
+			reputation_config,
+			MAX_STORED_SCORES_PER_PARA,
+		)
+		.await
+		{
+			Ok(result) => result,
+			Err(e) => {
+				gum::error!(
+					target: LOG_TARGET,
+					error = ?e,
+					"Failed to initialize persistent reputation DB"
+				);
+				return Err(FatalError::ReputationDbInit(e));
+			},
+		};
+
+		// Background task for async writes
+		ctx.spawn_blocking("collator-reputation-persistence-task", task)
+			.map_err(|e| FatalError::SpawnTask(e.to_string()))?;
+
+		gum::debug!(target: LOG_TARGET, "Spawned background reputation persistence task");
 
 		match PeerManager::startup(backend, ctx.sender(), scheduled_paras.into_iter().collect())
 			.await
@@ -134,9 +187,21 @@ fn create_timer(maybe_delay: Option<Duration>) -> Fuse<Pin<Box<dyn Future<Output
 	timer.fuse()
 }
 
+/// Create the persistence timer that fires after the given interval.
+fn create_persistence_timer(interval: Duration) -> Fuse<Pin<Box<dyn Future<Output = ()> + Send>>> {
+	let delay: Pin<Box<dyn Future<Output = ()> + Send>> = Box::pin(Delay::new(interval));
+	delay.fuse()
+}
+
 #[overseer::contextbounds(CollatorProtocol, prefix = self::overseer)]
-async fn run_inner<Context>(mut ctx: Context, mut state: State<Db>) -> FatalResult<()> {
+async fn run_inner<Context>(
+	mut ctx: Context,
+	mut state: State<PersistentDb>,
+	persist_interval: Duration,
+) -> FatalResult<()> {
 	let mut timer = create_timer(None);
+	let mut persistence_timer = create_persistence_timer(persist_interval);
+
 	loop {
 		select! {
 			// Calling `fuse()` here is useless, because the termination state of the resulting
@@ -154,7 +219,11 @@ async fn run_inner<Context>(mut ctx: Context, mut state: State<Db>) -> FatalResu
 							msg,
 						).await;
 					}
-					Ok(FromOrchestra::Signal(OverseerSignal::Conclude)) | Err(_) => break,
+					Ok(FromOrchestra::Signal(OverseerSignal::Conclude)) | Err(_) => {
+						// Persist to disk before shutdown
+						state.persist_reputations().await;
+						break
+					},
 					Ok(FromOrchestra::Signal(OverseerSignal::BlockFinalized(hash, number))) => {
 						state.handle_finalized_block(ctx.sender(), hash, number).await?;
 					},
@@ -168,6 +237,12 @@ async fn run_inner<Context>(mut ctx: Context, mut state: State<Db>) -> FatalResu
 				// We don't need to do anything specific here.
 				// If the timer expires, we only need to trigger the advertisement fetching logic.
 			},
+			_ = &mut persistence_timer => {
+				// Periodic persistence - write reputation DB to disk
+				state.background_persist_reputations();
+				// Reset the timer for the next interval
+				persistence_timer = create_persistence_timer(persist_interval);
+			},
 		}
 
 		// Now try triggering advertisement fetching, if we have room in any of the active leaves
@@ -187,7 +262,7 @@ async fn run_inner<Context>(mut ctx: Context, mut state: State<Db>) -> FatalResu
 /// The main message receiver switch.
 async fn process_msg<Sender: CollatorProtocolSenderTrait>(
 	sender: &mut Sender,
-	state: &mut State<Db>,
+	state: &mut State<PersistentDb>,
 	msg: CollatorProtocolMessage,
 ) {
 	use CollatorProtocolMessage::*;
@@ -241,7 +316,7 @@ async fn process_msg<Sender: CollatorProtocolSenderTrait>(
 /// Bridge event switch.
 async fn handle_network_msg<Sender: CollatorProtocolSenderTrait>(
 	sender: &mut Sender,
-	state: &mut State<Db>,
+	state: &mut State<PersistentDb>,
 	bridge_message: NetworkBridgeEvent<net_protocol::CollatorProtocolMessage>,
 ) -> Result<()> {
 	use NetworkBridgeEvent::*;
@@ -289,7 +364,7 @@ async fn handle_network_msg<Sender: CollatorProtocolSenderTrait>(
 
 async fn process_incoming_peer_message<Sender: CollatorProtocolSenderTrait>(
 	sender: &mut Sender,
-	state: &mut State<Db>,
+	state: &mut State<PersistentDb>,
 	origin: PeerId,
 	msg: CollationProtocols<
 		protocol_v1::CollatorProtocolMessage,