Add Subscription RPC for Grandpa Finality

bin/node/cli/src/service.rs

@@ -22,7 +22,12 @@ use std::sync::Arc;
 
 use sc_consensus_babe;
 use sc_client::{self, LongestChain};
-use grandpa::{self, FinalityProofProvider as GrandpaFinalityProofProvider, StorageAndProofProvider};
+use grandpa::{
+	self,
+	FinalityProofProvider as GrandpaFinalityProofProvider,
+	StorageAndProofProvider,
+	FinalityProofSubscription,
+};
 use node_executor;
 use node_primitives::Block;
 use node_runtime::RuntimeApi;
@@ -49,6 +54,7 @@ macro_rules! new_full_start {
 		type RpcExtension = jsonrpc_core::IoHandler<sc_rpc::Metadata>;
 		let mut import_setup = None;
 		let inherent_data_providers = sp_inherents::InherentDataProviders::new();
+		let finality_proof_subscription = grandpa::FinalityProofSubscription::new();
 
 		let builder = sc_service::ServiceBuilder::new_full::<
 			node_primitives::Block, node_runtime::RuntimeApi, node_executor::Executor
@@ -89,6 +95,10 @@ macro_rules! new_full_start {
 				Ok(import_queue)
 			})?
 			.with_rpc_extensions(|builder| -> std::result::Result<RpcExtension, _> {
+
+				// TODO: Do something with this stream
+				let justification_stream = finality_proof_subscription.stream();
+
 				let babe_link = import_setup.as_ref().map(|s| &s.2)
 					.expect("BabeLink is present for full services or set up failed; qed.");
 				let deps = node_rpc::FullDeps {
@@ -105,7 +115,7 @@ macro_rules! new_full_start {
 				Ok(node_rpc::create_full(deps))
 			})?;
 
-		(builder, import_setup, inherent_data_providers)
+		(builder, import_setup, inherent_data_providers, finality_tx)
 	}}
 }
 
@@ -131,7 +141,7 @@ macro_rules! new_full {
 			$config.disable_grandpa,
 		);
 
-		let (builder, mut import_setup, inherent_data_providers) = new_full_start!($config);
+		let (builder, mut import_setup, inherent_data_providers, finality_tx) = new_full_start!($config);
 
 		let service = builder
 			.with_finality_proof_provider(|client, backend| {
@@ -243,6 +253,7 @@ macro_rules! new_full {
 				telemetry_on_connect: Some(service.telemetry_on_connect_stream()),
 				voting_rule: grandpa::VotingRulesBuilder::default().build(),
 				prometheus_registry: service.prometheus_registry(),
+				finality_subscribers: Some(finality_proof_subscription),
 			};
 
 			// the GRANDPA voter task is considered infallible, i.e.

client/finality-grandpa/src/environment.rs

@@ -51,6 +51,7 @@ use sp_consensus::SelectChain;
 use crate::authorities::{AuthoritySet, SharedAuthoritySet};
 use crate::communication::Network as NetworkT;
 use crate::consensus_changes::SharedConsensusChanges;
+use crate::finality_proof::{SharedFinalityNotifiers, JustificationNotification};
 use crate::justification::GrandpaJustification;
 use crate::until_imported::UntilVoteTargetImported;
 use crate::voting_rule::VotingRule;
@@ -387,7 +388,6 @@ impl Metrics {
 	}
 }
 
-
 /// The environment we run GRANDPA in.
 pub(crate) struct Environment<Backend, Block: BlockT, C, N: NetworkT<Block>, SC, VR> {
 	pub(crate) client: Arc<C>,
@@ -401,6 +401,7 @@ pub(crate) struct Environment<Backend, Block: BlockT, C, N: NetworkT<Block>, SC,
 	pub(crate) voter_set_state: SharedVoterSetState<Block>,
 	pub(crate) voting_rule: VR,
 	pub(crate) metrics: Option<Metrics>,
+	pub(crate) finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
 	pub(crate) _phantom: PhantomData<Backend>,
 }
 
@@ -912,6 +913,7 @@ where
 			number,
 			(round, commit).into(),
 			false,
+			self.finality_notifiers.clone(),
 		)
 	}
 
@@ -972,6 +974,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 	number: NumberFor<Block>,
 	justification_or_commit: JustificationOrCommit<Block>,
 	initial_sync: bool,
+	finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
 ) -> Result<(), CommandOrError<Block::Hash, NumberFor<Block>>> where
 	Block:  BlockT,
 	BE: Backend<Block>,
@@ -983,6 +986,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 	let mut authority_set = authority_set.inner().write();
 
 	let status = client.info();
+
 	if number <= status.finalized_number && client.hash(number)? == Some(hash) {
 		// This can happen after a forced change (triggered by the finality tracker when finality is stalled), since
 		// the voter will be restarted at the median last finalized block, which can be lower than the local best
@@ -1077,6 +1081,24 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 			},
 		};
 
+		if let Some(notifiers) = finality_notifiers {
+			// Q: We `finalized()` this at L37, so can I be sure
+			// that it's fine to unwrap here?
+			if let Some(justification) = justification.clone() {
+				let header = client.header(BlockId::Hash(hash))?
+					.expect("");
+				let notification = JustificationNotification {
+					header,
+					justification,
+				};
+
+				for s in notifiers.read().iter() {
+					// TODO: Deal with Result
+					s.unbounded_send(notification.clone());
+				}
+			}
+		}
+
 		debug!(target: "afg", "Finalizing blocks up to ({:?}, {})", number, hash);
 
 		// ideally some handle to a synchronization oracle would be used

client/finality-grandpa/src/finality_proof.rs

@@ -36,6 +36,7 @@
 
 use std::sync::Arc;
 use log::{trace, warn};
+use parking_lot::RwLock;
 
 use sp_blockchain::{Backend as BlockchainBackend, Error as ClientError, Result as ClientResult};
 use sc_client_api::{
@@ -52,6 +53,7 @@ use sp_runtime::{
 use sp_core::storage::StorageKey;
 use sc_telemetry::{telemetry, CONSENSUS_INFO};
 use sp_finality_grandpa::{AuthorityId, AuthorityList, VersionedAuthorityList, GRANDPA_AUTHORITIES_KEY};
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender, TracingUnboundedReceiver};
 
 use crate::justification::GrandpaJustification;
 
@@ -145,6 +147,55 @@ impl<Block: BlockT> AuthoritySetForFinalityChecker<Block> for Arc<dyn FetchCheck
 	}
 }
 
+
+/// Justification for a finalized block.
+#[derive(Clone)]
+pub struct JustificationNotification<Block: BlockT> {
+	/// Highest finalized block header
+	pub header: Block::Header,
+	/// An encoded justification proving that the given header has been finalized
+	pub justification: Vec<u8>,
+}
+
+type JustificationStream<Block> = TracingUnboundedReceiver<JustificationNotification<Block>>;
+pub type SharedFinalitySubscribers<T> = Arc<RwLock<Vec<JustificationStream<T>>>>;
+
+type FinalityNotifier<T> = TracingUnboundedSender<JustificationNotification<T>>;
+pub type SharedFinalityNotifiers<T> = Arc<RwLock<Vec<FinalityNotifier<T>>>>;
+
+pub struct FinalityProofSubscription<Block: BlockT> {
+	notifiers: SharedFinalityNotifiers<Block>,
+}
+
+impl<Block: BlockT> FinalityProofSubscription<Block> {
+	pub fn new() -> Self {
+		Self {
+			notifiers: Arc::new(RwLock::new(vec![])),
+		}
+	}
+
+	pub fn notifiers(&self) -> SharedFinalityNotifiers<Block> {
+		self.notifiers.clone()
+	}
+
+	// Will notify subsribers (receivers)
+	pub fn notify(&self, notification: JustificationNotification<Block>) {
+		// TODO: Look at `notify_justified()` in `client`
+		for s in self.notifiers.read().iter() {
+			s.unbounded_send(notification.clone());
+		}
+
+		todo!()
+	}
+
+	pub fn stream(&self) -> JustificationStream<Block> {
+		// TODO: I only want one channel to be created per instance
+		let (sink, stream) = tracing_unbounded("mpsc_justification_notification_stream");
+		self.notifiers.write().push(sink);
+		stream
+	}
+}
+
 /// Finality proof provider for serving network requests.
 pub struct FinalityProofProvider<B,  Block: BlockT> {
 	backend: Arc<B>,

client/finality-grandpa/src/import.rs

@@ -37,11 +37,12 @@ use sp_runtime::traits::{
 	Block as BlockT, DigestFor, Header as HeaderT, NumberFor, Zero,
 };
 
-use crate::{Error, CommandOrError, NewAuthoritySet, VoterCommand};
+use crate::{Error, CommandOrError, NewAuthoritySet, VoterCommand, };
 use crate::authorities::{AuthoritySet, SharedAuthoritySet, DelayKind, PendingChange};
 use crate::consensus_changes::SharedConsensusChanges;
 use crate::environment::finalize_block;
 use crate::justification::GrandpaJustification;
+use crate::finality_proof::SharedFinalityNotifiers;
 use std::marker::PhantomData;
 
 /// A block-import handler for GRANDPA.
@@ -60,6 +61,7 @@ pub struct GrandpaBlockImport<Backend, Block: BlockT, Client, SC> {
 	send_voter_commands: TracingUnboundedSender<VoterCommand<Block::Hash, NumberFor<Block>>>,
 	consensus_changes: SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
 	authority_set_hard_forks: HashMap<Block::Hash, PendingChange<Block::Hash, NumberFor<Block>>>,
+	finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
 	_phantom: PhantomData<Backend>,
 }
 
@@ -74,6 +76,7 @@ impl<Backend, Block: BlockT, Client, SC: Clone> Clone for
 			send_voter_commands: self.send_voter_commands.clone(),
 			consensus_changes: self.consensus_changes.clone(),
 			authority_set_hard_forks: self.authority_set_hard_forks.clone(),
+			finality_notifiers: self.finality_notifiers.clone(),
 			_phantom: PhantomData,
 		}
 	}
@@ -558,6 +561,7 @@ impl<Backend, Block: BlockT, Client, SC> GrandpaBlockImport<Backend, Block, Clie
 		send_voter_commands: TracingUnboundedSender<VoterCommand<Block::Hash, NumberFor<Block>>>,
 		consensus_changes: SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
 		authority_set_hard_forks: Vec<(SetId, PendingChange<Block::Hash, NumberFor<Block>>)>,
+		finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
 	) -> GrandpaBlockImport<Backend, Block, Client, SC> {
 		// check for and apply any forced authority set hard fork that applies
 		// to the *current* authority set.
@@ -601,6 +605,7 @@ impl<Backend, Block: BlockT, Client, SC> GrandpaBlockImport<Backend, Block, Clie
 			send_voter_commands,
 			consensus_changes,
 			authority_set_hard_forks,
+			finality_notifiers,
 			_phantom: PhantomData,
 		}
 	}
@@ -646,6 +651,7 @@ where
 			number,
 			justification.into(),
 			initial_sync,
+			self.finality_notifiers.clone(),
 		);
 
 		match result {

client/finality-grandpa/src/lib.rs

@@ -69,7 +69,7 @@ use sc_keystore::KeyStorePtr;
 use sp_inherents::InherentDataProviders;
 use sp_consensus::{SelectChain, BlockImport};
 use sp_core::Pair;
-use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver};
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedReceiver, TracingUnboundedSender};
 use sc_telemetry::{telemetry, CONSENSUS_INFO, CONSENSUS_DEBUG};
 use serde_json;
 
@@ -127,6 +127,7 @@ use import::GrandpaBlockImport;
 use until_imported::UntilGlobalMessageBlocksImported;
 use communication::{NetworkBridge, Network as NetworkT};
 use sp_finality_grandpa::{AuthorityList, AuthorityPair, AuthoritySignature, SetId};
+use crate::finality_proof::{FinalityProofSubscription, SharedFinalityNotifiers};
 
 // Re-export these two because it's just so damn convenient.
 pub use sp_finality_grandpa::{AuthorityId, ScheduledChange};
@@ -521,6 +522,7 @@ where
 			voter_commands_tx,
 			persistent_data.consensus_changes.clone(),
 			authority_set_hard_forks,
+			None, //finality_notifiers.clone(), // TODO: Figure out how to get subs here
 		),
 		LinkHalf {
 			client,
@@ -620,6 +622,8 @@ pub struct GrandpaParams<Block: BlockT, C, N, SC, VR> {
 	pub voting_rule: VR,
 	/// The prometheus metrics registry.
 	pub prometheus_registry: Option<prometheus_endpoint::Registry>,
+	/// A subscription to new block justifications
+	pub finality_subscription: Option<FinalityProofSubscription<Block>>,
 }
 
 /// Run a GRANDPA voter as a task. Provide configuration and a link to a
@@ -644,6 +648,7 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		telemetry_on_connect,
 		voting_rule,
 		prometheus_registry,
+		finality_subscription,
 	} = grandpa_params;
 
 	// NOTE: we have recently removed `run_grandpa_observer` from the public
@@ -695,6 +700,12 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		future::Either::Right(future::pending())
 	};
 
+	let finality_notifiers = if let Some(s) = finality_subscription {
+		Some(s.notifiers())
+	} else {
+		None
+	};
+
 	let voter_work = VoterWork::new(
 		client,
 		config,
@@ -704,6 +715,7 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		persistent_data,
 		voter_commands_rx,
 		prometheus_registry,
+		finality_notifiers,
 	);
 
 	let voter_work = voter_work
@@ -761,6 +773,7 @@ where
 		persistent_data: PersistentData<Block>,
 		voter_commands_rx: TracingUnboundedReceiver<VoterCommand<Block::Hash, NumberFor<Block>>>,
 		prometheus_registry: Option<prometheus_endpoint::Registry>,
+		finality_notifiers: Option<SharedFinalityNotifiers<Block>>,
 	) -> Self {
 		let metrics = match prometheus_registry.as_ref().map(Metrics::register) {
 			Some(Ok(metrics)) => Some(metrics),
@@ -784,6 +797,7 @@ where
 			consensus_changes: persistent_data.consensus_changes.clone(),
 			voter_set_state: persistent_data.set_state.clone(),
 			metrics: metrics.as_ref().map(|m| m.environment.clone()),
+			finality_notifiers,
 			_phantom: PhantomData,
 		});
 
@@ -907,6 +921,7 @@ where
 					network: self.env.network.clone(),
 					voting_rule: self.env.voting_rule.clone(),
 					metrics: self.env.metrics.clone(),
+					finality_notifiers: self.env.finality_notifiers.clone(),
 					_phantom: PhantomData,
 				});
 

client/finality-grandpa/src/observer.rs

@@ -125,6 +125,7 @@ fn grandpa_observer<BE, Block: BlockT, Client, S, F>(
 				finalized_number,
 				(round, commit).into(),
 				false,
+				None, // TODO: Should I include the finality_subscribers here?
 			) {
 				Ok(_) => {},
 				Err(e) => return future::err(e),