Keep sessions in window for the full unfinalized chain (#6054)

* Impl dynamic window size. Keep sessions for unfinalized chain

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

* feedback

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

* Stretch also in contructor plus  tests

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

* review feedback

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

* fix approval-voting tests

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

* grunting: dispute coordinator tests

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

Signed-off-by: Andrei Sandu <andrei-mihail@parity.io>

Author: sandreim

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

@@ -1296,6 +1296,38 @@ 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/tests.rs

@@ -807,6 +807,37 @@ async fn import_block(
 		}
 	);
 
+	assert_matches!(
+		overseer_recv(overseer).await,
+		AllMessages::ChainApi(ChainApiMessage::FinalizedBlockNumber(
+			s_tx,
+		)) => {
+			let _ = s_tx.send(Ok(number));
+		}
+	);
+
+	assert_matches!(
+		overseer_recv(overseer).await,
+		AllMessages::ChainApi(ChainApiMessage::FinalizedBlockHash(
+			block_number,
+			s_tx,
+		)) => {
+			assert_eq!(block_number, number);
+			let _ = s_tx.send(Ok(Some(hashes[number as usize].0)));
+		}
+	);
+
+	assert_matches!(
+		overseer_recv(overseer).await,
+		AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+			h,
+			RuntimeApiRequest::SessionIndexForChild(s_tx),
+		)) => {
+			assert_eq!(h, hashes[number as usize].0);
+			let _ = s_tx.send(Ok(number.into()));
+		}
+	);
+
 	if !fork {
 		assert_matches!(
 			overseer_recv(overseer).await,

node/core/dispute-coordinator/src/tests.rs

@@ -239,13 +239,15 @@ impl TestState {
 			)))
 			.await;
 
-		self.handle_sync_queries(virtual_overseer, block_hash, session).await;
+		self.handle_sync_queries(virtual_overseer, block_hash, block_number, session)
+			.await;
 	}
 
 	async fn handle_sync_queries(
 		&mut self,
 		virtual_overseer: &mut VirtualOverseer,
 		block_hash: Hash,
+		block_number: BlockNumber,
 		session: SessionIndex,
 	) {
 		// Order of messages is not fixed (different on initializing):
@@ -278,11 +280,45 @@ impl TestState {
 					finished_steps.got_session_information = true;
 					assert_eq!(h, block_hash);
 					let _ = tx.send(Ok(session));
+
+					// Queries for fetching earliest unfinalized block session. See `RollingSessionWindow`.
+					assert_matches!(
+						overseer_recv(virtual_overseer).await,
+						AllMessages::ChainApi(ChainApiMessage::FinalizedBlockNumber(
+							s_tx,
+						)) => {
+							let _ = s_tx.send(Ok(block_number));
+						}
+					);
+
+					assert_matches!(
+						overseer_recv(virtual_overseer).await,
+						AllMessages::ChainApi(ChainApiMessage::FinalizedBlockHash(
+							number,
+							s_tx,
+						)) => {
+							assert_eq!(block_number, number);
+							let _ = s_tx.send(Ok(Some(block_hash)));
+						}
+					);
+
+					assert_matches!(
+						overseer_recv(virtual_overseer).await,
+						AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+							h,
+							RuntimeApiRequest::SessionIndexForChild(s_tx),
+						)) => {
+							assert_eq!(h, block_hash);
+							let _ = s_tx.send(Ok(session));
+						}
+					);
+
 					// No queries, if subsystem knows about this session already.
 					if self.known_session == Some(session) {
 						continue
 					}
 					self.known_session = Some(session);
+
 					loop {
 						// answer session info queries until the current session is reached.
 						assert_matches!(
@@ -361,7 +397,8 @@ impl TestState {
 				)))
 				.await;
 
-			self.handle_sync_queries(virtual_overseer, *leaf, session).await;
+			self.handle_sync_queries(virtual_overseer, *leaf, n as BlockNumber, session)
+				.await;
 		}
 	}