client: allow reverting finalized blocks (#4535)

* client: allow reverting blocks past finality

* client: fix leaves reversion

* client: extend docs on revert

* client: add comment on leaves revert

substrate/client/api/src/backend.rs

@@ -283,10 +283,16 @@ pub trait Backend<Block, H>: AuxStore + Send + Sync where
 		Ok(())
 	}
 
-	/// Attempts to revert the chain by `n` blocks.
+	/// Attempts to revert the chain by `n` blocks. If `revert_finalized` is set
+	/// it will attempt to revert past any finalized block, this is unsafe and
+	/// can potentially leave the node in an inconsistent state.
 	///
 	/// Returns the number of blocks that were successfully reverted.
-	fn revert(&self, n: NumberFor<Block>) -> sp_blockchain::Result<NumberFor<Block>>;
+	fn revert(
+		&self,
+		n: NumberFor<Block>,
+		revert_finalized: bool,
+	) -> sp_blockchain::Result<NumberFor<Block>>;
 
 	/// Insert auxiliary data into key-value store.
 	fn insert_aux<

substrate/client/db/src/lib.rs

@@ -1484,40 +1484,71 @@ impl<Block> sc_client_api::backend::Backend<Block, Blake2Hasher> for Backend<Blo
 		Some(self.offchain_storage.clone())
 	}
 
-	fn revert(&self, n: NumberFor<Block>) -> ClientResult<NumberFor<Block>> {
-		let mut best = self.blockchain.info().best_number;
+	fn revert(&self, n: NumberFor<Block>, revert_finalized: bool) -> ClientResult<NumberFor<Block>> {
+		let mut best_number = self.blockchain.info().best_number;
+		let mut best_hash = self.blockchain.info().best_hash;
 		let finalized = self.blockchain.info().finalized_number;
-		let revertible = best - finalized;
-		let n = if revertible < n { revertible } else { n };
 
-		for c in 0 .. n.saturated_into::<u64>() {
-			if best.is_zero() {
-				return Ok(c.saturated_into::<NumberFor<Block>>())
-			}
-			let mut transaction = DBTransaction::new();
-			match self.storage.state_db.revert_one() {
-				Some(commit) => {
-					apply_state_commit(&mut transaction, commit);
-					let removed = self.blockchain.header(BlockId::Number(best))?.ok_or_else(
-						|| sp_blockchain::Error::UnknownBlock(
-							format!("Error reverting to {}. Block hash not found.", best)))?;
+		let revertible = best_number - finalized;
+		let n = if !revert_finalized && revertible < n {
+			revertible
+		} else {
+			n
+		};
 
-					best -= One::one();	// prev block
-					let hash = self.blockchain.hash(best)?.ok_or_else(
-						|| sp_blockchain::Error::UnknownBlock(
-							format!("Error reverting to {}. Block hash not found.", best)))?;
-					let key = utils::number_and_hash_to_lookup_key(best.clone(), &hash)?;
-					transaction.put(columns::META, meta_keys::BEST_BLOCK, &key);
-					transaction.delete(columns::KEY_LOOKUP, removed.hash().as_ref());
-					children::remove_children(&mut transaction, columns::META, meta_keys::CHILDREN_PREFIX, hash);
-					self.storage.db.write(transaction).map_err(db_err)?;
-					self.blockchain.update_meta(hash, best, true, false);
-					self.blockchain.leaves.write().revert(removed.hash().clone(), removed.number().clone(), removed.parent_hash().clone());
+		let mut revert_blocks = || -> ClientResult<NumberFor<Block>> {
+			for c in 0 .. n.saturated_into::<u64>() {
+				if best_number.is_zero() {
+					return Ok(c.saturated_into::<NumberFor<Block>>())
+				}
+				let mut transaction = DBTransaction::new();
+				match self.storage.state_db.revert_one() {
+					Some(commit) => {
+						apply_state_commit(&mut transaction, commit);
+						let removed = self.blockchain.header(BlockId::Number(best_number))?.ok_or_else(
+							|| sp_blockchain::Error::UnknownBlock(
+								format!("Error reverting to {}. Block hash not found.", best_number)))?;
+
+						best_number -= One::one();	// prev block
+						best_hash = self.blockchain.hash(best_number)?.ok_or_else(
+							|| sp_blockchain::Error::UnknownBlock(
+								format!("Error reverting to {}. Block hash not found.", best_number)))?;
+
+						let update_finalized = best_number < finalized;
+
+						let key = utils::number_and_hash_to_lookup_key(best_number.clone(), &best_hash)?;
+						transaction.put(columns::META, meta_keys::BEST_BLOCK, &key);
+						if update_finalized {
+							transaction.put(columns::META, meta_keys::FINALIZED_BLOCK, &key);
+						}
+						transaction.delete(columns::KEY_LOOKUP, removed.hash().as_ref());
+						children::remove_children(&mut transaction, columns::META, meta_keys::CHILDREN_PREFIX, best_hash);
+						self.storage.db.write(transaction).map_err(db_err)?;
+						self.blockchain.update_meta(best_hash, best_number, true, update_finalized);
+					}
+					None => return Ok(c.saturated_into::<NumberFor<Block>>())
 				}
-				None => return Ok(c.saturated_into::<NumberFor<Block>>())
 			}
-		}
-		Ok(n)
+
+			Ok(n)
+		};
+
+		let reverted = revert_blocks()?;
+
+		let revert_leaves = || -> ClientResult<()> {
+			let mut transaction = DBTransaction::new();
+			let mut leaves = self.blockchain.leaves.write();
+
+			leaves.revert(best_hash, best_number);
+			leaves.prepare_transaction(&mut transaction, columns::META, meta_keys::LEAF_PREFIX);
+			self.storage.db.write(transaction).map_err(db_err)?;
+
+			Ok(())
+		};
+
+		revert_leaves()?;
+
+		Ok(reverted)
 	}
 
 	fn blockchain(&self) -> &BlockchainDb<Block> {

substrate/client/src/client.rs

@@ -1142,10 +1142,24 @@ impl<B, E, Block, RA> Client<B, E, Block, RA> where
 		Ok(())
 	}
 
-	/// Attempts to revert the chain by `n` blocks. Returns the number of blocks that were
-	/// successfully reverted.
+	/// Attempts to revert the chain by `n` blocks guaranteeing that no block is
+	/// reverted past the last finalized block. Returns the number of blocks
+	/// that were successfully reverted.
 	pub fn revert(&self, n: NumberFor<Block>) -> sp_blockchain::Result<NumberFor<Block>> {
-		Ok(self.backend.revert(n)?)
+		Ok(self.backend.revert(n, false)?)
+	}
+
+	/// Attempts to revert the chain by `n` blocks disregarding finality. This
+	/// method will revert any finalized blocks as requested and can potentially
+	/// lead the node in an inconsistent state. Other modules in the system that
+	/// persist data and that rely on finality (e.g. consensus parts) will be
+	/// unaffected by the revert. Use this method with caution and making sure
+	/// that no other data needs to be reverted for consistency aside from the
+	/// block data.
+	///
+	/// Returns the number of blocks that were successfully reverted.
+	pub fn unsafe_revert(&self, n: NumberFor<Block>) -> sp_blockchain::Result<NumberFor<Block>> {
+		Ok(self.backend.revert(n, true)?)
 	}
 
 	/// Get usage info about current client.

substrate/client/src/in_mem.rs

@@ -707,7 +707,11 @@ where
 		}
 	}
 
-	fn revert(&self, _n: NumberFor<Block>) -> sp_blockchain::Result<NumberFor<Block>> {
+	fn revert(
+		&self,
+		_n: NumberFor<Block>,
+		_revert_finalized: bool,
+	) -> sp_blockchain::Result<NumberFor<Block>> {
 		Ok(Zero::zero())
 	}
 

substrate/client/src/leaves.rs

@@ -158,12 +158,35 @@ impl<H, N> LeafSet<H, N> where
 		Undo { inner: self }
 	}
 
-	/// currently since revert only affects the canonical chain
-	/// we assume that parent has no further children
-	/// and we add it as leaf again
-	pub fn revert(&mut self, hash: H, number: N, parent_hash: H) {
-		self.insert_leaf(Reverse(number.clone() - N::one()), parent_hash);
-		self.remove_leaf(&Reverse(number), &hash);
+	/// Revert to the given block height by dropping all leaves in the leaf set
+	/// with a block number higher than the target.
+	pub fn revert(&mut self, best_hash: H, best_number: N) {
+		let items = self.storage.iter()
+			.flat_map(|(number, hashes)| hashes.iter().map(move |h| (h.clone(), number.clone())))
+			.collect::<Vec<_>>();
+
+		for (hash, number) in &items {
+			if number.0 > best_number {
+				assert!(
+					self.remove_leaf(number, hash),
+					"item comes from an iterator over storage; qed",
+				);
+
+				self.pending_removed.push(hash.clone());
+			}
+		}
+
+		let best_number = Reverse(best_number);
+		let leaves_contains_best = self.storage
+			.get(&best_number)
+			.map_or(false, |hashes| hashes.contains(&best_hash));
+
+		// we need to make sure that the best block exists in the leaf set as
+		// this is an invariant of regular block import.
+		if !leaves_contains_best {
+			self.insert_leaf(best_number.clone(), best_hash.clone());
+			self.pending_added.push(LeafSetItem { hash: best_hash, number: best_number });
+		}
 	}
 
 	/// returns an iterator over all hashes in the leaf set

substrate/client/src/light/backend.rs

@@ -213,7 +213,11 @@ impl<S, Block, H> ClientBackend<Block, H> for Backend<S, H> where
 		Ok(GenesisOrUnavailableState::Unavailable)
 	}
 
-	fn revert(&self, _n: NumberFor<Block>) -> ClientResult<NumberFor<Block>> {
+	fn revert(
+		&self,
+		_n: NumberFor<Block>,
+		_revert_finalized: bool,
+	) -> ClientResult<NumberFor<Block>> {
 		Err(ClientError::NotAvailableOnLightClient)
 	}