Make rolling session more resilient in case of long finality stalls

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

@@ -90,41 +90,45 @@ impl Backend for DbBackend {
 			match op {
 				BackendWriteOp::WriteStoredBlockRange(stored_block_range) => {
 					tx.put_vec(
-						self.config.col_data,
+						self.config.col_approval_data,
 						&STORED_BLOCKS_KEY,
 						stored_block_range.encode(),
 					);
 				},
 				BackendWriteOp::DeleteStoredBlockRange => {
-					tx.delete(self.config.col_data, &STORED_BLOCKS_KEY);
+					tx.delete(self.config.col_approval_data, &STORED_BLOCKS_KEY);
 				},
 				BackendWriteOp::WriteBlocksAtHeight(h, blocks) => {
-					tx.put_vec(self.config.col_data, &blocks_at_height_key(h), blocks.encode());
+					tx.put_vec(
+						self.config.col_approval_data,
+						&blocks_at_height_key(h),
+						blocks.encode(),
+					);
 				},
 				BackendWriteOp::DeleteBlocksAtHeight(h) => {
-					tx.delete(self.config.col_data, &blocks_at_height_key(h));
+					tx.delete(self.config.col_approval_data, &blocks_at_height_key(h));
 				},
 				BackendWriteOp::WriteBlockEntry(block_entry) => {
 					let block_entry: BlockEntry = block_entry.into();
 					tx.put_vec(
-						self.config.col_data,
+						self.config.col_approval_data,
 						&block_entry_key(&block_entry.block_hash),
 						block_entry.encode(),
 					);
 				},
 				BackendWriteOp::DeleteBlockEntry(hash) => {
-					tx.delete(self.config.col_data, &block_entry_key(&hash));
+					tx.delete(self.config.col_approval_data, &block_entry_key(&hash));
 				},
 				BackendWriteOp::WriteCandidateEntry(candidate_entry) => {
 					let candidate_entry: CandidateEntry = candidate_entry.into();
 					tx.put_vec(
-						self.config.col_data,
+						self.config.col_approval_data,
 						&candidate_entry_key(&candidate_entry.candidate.hash()),
 						candidate_entry.encode(),
 					);
 				},
 				BackendWriteOp::DeleteCandidateEntry(candidate_hash) => {
-					tx.delete(self.config.col_data, &candidate_entry_key(&candidate_hash));
+					tx.delete(self.config.col_approval_data, &candidate_entry_key(&candidate_hash));
 				},
 			}
 		}
@@ -149,7 +153,9 @@ pub type Bitfield = BitVec<u8, BitOrderLsb0>;
 #[derive(Debug, Clone, Copy)]
 pub struct Config {
 	/// The column family in the database where data is stored.
-	pub col_data: u32,
+	pub col_approval_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.
@@ -243,10 +249,10 @@ pub type Result<T> = std::result::Result<T, Error>;
 
 pub(crate) fn load_decode<D: Decode>(
 	store: &dyn Database,
-	col_data: u32,
+	col_approval_data: u32,
 	key: &[u8],
 ) -> Result<Option<D>> {
-	match store.get(col_data, key)? {
+	match store.get(col_approval_data, key)? {
 		None => Ok(None),
 		Some(raw) => D::decode(&mut &raw[..]).map(Some).map_err(Into::into),
 	}
@@ -303,7 +309,7 @@ pub fn load_stored_blocks(
 	store: &dyn Database,
 	config: &Config,
 ) -> SubsystemResult<Option<StoredBlockRange>> {
-	load_decode(store, config.col_data, STORED_BLOCKS_KEY)
+	load_decode(store, config.col_approval_data, STORED_BLOCKS_KEY)
 		.map_err(|e| SubsystemError::with_origin("approval-voting", e))
 }
 
@@ -313,7 +319,7 @@ pub fn load_blocks_at_height(
 	config: &Config,
 	block_number: &BlockNumber,
 ) -> SubsystemResult<Vec<Hash>> {
-	load_decode(store, config.col_data, &blocks_at_height_key(*block_number))
+	load_decode(store, config.col_approval_data, &blocks_at_height_key(*block_number))
 		.map(|x| x.unwrap_or_default())
 		.map_err(|e| SubsystemError::with_origin("approval-voting", e))
 }
@@ -324,7 +330,7 @@ pub fn load_block_entry(
 	config: &Config,
 	block_hash: &Hash,
 ) -> SubsystemResult<Option<BlockEntry>> {
-	load_decode(store, config.col_data, &block_entry_key(block_hash))
+	load_decode(store, config.col_approval_data, &block_entry_key(block_hash))
 		.map(|u: Option<BlockEntry>| u.map(|v| v.into()))
 		.map_err(|e| SubsystemError::with_origin("approval-voting", e))
 }
@@ -335,7 +341,7 @@ pub fn load_candidate_entry(
 	config: &Config,
 	candidate_hash: &CandidateHash,
 ) -> SubsystemResult<Option<CandidateEntry>> {
-	load_decode(store, config.col_data, &candidate_entry_key(candidate_hash))
+	load_decode(store, config.col_approval_data, &candidate_entry_key(candidate_hash))
 		.map(|u: Option<CandidateEntry>| u.map(|v| v.into()))
 		.map_err(|e| SubsystemError::with_origin("approval-voting", e))
 }

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

@@ -28,9 +28,12 @@ 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_approval_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

@@ -632,14 +632,15 @@ pub(crate) mod tests {
 	pub(crate) use sp_runtime::{Digest, DigestItem};
 	use std::{pin::Pin, sync::Arc};
 
-	use crate::{
-		approval_db::v1::Config as DatabaseConfig, criteria, BlockEntry, APPROVAL_SESSIONS,
-	};
+	use crate::{approval_db::v1::Config as DatabaseConfig, criteria, BlockEntry};
 
 	const DATA_COL: u32 = 0;
-	const NUM_COLUMNS: u32 = 1;
+	const SESSION_DATA_COL: u32 = 1;
+
+	const NUM_COLUMNS: u32 = 2;
 
-	const TEST_CONFIG: DatabaseConfig = DatabaseConfig { col_data: DATA_COL };
+	const TEST_CONFIG: DatabaseConfig =
+		DatabaseConfig { col_approval_data: DATA_COL, col_session_data: SESSION_DATA_COL };
 	#[derive(Default)]
 	struct MockClock;
 
@@ -654,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()
 		}
 	}
@@ -782,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 {
@@ -891,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 {
@@ -1089,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 {
@@ -1304,38 +1297,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!(