Add Subscription RPC for Grandpa Finality (#5732)

* Rough skeleton for what I think the RPC should look like

* Create channel for sending justifications

Sends finalized header and justification from Grandpa to the
client. This lays the groundwork for hooking into the RPC module.

* WIP: Add subscribers for justifications to Grandpa

Adds the Sender end of a channel into Grandpa, through which notifications
about block finality events can be sent.

* WIP: Add a struct for managing subscriptions

Slightly different approach from the last commit, but same
basic idea. Still a rough sketch, very much doesn't compile yet.

* Make naming more clear and lock data in Arc

* Rough idea of what RPC would look like

* Remove code from previous approach

* Missed some things

* Update client/rpc-api/src/chain/mod.rs

Co-Authored-By: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* Update client/rpc-api/src/chain/mod.rs

Co-Authored-By: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

* Split justification subscription into sender and receiver halves

* Replace RwLock with a Mutex

* Add sample usage from the Service's point of view

* Remove code that referred to "chain_" RPC

* Use the Justification sender/receivers from Grandpa LinkHalf

* Add some PubSub boilerplate

* Add guiding comments

* TMP: comment out to fix compilation

* Return MetaIoHandler from PubSubHandler in create_full

* Uncomment pubsub methods in rpc handler (fails to build)

* node/rpc: make Metadata concrete in create_full to fix compilation

* node: pass in SubscriptionManger to grandpa rpc handler

* grandpa-rpc: use SubscriptionManger to add subscriber

* grandpa-rpc: attempt at setting up the justification stream (fails to build)

* grandpa-rpc: fix compilation of connecting stream to sink

* grandpa-rpc: implement unsubscribe

* grandpa-rpc: update older tests

* grandpa-rpc: add full prefix to avoid confusing rust-analyzer

* grandpa-rpc: add test for pubsub not available

* grandpa-rpc: tidy up leftover code

* grandpa-rpc: add test for sub and unsub of justifications

* grandpa-rpc: minor stylistic changes

* grandpa-rpc: split unit test

* grandpa-rpc: minor stylistic changes in test

* grandpa-rpc: skip returning future when cancelling

* grandpa-rpc: reuse testing executor from sc-rpc

* grandpa-rpc: don't need to use PubSubHandler in tests

* node-rpc: use MetaIoHandler rather than PubSubHandler

* grandpa: log if getting header failed

* grandpa: move justification channel creation into factory function

* grandpa: make the justification sender optional

* grandpa: fix compilation warnings

* grandpa: move justification notification types to new file

* grandpa-rpc: move JustificationNotification to grandpa-rpc

* grandpa-rpc: move JustificationNotification to its own file

* grandpa: rename justification channel pairs

* grandpa: rename notifier types

* grandpa: pass justification as GrandpaJustification to the rpc module

* Move Metadata to sc-rpc-api

* grandpa-rpc: remove unsed error code

* grandpa: fix bug for checking if channel is closed before sendind

* grandpa-rpc: unit test for sending justifications

* grandpa-rpc: update comments for the pubsub test

* grandpa-rpc: update pubsub tests with more steps

* grandpa-rpc: fix pubsub test

* grandpa-rpc: minor indendation

* grandpa-rpc: decode instead of encode in test

* grandpa: fix review comments

* grandpa: remove unused serde dependency

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>
Co-authored-by: Jon Häggblad <jon.haggblad@gmail.com>
Co-authored-by: Tomasz Drwięga <tomasz@parity.io>

bin/node-template/node/src/service.rs

@@ -113,7 +113,7 @@ pub fn new_full(config: Configuration) -> Result<TaskManager, ServiceError> {
 		let client = client.clone();
 		let pool = transaction_pool.clone();
 
-		Box::new(move |deny_unsafe| {
+		Box::new(move |deny_unsafe, _| {
 			let deps = crate::rpc::FullDeps {
 				client: client.clone(),
 				pool: pool.clone(),
@@ -278,7 +278,7 @@ pub fn new_light(config: Configuration) -> Result<TaskManager, ServiceError> {
 		transaction_pool,
 		task_manager: &mut task_manager,
 		on_demand: Some(on_demand),
-		rpc_extensions_builder: Box::new(|_| ()),
+		rpc_extensions_builder: Box::new(|_, _| ()),
 		telemetry_connection_sinks: sc_service::TelemetryConnectionSinks::default(),
 		config,
 		client,

bin/node/cli/Cargo.toml

@@ -39,6 +39,7 @@ serde = { version = "1.0.102", features = ["derive"] }
 futures = { version = "0.3.1", features = ["compat"] }
 hex-literal = "0.2.1"
 jsonrpc-core = "14.2.0"
+jsonrpc-pubsub = "14.2.0"
 log = "0.4.8"
 rand = "0.7.2"
 structopt = { version = "0.3.8", optional = true }

bin/node/cli/src/service.rs

@@ -49,7 +49,10 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 	sp_consensus::DefaultImportQueue<Block, FullClient>,
 	sc_transaction_pool::FullPool<Block, FullClient>,
 	(
-		impl Fn(node_rpc::DenyUnsafe) -> node_rpc::IoHandler,
+		impl Fn(
+			node_rpc::DenyUnsafe,
+			jsonrpc_pubsub::manager::SubscriptionManager
+		) -> node_rpc::IoHandler,
 		(
 			sc_consensus_babe::BabeBlockImport<Block, FullClient, FullGrandpaBlockImport>,
 			grandpa::LinkHalf<Block, FullClient, FullSelectChain>,
@@ -101,6 +104,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 	let (rpc_extensions_builder, rpc_setup) = {
 		let (_, grandpa_link, babe_link) = &import_setup;
 
+		let justification_stream = grandpa_link.justification_stream();
 		let shared_authority_set = grandpa_link.shared_authority_set().clone();
 		let shared_voter_state = grandpa::SharedVoterState::empty();
 
@@ -114,7 +118,7 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 		let select_chain = select_chain.clone();
 		let keystore = keystore.clone();
 
-		let rpc_extensions_builder = move |deny_unsafe| {
+		let rpc_extensions_builder = move |deny_unsafe, subscriptions| {
 			let deps = node_rpc::FullDeps {
 				client: client.clone(),
 				pool: pool.clone(),
@@ -128,6 +132,8 @@ pub fn new_partial(config: &Configuration) -> Result<sc_service::PartialComponen
 				grandpa: node_rpc::GrandpaDeps {
 					shared_voter_state: shared_voter_state.clone(),
 					shared_authority_set: shared_authority_set.clone(),
+					justification_stream: justification_stream.clone(),
+					subscriptions,
 				},
 			};
 

bin/node/rpc/Cargo.toml

@@ -11,25 +11,25 @@ repository = "https://github.com/paritytech/substrate/"
 targets = ["x86_64-unknown-linux-gnu"]
 
 [dependencies]
-sc-client-api = { version = "2.0.0-rc5", path = "../../../client/api" }
-sc-rpc = { version = "2.0.0-rc5", path = "../../../client/rpc" }
 jsonrpc-core = "14.2.0"
+jsonrpc-pubsub = "14.2.0"
 node-primitives = { version = "2.0.0-rc5", path = "../primitives" }
 node-runtime = { version = "2.0.0-rc5", path = "../runtime" }
-sp-runtime = { version = "2.0.0-rc5", path = "../../../primitives/runtime" }
-sp-api = { version = "2.0.0-rc5", path = "../../../primitives/api" }
 pallet-contracts-rpc = { version = "0.8.0-rc5", path = "../../../frame/contracts/rpc/" }
 pallet-transaction-payment-rpc = { version = "2.0.0-rc5", path = "../../../frame/transaction-payment/rpc/" }
-substrate-frame-rpc-system = { version = "2.0.0-rc5", path = "../../../utils/frame/rpc/system" }
-sp-transaction-pool = { version = "2.0.0-rc5", path = "../../../primitives/transaction-pool" }
+sc-client-api = { version = "2.0.0-rc5", path = "../../../client/api" }
 sc-consensus-babe = { version = "0.8.0-rc5", path = "../../../client/consensus/babe" }
 sc-consensus-babe-rpc = { version = "0.8.0-rc5", path = "../../../client/consensus/babe/rpc" }
-sp-consensus-babe = { version = "0.8.0-rc5", path = "../../../primitives/consensus/babe" }
-sc-keystore = { version = "2.0.0-rc5", path = "../../../client/keystore" }
 sc-consensus-epochs = { version = "0.8.0-rc5", path = "../../../client/consensus/epochs" }
-sp-consensus = { version = "0.8.0-rc5", path = "../../../primitives/consensus/common" }
-sp-blockchain = { version = "2.0.0-rc5", path = "../../../primitives/blockchain" }
 sc-finality-grandpa = { version = "0.8.0-rc5", path = "../../../client/finality-grandpa" }
 sc-finality-grandpa-rpc = { version = "0.8.0-rc5", path = "../../../client/finality-grandpa/rpc" }
+sc-keystore = { version = "2.0.0-rc5", path = "../../../client/keystore" }
 sc-rpc-api = { version = "0.8.0-rc5", path = "../../../client/rpc-api" }
+sc-rpc = { version = "2.0.0-rc5", path = "../../../client/rpc" }
+sp-api = { version = "2.0.0-rc5", path = "../../../primitives/api" }
 sp-block-builder = { version = "2.0.0-rc5", path = "../../../primitives/block-builder" }
+sp-blockchain = { version = "2.0.0-rc5", path = "../../../primitives/blockchain" }
+sp-consensus = { version = "0.8.0-rc5", path = "../../../primitives/consensus/common" }
+sp-consensus-babe = { version = "0.8.0-rc5", path = "../../../primitives/consensus/babe" }
+sp-transaction-pool = { version = "2.0.0-rc5", path = "../../../primitives/transaction-pool" }
+substrate-frame-rpc-system = { version = "2.0.0-rc5", path = "../../../utils/frame/rpc/system" }

bin/node/rpc/src/lib.rs

@@ -32,20 +32,21 @@
 
 use std::sync::Arc;
 
+use jsonrpc_pubsub::manager::SubscriptionManager;
 use node_primitives::{Block, BlockNumber, AccountId, Index, Balance, Hash};
-use sp_api::ProvideRuntimeApi;
-use sp_transaction_pool::TransactionPool;
-use sp_blockchain::{Error as BlockChainError, HeaderMetadata, HeaderBackend};
-use sp_consensus::SelectChain;
-use sc_keystore::KeyStorePtr;
-use sp_consensus_babe::BabeApi;
-use sc_consensus_epochs::SharedEpochChanges;
 use sc_consensus_babe::{Config, Epoch};
 use sc_consensus_babe_rpc::BabeRpcHandler;
-use sc_finality_grandpa::{SharedVoterState, SharedAuthoritySet};
+use sc_consensus_epochs::SharedEpochChanges;
+use sc_finality_grandpa::{SharedVoterState, SharedAuthoritySet, GrandpaJustificationStream};
 use sc_finality_grandpa_rpc::GrandpaRpcHandler;
-use sp_block_builder::BlockBuilder;
+use sc_keystore::KeyStorePtr;
 pub use sc_rpc_api::DenyUnsafe;
+use sp_api::ProvideRuntimeApi;
+use sp_block_builder::BlockBuilder;
+use sp_blockchain::{Error as BlockChainError, HeaderMetadata, HeaderBackend};
+use sp_consensus::SelectChain;
+use sp_consensus_babe::BabeApi;
+use sp_transaction_pool::TransactionPool;
 
 /// Light client extra dependencies.
 pub struct LightDeps<C, F, P> {
@@ -75,6 +76,10 @@ pub struct GrandpaDeps {
 	pub shared_voter_state: SharedVoterState,
 	/// Authority set info.
 	pub shared_authority_set: SharedAuthoritySet<Hash, BlockNumber>,
+	/// Receives notifications about justification events from Grandpa.
+	pub justification_stream: GrandpaJustificationStream<Block>,
+	/// Subscription manager to keep track of pubsub subscribers.
+	pub subscriptions: SubscriptionManager,
 }
 
 /// Full client dependencies.
@@ -97,9 +102,9 @@ pub struct FullDeps<C, P, SC> {
 pub type IoHandler = jsonrpc_core::IoHandler<sc_rpc::Metadata>;
 
 /// Instantiate all Full RPC extensions.
-pub fn create_full<C, P, M, SC>(
+pub fn create_full<C, P, SC>(
 	deps: FullDeps<C, P, SC>,
-) -> jsonrpc_core::IoHandler<M> where
+) -> jsonrpc_core::IoHandler<sc_rpc_api::Metadata> where
 	C: ProvideRuntimeApi<Block>,
 	C: HeaderBackend<Block> + HeaderMetadata<Block, Error=BlockChainError> + 'static,
 	C: Send + Sync + 'static,
@@ -109,7 +114,6 @@ pub fn create_full<C, P, M, SC>(
 	C::Api: BabeApi<Block>,
 	C::Api: BlockBuilder<Block>,
 	P: TransactionPool + 'static,
-	M: jsonrpc_core::Metadata + Default,
 	SC: SelectChain<Block> +'static,
 {
 	use substrate_frame_rpc_system::{FullSystem, SystemApi};
@@ -125,6 +129,7 @@ pub fn create_full<C, P, M, SC>(
 		babe,
 		grandpa,
 	} = deps;
+
 	let BabeDeps {
 		keystore,
 		babe_config,
@@ -133,6 +138,8 @@ pub fn create_full<C, P, M, SC>(
 	let GrandpaDeps {
 		shared_voter_state,
 		shared_authority_set,
+		justification_stream,
+		subscriptions,
 	} = grandpa;
 
 	io.extend_with(
@@ -161,7 +168,12 @@ pub fn create_full<C, P, M, SC>(
 	);
 	io.extend_with(
 		sc_finality_grandpa_rpc::GrandpaApi::to_delegate(
-			GrandpaRpcHandler::new(shared_authority_set, shared_voter_state)
+			GrandpaRpcHandler::new(
+				shared_authority_set,
+				shared_voter_state,
+				justification_stream,
+				subscriptions,
+			)
 		)
 	);
 

client/finality-grandpa/rpc/Cargo.toml

@@ -8,16 +8,29 @@ edition = "2018"
 license = "GPL-3.0-or-later WITH Classpath-exception-2.0"
 
 [dependencies]
+sc-rpc = { version = "2.0.0-rc5", path = "../../rpc" }
+sp-runtime = { version = "2.0.0-rc5", path = "../../../primitives/runtime" }
 sc-finality-grandpa = { version = "0.8.0-rc5", path = "../" }
 finality-grandpa = { version = "0.12.3", features = ["derive-codec"] }
 jsonrpc-core = "14.2.0"
 jsonrpc-core-client = "14.2.0"
 jsonrpc-derive = "14.2.1"
+jsonrpc-pubsub = "14.2.0"
 futures = { version = "0.3.4", features = ["compat"] }
 serde = { version = "1.0.105", features = ["derive"] }
 serde_json = "1.0.50"
 log = "0.4.8"
 derive_more = "0.99.2"
+parity-scale-codec = { version = "1.3.0", features = ["derive"] }
 
 [dev-dependencies]
+sc-block-builder = { version = "0.8.0-rc5", path = "../../block-builder" }
+sc-network-test = { version = "0.8.0-rc5", path = "../../network/test" }
+sc-rpc = { version = "2.0.0-rc5", path = "../../rpc", features = ["test-helpers"] }
+sp-blockchain = { version = "2.0.0-rc5", path = "../../../primitives/blockchain" }
+sp-consensus = { version = "0.8.0-rc5", path = "../../../primitives/consensus/common" }
 sp-core = { version = "2.0.0-rc5", path = "../../../primitives/core" }
+sp-finality-grandpa = { version = "2.0.0-rc5", path = "../../../primitives/finality-grandpa" }
+sp-keyring = { version = "2.0.0-rc5", path = "../../../primitives/keyring" }
+substrate-test-runtime-client = { version = "2.0.0-rc5",  path = "../../../test-utils/runtime/client" }
+lazy_static = "1.4"