sc-consensus-beefy: graceful support for pallet-beefy reset (#14217)

BEEFY consensus can be restarted by resetting "genesisBlock" in
pallet-beefy, but we don't want to also reset authority set IDs so
that they are uniquely identified across the entire chain history
regardless of how many times BEEFY consensus has been reset/restarted.

This is why the client now also accepts initial authority_set_id != 0.

BEEFY client now detects pallet-beefy reset/reinit and errors-out and
asks for a restart.
BEEFY client persisted state should be discarded on client restarts
following pallet-beefy reset/reinit.

End result is BEEFY client/voter can now completely reinitialize using
"new" on-chain info following pallet-beefy reset/reinit, discarding old state.

Fixes #14203
Fixes #14204

Signed-off-by: acatangiu <adrian@parity.io>

substrate/client/consensus/beefy/src/worker.rs

@@ -69,6 +69,9 @@ pub(crate) enum RoundAction {
 /// Responsible for the voting strategy.
 /// It chooses which incoming votes to accept and which votes to generate.
 /// Keeps track of voting seen for current and future rounds.
+///
+/// Note: this is part of `PersistedState` so any changes here should also bump
+/// aux-db schema version.
 #[derive(Debug, Decode, Encode, PartialEq)]
 pub(crate) struct VoterOracle<B: Block> {
 	/// Queue of known sessions. Keeps track of voting rounds (block numbers) within each session.
@@ -256,6 +259,9 @@ impl<B: Block> VoterOracle<B> {
 	}
 }
 
+/// BEEFY voter state persisted in aux DB.
+///
+/// Note: Any changes here should also bump aux-db schema version.
 #[derive(Debug, Decode, Encode, PartialEq)]
 pub(crate) struct PersistedState<B: Block> {
 	/// Best block we voted on.
@@ -263,6 +269,8 @@ pub(crate) struct PersistedState<B: Block> {
 	/// Chooses which incoming votes to accept and which votes to generate.
 	/// Keeps track of voting seen for current and future rounds.
 	voting_oracle: VoterOracle<B>,
+	/// Pallet-beefy genesis block - block number when BEEFY consensus started for this chain.
+	pallet_genesis: NumberFor<B>,
 }
 
 impl<B: Block> PersistedState<B> {
@@ -271,9 +279,19 @@ impl<B: Block> PersistedState<B> {
 		best_beefy: NumberFor<B>,
 		sessions: VecDeque<Rounds<B>>,
 		min_block_delta: u32,
+		pallet_genesis: NumberFor<B>,
 	) -> Option<Self> {
-		VoterOracle::checked_new(sessions, min_block_delta, grandpa_header, best_beefy)
-			.map(|voting_oracle| PersistedState { best_voted: Zero::zero(), voting_oracle })
+		VoterOracle::checked_new(sessions, min_block_delta, grandpa_header, best_beefy).map(
+			|voting_oracle| PersistedState {
+				best_voted: Zero::zero(),
+				voting_oracle,
+				pallet_genesis,
+			},
+		)
+	}
+
+	pub fn pallet_genesis(&self) -> NumberFor<B> {
+		self.pallet_genesis
 	}
 
 	pub(crate) fn set_min_block_delta(&mut self, min_block_delta: u32) {
@@ -411,7 +429,10 @@ where
 		);
 	}
 
-	fn handle_finality_notification(&mut self, notification: &FinalityNotification<B>) {
+	fn handle_finality_notification(
+		&mut self,
+		notification: &FinalityNotification<B>,
+	) -> Result<(), Error> {
 		debug!(
 			target: LOG_TARGET,
 			"🥩 Finality notification: header {:?} tree_route {:?}",
@@ -420,6 +441,18 @@ where
 		);
 		let header = &notification.header;
 
+		self.runtime
+			.runtime_api()
+			.beefy_genesis(header.hash())
+			.ok()
+			.flatten()
+			.filter(|genesis| *genesis == self.persisted_state.pallet_genesis)
+			.ok_or_else(|| {
+				let err = Error::ConsensusReset;
+				error!(target: LOG_TARGET, "🥩 Error: {}", err);
+				err
+			})?;
+
 		if *header.number() > self.best_grandpa_block() {
 			// update best GRANDPA finalized block we have seen
 			self.persisted_state.set_best_grandpa(header.clone());
@@ -451,6 +484,8 @@ where
 				error!(target: LOG_TARGET, "🥩 Voter error: {:?}", e);
 			}
 		}
+
+		Ok(())
 	}
 
 	/// Based on [VoterOracle] this vote is either processed here or discarded.
@@ -813,9 +848,9 @@ where
 				// Use `select_biased!` to prioritize order below.
 				// Process finality notifications first since these drive the voter.
 				notification = finality_notifications.next() => {
-					if let Some(notification) = notification {
-						self.handle_finality_notification(&notification);
-					} else {
+					if notification.and_then(|notif| {
+						self.handle_finality_notification(&notif).ok()
+					}).is_none() {
 						error!(target: LOG_TARGET, "🥩 Finality stream terminated, closing worker.");
 						return;
 					}
@@ -1086,6 +1121,7 @@ pub(crate) mod tests {
 		};
 
 		let backend = peer.client().as_backend();
+		let beefy_genesis = 1;
 		let api = Arc::new(TestApi::with_validator_set(&genesis_validator_set));
 		let network = peer.network_service().clone();
 		let sync = peer.sync_service().clone();
@@ -1118,6 +1154,7 @@ pub(crate) mod tests {
 			Zero::zero(),
 			vec![Rounds::new(One::one(), genesis_validator_set)].into(),
 			min_block_delta,
+			beefy_genesis,
 		)
 		.unwrap();
 		let payload_provider = MmrRootProvider::new(api.clone());