rpc: Use the blocks pinning API for chainHead methods

client/api/src/in_mem.rs

@@ -440,7 +440,7 @@ impl<Block: BlockT> blockchain::Backend<Block> for Blockchain<Block> {
 	}
 
 	fn children(&self, _parent_hash: Block::Hash) -> sp_blockchain::Result<Vec<Block::Hash>> {
-		unimplemented!()
+		Ok(Default::default())
 	}
 
 	fn indexed_transaction(&self, _hash: Block::Hash) -> sp_blockchain::Result<Option<Vec<u8>>> {
@@ -626,20 +626,36 @@ where
 	states: RwLock<HashMap<Block::Hash, InMemoryBackend<HashFor<Block>>>>,
 	blockchain: Blockchain<Block>,
 	import_lock: RwLock<()>,
+	pinned_blocks: RwLock<HashMap<Block::Hash, i64>>,
 }
 
 impl<Block: BlockT> Backend<Block>
 where
 	Block::Hash: Ord,
 {
 	/// Create a new instance of in-mem backend.
+	///
+	/// # Warning
+	///
+	/// For testing purposes only!
 	pub fn new() -> Self {
 		Backend {
 			states: RwLock::new(HashMap::new()),
 			blockchain: Blockchain::new(),
 			import_lock: Default::default(),
+			pinned_blocks: Default::default(),
 		}
 	}
+
+	/// Return the number of references active for a pinned block.
+	///
+	/// # Warning
+	///
+	/// For testing purposes only!
+	pub fn pin_refs(&self, hash: &<Block as BlockT>::Hash) -> Option<i64> {
+		let blocks = self.pinned_blocks.read();
+		blocks.get(hash).map(|value| *value)
+	}
 }
 
 impl<Block: BlockT> backend::AuxStore for Backend<Block>
@@ -789,11 +805,17 @@ where
 		false
 	}
 
-	fn pin_block(&self, _: <Block as BlockT>::Hash) -> blockchain::Result<()> {
+	fn pin_block(&self, hash: <Block as BlockT>::Hash) -> blockchain::Result<()> {
+		let mut blocks = self.pinned_blocks.write();
+		blocks.entry(hash).and_modify(|counter| *counter += 1).or_insert(1);
+
 		Ok(())
 	}
 
-	fn unpin_block(&self, _: <Block as BlockT>::Hash) {}
+	fn unpin_block(&self, hash: <Block as BlockT>::Hash) {
+		let mut blocks = self.pinned_blocks.write();
+		blocks.entry(hash).and_modify(|counter| *counter -= 1).or_insert(-1);
+	}
 }
 
 impl<Block: BlockT> backend::LocalBackend<Block> for Backend<Block> where Block::Hash: Ord {}

client/rpc-spec-v2/Cargo.toml

@@ -43,4 +43,5 @@ substrate-test-runtime = { version = "2.0.0", path = "../../test-utils/runtime"
 sp-consensus = { version = "0.10.0-dev", path = "../../primitives/consensus/common" }
 sp-maybe-compressed-blob = { version = "4.1.0-dev", path = "../../primitives/maybe-compressed-blob" }
 sc-block-builder = { version = "0.10.0-dev", path = "../block-builder" }
+sc-service = { version = "0.10.0-dev", features = ["test-helpers"], path = "../service" }
 assert_matches = "1.3.0"

client/rpc-spec-v2/src/chain_head/chain_head.rs

@@ -60,15 +60,15 @@ use sp_runtime::{
 use std::{marker::PhantomData, sync::Arc};
 
 /// An API for chain head RPC calls.
-pub struct ChainHead<BE, Block: BlockT, Client> {
+pub struct ChainHead<BE: Backend<Block> + 'static, Block: BlockT, Client> {
 	/// Substrate client.
 	client: Arc<Client>,
 	/// Backend of the chain.
 	backend: Arc<BE>,
 	/// Executor to spawn subscriptions.
 	executor: SubscriptionTaskExecutor,
 	/// Keep track of the pinned blocks for each subscription.
-	subscriptions: Arc<SubscriptionManagement<Block>>,
+	subscriptions: Arc<SubscriptionManagement<Block, BE>>,
 	/// The hexadecimal encoded hash of the genesis block.
 	genesis_hash: String,
 	/// The maximum number of pinned blocks allowed per connection.
@@ -77,7 +77,7 @@ pub struct ChainHead<BE, Block: BlockT, Client> {
 	_phantom: PhantomData<Block>,
 }
 
-impl<BE, Block: BlockT, Client> ChainHead<BE, Block, Client> {
+impl<BE: Backend<Block> + 'static, Block: BlockT, Client> ChainHead<BE, Block, Client> {
 	/// Create a new [`ChainHead`].
 	pub fn new<GenesisHash: AsRef<[u8]>>(
 		client: Arc<Client>,
@@ -129,7 +129,7 @@ impl<BE, Block: BlockT, Client> ChainHead<BE, Block, Client> {
 fn generate_initial_events<Block, BE, Client>(
 	client: &Arc<Client>,
 	backend: &Arc<BE>,
-	handle: &SubscriptionHandle<Block>,
+	handle: &SubscriptionHandle<Block, BE>,
 	runtime_updates: bool,
 ) -> Result<Vec<FollowEvent<Block::Hash>>, SubscriptionManagementError>
 where
@@ -314,13 +314,14 @@ async fn submit_events<EventStream, T>(
 
 /// Generate the "NewBlock" event and potentially the "BestBlockChanged" event for
 /// every notification.
-fn handle_import_blocks<Client, Block>(
+fn handle_import_blocks<BE, Client, Block>(
 	client: &Arc<Client>,
-	handle: &SubscriptionHandle<Block>,
+	handle: &SubscriptionHandle<Block, BE>,
 	runtime_updates: bool,
 	notification: BlockImportNotification<Block>,
 ) -> Result<(FollowEvent<Block::Hash>, Option<FollowEvent<Block::Hash>>), SubscriptionManagementError>
 where
+	BE: Backend<Block> + 'static,
 	Block: BlockT + 'static,
 	Client: CallApiAt<Block> + 'static,
 {
@@ -370,12 +371,13 @@ where
 
 /// Generate the "Finalized" event and potentially the "BestBlockChanged" for
 /// every notification.
-fn handle_finalized_blocks<Client, Block>(
+fn handle_finalized_blocks<BE, Client, Block>(
 	client: &Arc<Client>,
-	handle: &SubscriptionHandle<Block>,
+	handle: &SubscriptionHandle<Block, BE>,
 	notification: FinalityNotification<Block>,
 ) -> Result<(FollowEvent<Block::Hash>, Option<FollowEvent<Block::Hash>>), SubscriptionManagementError>
 where
+	BE: Backend<Block> + 'static,
 	Block: BlockT + 'static,
 	Client: HeaderBackend<Block> + HeaderMetadata<Block, Error = BlockChainError> + 'static,
 {
@@ -474,9 +476,8 @@ where
 				return Err(err)
 			},
 		};
-
 		// Keep track of the subscription.
-		let Some((rx_stop, sub_handle)) = self.subscriptions.insert_subscription(sub_id.clone(), runtime_updates, self.max_pinned_blocks) else {
+		let Some((rx_stop, sub_handle)) = self.subscriptions.insert_subscription(sub_id.clone(), runtime_updates, self.max_pinned_blocks, self.backend.clone()) else {
 			// Inserting the subscription can only fail if the JsonRPSee
 			// generated a duplicate subscription ID.
 			debug!(target: "rpc-spec-v2", "[follow][id={:?}] Subscription already accepted", sub_id);