Make rolling session more resilient in case of long finality stalls

node/core/approval-voting/src/approval_db/v1/mod.rs

@@ -150,6 +150,8 @@ pub type Bitfield = BitVec<u8, BitOrderLsb0>;
 pub struct Config {
 	/// The column family in the database where data is stored.
 	pub col_data: u32,
+	/// The column of the database where rolling session window data is stored.
+	pub col_session_data: u32,
 }
 
 /// Details pertaining to our assignment on a block.

node/core/approval-voting/src/approval_db/v1/tests.rs

@@ -28,9 +28,11 @@ use std::{collections::HashMap, sync::Arc};
 use ::test_helpers::{dummy_candidate_receipt, dummy_candidate_receipt_bad_sig, dummy_hash};
 
 const DATA_COL: u32 = 0;
-const NUM_COLUMNS: u32 = 1;
+const SESSION_DATA_COL: u32 = 1;
 
-const TEST_CONFIG: Config = Config { col_data: DATA_COL };
+const NUM_COLUMNS: u32 = 2;
+
+const TEST_CONFIG: Config = Config { col_data: DATA_COL, col_session_data: SESSION_DATA_COL };
 
 fn make_db() -> (DbBackend, Arc<dyn Database>) {
 	let db = kvdb_memorydb::create(NUM_COLUMNS);

node/core/approval-voting/src/import.rs

@@ -635,9 +635,12 @@ pub(crate) mod tests {
 	};
 
 	const DATA_COL: u32 = 0;
-	const NUM_COLUMNS: u32 = 1;
+	const SESSION_DATA_COL: u32 = 1;
 
-	const TEST_CONFIG: DatabaseConfig = DatabaseConfig { col_data: DATA_COL };
+	const NUM_COLUMNS: u32 = 2;
+
+	const TEST_CONFIG: DatabaseConfig =
+		DatabaseConfig { col_data: DATA_COL, col_session_data: SESSION_DATA_COL };
 	#[derive(Default)]
 	struct MockClock;
 
@@ -652,22 +655,23 @@ pub(crate) mod tests {
 	}
 
 	fn blank_state() -> State {
+		let db = kvdb_memorydb::create(NUM_COLUMNS);
+		let db = polkadot_node_subsystem_util::database::kvdb_impl::DbAdapter::new(db, &[]);
+		let db: Arc<dyn Database> = Arc::new(db);
 		State {
 			session_window: None,
 			keystore: Arc::new(LocalKeystore::in_memory()),
 			slot_duration_millis: 6_000,
 			clock: Box::new(MockClock::default()),
 			assignment_criteria: Box::new(MockAssignmentCriteria),
+			db,
+			db_config: TEST_CONFIG,
 		}
 	}
 
 	fn single_session_state(index: SessionIndex, info: SessionInfo) -> State {
 		State {
-			session_window: Some(RollingSessionWindow::with_session_info(
-				APPROVAL_SESSIONS,
-				index,
-				vec![info],
-			)),
+			session_window: Some(RollingSessionWindow::with_session_info(index, vec![info])),
 			..blank_state()
 		}
 	}
@@ -780,11 +784,8 @@ pub(crate) mod tests {
 				.map(|(r, c, g)| (r.hash(), r.clone(), *c, *g))
 				.collect::<Vec<_>>();
 
-			let session_window = RollingSessionWindow::with_session_info(
-				APPROVAL_SESSIONS,
-				session,
-				vec![session_info],
-			);
+			let session_window =
+				RollingSessionWindow::with_session_info(session, vec![session_info]);
 
 			let header = header.clone();
 			Box::pin(async move {
@@ -889,11 +890,8 @@ pub(crate) mod tests {
 			.collect::<Vec<_>>();
 
 		let test_fut = {
-			let session_window = RollingSessionWindow::with_session_info(
-				APPROVAL_SESSIONS,
-				session,
-				vec![session_info],
-			);
+			let session_window =
+				RollingSessionWindow::with_session_info(session, vec![session_info]);
 
 			let header = header.clone();
 			Box::pin(async move {
@@ -1087,11 +1085,8 @@ pub(crate) mod tests {
 				.map(|(r, c, g)| (r.hash(), r.clone(), *c, *g))
 				.collect::<Vec<_>>();
 
-			let session_window = Some(RollingSessionWindow::with_session_info(
-				APPROVAL_SESSIONS,
-				session,
-				vec![session_info],
-			));
+			let session_window =
+				Some(RollingSessionWindow::with_session_info(session, vec![session_info]));
 
 			let header = header.clone();
 			Box::pin(async move {
@@ -1296,38 +1291,6 @@ pub(crate) mod tests {
 				}
 			);
 
-			// Caching of sesssions needs sessoion of first unfinalied block.
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::ChainApi(ChainApiMessage::FinalizedBlockNumber(
-					s_tx,
-				)) => {
-					let _ = s_tx.send(Ok(header.number));
-				}
-			);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::ChainApi(ChainApiMessage::FinalizedBlockHash(
-					block_number,
-					s_tx,
-				)) => {
-					assert_eq!(block_number, header.number);
-					let _ = s_tx.send(Ok(Some(header.hash())));
-				}
-			);
-
-			assert_matches!(
-				handle.recv().await,
-				AllMessages::RuntimeApi(RuntimeApiMessage::Request(
-					h,
-					RuntimeApiRequest::SessionIndexForChild(s_tx),
-				)) => {
-					assert_eq!(h, header.hash());
-					let _ = s_tx.send(Ok(session));
-				}
-			);
-
 			// determine_new_blocks exits early as the parent_hash is in the DB
 
 			assert_matches!(

node/core/approval-voting/src/lib.rs

@@ -44,7 +44,7 @@ use polkadot_node_subsystem_util::{
 	database::Database,
 	metrics::{self, prometheus},
 	rolling_session_window::{
-		new_session_window_size, RollingSessionWindow, SessionWindowSize, SessionWindowUpdate,
+		DatabaseParams, RollingSessionWindow, SessionWindowSize, SessionWindowUpdate,
 		SessionsUnavailable,
 	},
 	TimeoutExt,
@@ -97,8 +97,6 @@ use crate::{
 #[cfg(test)]
 mod tests;
 
-pub const APPROVAL_SESSIONS: SessionWindowSize = new_session_window_size!(6);
-
 const APPROVAL_CHECKING_TIMEOUT: Duration = Duration::from_secs(120);
 /// How long are we willing to wait for approval signatures?
 ///
@@ -119,6 +117,8 @@ const LOG_TARGET: &str = "parachain::approval-voting";
 pub struct Config {
 	/// The column family in the DB where approval-voting data is stored.
 	pub col_data: u32,
+	/// The of the DB where rolling session info is stored.
+	pub col_session_data: u32,
 	/// The slot duration of the consensus algorithm, in milliseconds. Should be evenly
 	/// divisible by 500.
 	pub slot_duration_millis: u64,
@@ -358,7 +358,10 @@ impl ApprovalVotingSubsystem {
 			keystore,
 			slot_duration_millis: config.slot_duration_millis,
 			db,
-			db_config: DatabaseConfig { col_data: config.col_data },
+			db_config: DatabaseConfig {
+				col_data: config.col_data,
+				col_session_data: config.col_session_data,
+			},
 			mode: Mode::Syncing(sync_oracle),
 			metrics,
 		}
@@ -367,7 +370,10 @@ impl ApprovalVotingSubsystem {
 	/// Revert to the block corresponding to the specified `hash`.
 	/// The operation is not allowed for blocks older than the last finalized one.
 	pub fn revert_to(&self, hash: Hash) -> Result<(), SubsystemError> {
-		let config = approval_db::v1::Config { col_data: self.db_config.col_data };
+		let config = approval_db::v1::Config {
+			col_data: self.db_config.col_data,
+			col_session_data: self.db_config.col_session_data,
+		};
 		let mut backend = approval_db::v1::DbBackend::new(self.db.clone(), config);
 		let mut overlay = OverlayedBackend::new(&backend);
 
@@ -615,6 +621,9 @@ struct State {
 	slot_duration_millis: u64,
 	clock: Box<dyn Clock + Send + Sync>,
 	assignment_criteria: Box<dyn AssignmentCriteria + Send + Sync>,
+	// Require for `RollingSessionWindow`.
+	db_config: DatabaseConfig,
+	db: Arc<dyn Database>,
 }
 
 #[overseer::contextbounds(ApprovalVoting, prefix = self::overseer)]
@@ -636,8 +645,17 @@ impl State {
 		match session_window {
 			None => {
 				let sender = ctx.sender().clone();
-				self.session_window =
-					Some(RollingSessionWindow::new(sender, APPROVAL_SESSIONS, head).await?);
+				self.session_window = Some(
+					RollingSessionWindow::new(
+						sender,
+						head,
+						DatabaseParams {
+							db: self.db.clone(),
+							db_column: self.db_config.col_session_data,
+						},
+					)
+					.await?,
+				);
 				Ok(None)
 			},
 			Some(mut session_window) => {
@@ -732,12 +750,17 @@ async fn run<B, Context>(
 where
 	B: Backend,
 {
+	let db = subsystem.db.clone();
+	let db_config = subsystem.db_config.clone();
+
 	let mut state = State {
 		session_window: None,
 		keystore: subsystem.keystore,
 		slot_duration_millis: subsystem.slot_duration_millis,
 		clock,
 		assignment_criteria,
+		db_config,
+		db,
 	};
 
 	let mut wakeups = Wakeups::default();