Forward/Revert blocks to the specified height

client/consensus/manual-seal/src/rpc.rs

@@ -3,6 +3,8 @@
 use crate::error::Error;
 use futures::{
 	channel::{mpsc, oneshot},
+	prelude::*,
+	stream::StreamExt,
 	SinkExt,
 };
 use jsonrpsee::{
@@ -11,7 +13,12 @@ use jsonrpsee::{
 };
 use sc_consensus::ImportedAux;
 use serde::{Deserialize, Serialize};
-use sp_runtime::EncodedJustification;
+use sp_blockchain::HeaderBackend;
+use sp_runtime::{
+	traits::{Block as BlockT, Header},
+	EncodedJustification, SaturatedConversion,
+};
+use std::sync::Arc;
 
 /// Sender passed to the authorship task to report errors or successes.
 pub type Sender<T> = Option<oneshot::Sender<std::result::Result<T, Error>>>;
@@ -48,28 +55,45 @@ pub enum EngineCommand<Hash> {
 
 /// RPC trait that provides methods for interacting with the manual-seal authorship task over rpc.
 #[rpc(client, server)]
-pub trait ManualSealApi<Hash> {
+pub trait ManualSealApi<Block>
+where
+	Block: BlockT,
+{
 	/// Instructs the manual-seal authorship task to create a new block
 	#[method(name = "engine_createBlock")]
 	async fn create_block(
 		&self,
 		create_empty: bool,
 		finalize: bool,
-		parent_hash: Option<Hash>,
-	) -> RpcResult<CreatedBlock<Hash>>;
+		parent_hash: Option<Block::Hash>,
+	) -> RpcResult<CreatedBlock<Block::Hash>>;
 
 	/// Instructs the manual-seal authorship task to finalize a block
 	#[method(name = "engine_finalizeBlock")]
 	async fn finalize_block(
 		&self,
-		hash: Hash,
+		hash: Block::Hash,
 		justification: Option<EncodedJustification>,
 	) -> RpcResult<bool>;
+
+	#[method(name = "engine_forwardBlocksTo")]
+	async fn forward_blocks_to(
+		&self,
+		height: <<Block as BlockT>::Header as Header>::Number,
+	) -> RpcResult<()>;
+
+	#[method(name = "engine_revertBlocksTo")]
+	async fn revert_blocks_to(
+		&self,
+		height: <<Block as BlockT>::Header as Header>::Number,
+	) -> RpcResult<()>;
 }
 
 /// A struct that implements the [`ManualSealApiServer`].
-pub struct ManualSeal<Hash> {
-	import_block_channel: mpsc::Sender<EngineCommand<Hash>>,
+pub struct ManualSeal<Block: BlockT, Client, Backend> {
+	client: Arc<Client>,
+	backend: Arc<Backend>,
+	import_block_channel: mpsc::Sender<EngineCommand<Block::Hash>>,
 }
 
 /// return type of `engine_createBlock`
@@ -81,21 +105,32 @@ pub struct CreatedBlock<Hash> {
 	pub aux: ImportedAux,
 }
 
-impl<Hash> ManualSeal<Hash> {
+impl<Block: BlockT, Client, Backend> ManualSeal<Block, Client, Backend> {
 	/// Create new `ManualSeal` with the given reference to the client.
-	pub fn new(import_block_channel: mpsc::Sender<EngineCommand<Hash>>) -> Self {
-		Self { import_block_channel }
+	pub fn new(
+		client: Arc<Client>,
+		backend: Arc<Backend>,
+		import_block_channel: mpsc::Sender<EngineCommand<Block::Hash>>,
+	) -> Self {
+		Self { client, backend, import_block_channel }
 	}
 }
 
 #[async_trait]
-impl<Hash: Send + 'static> ManualSealApiServer<Hash> for ManualSeal<Hash> {
+impl<Block, Client, Backend> ManualSealApiServer<Block> for ManualSeal<Block, Client, Backend>
+where
+	Block: BlockT,
+	Client: sp_api::ProvideRuntimeApi<Block>,
+	Client: HeaderBackend<Block>,
+	Client: Send + Sync + 'static,
+	Backend: sc_client_api::backend::Backend<Block> + Send + Sync + 'static,
+{
 	async fn create_block(
 		&self,
 		create_empty: bool,
 		finalize: bool,
-		parent_hash: Option<Hash>,
-	) -> RpcResult<CreatedBlock<Hash>> {
+		parent_hash: Option<Block::Hash>,
+	) -> RpcResult<CreatedBlock<Block::Hash>> {
 		let mut sink = self.import_block_channel.clone();
 		let (sender, receiver) = oneshot::channel();
 		// NOTE: this sends a Result over the channel.
@@ -117,7 +152,7 @@ impl<Hash: Send + 'static> ManualSealApiServer<Hash> for ManualSeal<Hash> {
 
 	async fn finalize_block(
 		&self,
-		hash: Hash,
+		hash: Block::Hash,
 		justification: Option<EncodedJustification>,
 	) -> RpcResult<bool> {
 		let mut sink = self.import_block_channel.clone();
@@ -126,6 +161,61 @@ impl<Hash: Send + 'static> ManualSealApiServer<Hash> for ManualSeal<Hash> {
 		sink.send(command).await?;
 		receiver.await.map(|_| true).map_err(|e| JsonRpseeError::to_call_error(e))
 	}
+
+	async fn forward_blocks_to(
+		&self,
+		height: <<Block as BlockT>::Header as Header>::Number,
+	) -> RpcResult<()> {
+		let best_number = self.client.info().best_number;
+		if height <= best_number {
+			return Err(JsonRpseeError::Custom(
+				"Target height is lower than current best height".into(),
+			))
+		}
+
+		let diff = height - best_number;
+		let to_height = (0..diff.saturated_into::<u64>())
+			.into_iter()
+			.map(|_| EngineCommand::SealNewBlock {
+				create_empty: true,
+				finalize: false,
+				parent_hash: None,
+				sender: None,
+			})
+			.collect::<Vec<EngineCommand<Block::Hash>>>();
+
+		let mut forward_blocks_stream = stream::iter(to_height).map(Ok);
+
+		let mut sink = self.import_block_channel.clone();
+		sink.send_all(&mut forward_blocks_stream).await?;
+
+		Ok(())
+	}
+
+	async fn revert_blocks_to(
+		&self,
+		height: <<Block as BlockT>::Header as Header>::Number,
+	) -> RpcResult<()> {
+		let best_number = self.client.info().best_number;
+		if height >= best_number {
+			return Err(JsonRpseeError::Custom(
+				"Target height is higher than current best height".into(),
+			))
+		}
+
+		let diff = best_number - height;
+
+		println!("Diff: {:?}", diff);
+
+		let reverted = self
+			.backend
+			.revert(diff, true)
+			.map_err(|e| JsonRpseeError::Custom(format!("Backend Revert Error: {}", e)))?;
+
+		println!("Reverted: {:?}", reverted);
+
+		Ok(())
+	}
 }
 
 /// report any errors or successes encountered by the authorship task back

node/src/rpc.rs

@@ -22,9 +22,11 @@ use sp_block_builder::BlockBuilder;
 use sp_blockchain::{Error as BlockChainError, HeaderBackend, HeaderMetadata};
 
 /// Full client dependencies.
-pub struct FullDeps<C, P> {
+pub struct FullDeps<C, B, P> {
 	/// The client instance to use.
 	pub client: Arc<C>,
+	/// The backend instance to use.
+	pub backend: Arc<B>,
 	/// Transaction pool instance.
 	pub pool: Arc<P>,
 	/// Whether to deny unsafe calls
@@ -34,8 +36,8 @@ pub struct FullDeps<C, P> {
 }
 
 /// Instantiate all full RPC extensions.
-pub fn create_full<C, P>(
-	deps: FullDeps<C, P>,
+pub fn create_full<C, B, P>(
+	deps: FullDeps<C, B, P>,
 ) -> Result<RpcModule<()>, Box<dyn std::error::Error + Send + Sync>>
 where
 	C: ProvideRuntimeApi<Block>,
@@ -44,19 +46,20 @@ where
 	C::Api: substrate_frame_rpc_system::AccountNonceApi<Block, AccountId, Index>,
 	C::Api: pallet_transaction_payment_rpc::TransactionPaymentRuntimeApi<Block, Balance>,
 	C::Api: BlockBuilder<Block>,
+	B: sc_client_api::backend::Backend<Block> + Send + Sync + 'static,
 	P: TransactionPool + 'static,
 {
 	use pallet_transaction_payment_rpc::{TransactionPayment, TransactionPaymentApiServer};
 	use substrate_frame_rpc_system::{System, SystemApiServer};
 
 	let mut io = RpcModule::new(());
-	let FullDeps { client, pool, deny_unsafe, command_sink } = deps;
+	let FullDeps { client, backend, pool, deny_unsafe, command_sink } = deps;
 
 	io.merge(System::new(client.clone(), pool.clone(), deny_unsafe).into_rpc())?;
 	io.merge(TransactionPayment::new(client.clone()).into_rpc())?;
 
 	// The final RPC extension receives commands for the manual seal consensus engine.
-	io.merge(ManualSeal::new(command_sink).into_rpc())?;
+	io.merge(ManualSeal::new(client, backend, command_sink).into_rpc())?;
 
 	Ok(io)
 }

node/src/service.rs

@@ -169,11 +169,13 @@ pub fn new_full(
 
 	let rpc_extensions_builder = {
 		let client = client.clone();
+		let backend = backend.clone();
 		let pool = transaction_pool.clone();
 
 		Box::new(move |deny_unsafe, _| {
 			let deps = crate::rpc::FullDeps {
 				client: client.clone(),
+				backend: backend.clone(),
 				pool: pool.clone(),
 				deny_unsafe,
 				command_sink: rpc_command_sink.clone(),