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>
2026-06-14 15:41:02 +00:00 · 2020-08-10 06:31:36 -04:00
parent a936771c0f
commit 433b7214f5
28 changed files with 589 additions and 94 deletions
@@ -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::notification::GrandpaJustificationSender;
 use crate::justification::GrandpaJustification;
 use crate::until_imported::UntilVoteTargetImported;
 use crate::voting_rule::VotingRule;
@@ -390,7 +391,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>,
@@ -404,6 +404,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) justification_sender: Option<GrandpaJustificationSender<Block>>,
 	pub(crate) _phantom: PhantomData<Backend>,
 }

@@ -1022,6 +1023,7 @@ where
 			number,
 			(round, commit).into(),
 			false,
+			&self.justification_sender,
 		)
 	}

@@ -1086,6 +1088,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 	number: NumberFor<Block>,
 	justification_or_commit: JustificationOrCommit<Block>,
 	initial_sync: bool,
+	justification_sender: &Option<GrandpaJustificationSender<Block>>,
 ) -> Result<(), CommandOrError<Block::Hash, NumberFor<Block>>> where
 	Block:  BlockT,
 	BE: Backend<Block>,
@@ -1097,6 +1100,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
@@ -1157,7 +1161,7 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 		// justifications for transition blocks which will be requested by
 		// syncing clients.
 		let justification = match justification_or_commit {
-			JustificationOrCommit::Justification(justification) => Some(justification.encode()),
+			JustificationOrCommit::Justification(justification) => Some(justification),
 			JustificationOrCommit::Commit((round_number, commit)) => {
 				let mut justification_required =
 					// justification is always required when block that enacts new authorities
@@ -1184,13 +1188,22 @@ pub(crate) fn finalize_block<BE, Block, Client>(
 						commit,
 					)?;

-					Some(justification.encode())
+					Some(justification)
 				} else {
 					None
 				}
 			},
 		};

+		// Notify any registered listeners in case we have a justification
+		if let Some(sender) = justification_sender {
+			if let Some(ref justification) = justification {
+				let _ = sender.notify(justification.clone());
+			}
+		}
+
+		let justification = justification.map(|j| j.encode());
+
 		debug!(target: "afg", "Finalizing blocks up to ({:?}, {})", number, hash);

 		// ideally some handle to a synchronization oracle would be used
@@ -149,7 +149,7 @@ impl<Block: BlockT> AuthoritySetForFinalityChecker<Block> for Arc<dyn FetchCheck
 }

 /// Finality proof provider for serving network requests.
-pub struct FinalityProofProvider<B,  Block: BlockT> {
+pub struct FinalityProofProvider<B, Block: BlockT> {
 	backend: Arc<B>,
 	authority_provider: Arc<dyn AuthoritySetForFinalityProver<Block>>,
 }
@@ -44,6 +44,7 @@ use crate::authorities::{AuthoritySet, SharedAuthoritySet, DelayKind, PendingCha
 use crate::consensus_changes::SharedConsensusChanges;
 use crate::environment::finalize_block;
 use crate::justification::GrandpaJustification;
+use crate::notification::GrandpaJustificationSender;
 use std::marker::PhantomData;

 /// A block-import handler for GRANDPA.
@@ -62,6 +63,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>>>,
+	justification_sender: GrandpaJustificationSender<Block>,
 	_phantom: PhantomData<Backend>,
 }

@@ -76,6 +78,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(),
+			justification_sender: self.justification_sender.clone(),
 			_phantom: PhantomData,
 		}
 	}
@@ -560,6 +563,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>>)>,
+		justification_sender: GrandpaJustificationSender<Block>,
 	) -> GrandpaBlockImport<Backend, Block, Client, SC> {
 		// check for and apply any forced authority set hard fork that applies
 		// to the *current* authority set.
@@ -603,6 +607,7 @@ impl<Backend, Block: BlockT, Client, SC> GrandpaBlockImport<Backend, Block, Clie
 			send_voter_commands,
 			consensus_changes,
 			authority_set_hard_forks,
+			justification_sender,
 			_phantom: PhantomData,
 		}
 	}
@@ -648,6 +653,7 @@ where
 			number,
 			justification.into(),
 			initial_sync,
+			&Some(self.justification_sender.clone()),
 		);

 		match result {
@@ -37,7 +37,7 @@ use crate::{Commit, Error};
 ///
 /// This is meant to be stored in the db and passed around the network to other
 /// nodes, and are used by syncing nodes to prove authority set handoffs.
-#[derive(Encode, Decode)]
+#[derive(Clone, Encode, Decode, PartialEq, Eq, Debug)]
 pub struct GrandpaJustification<Block: BlockT> {
 	round: u64,
 	pub(crate) commit: Commit<Block>,
@@ -47,7 +47,7 @@ pub struct GrandpaJustification<Block: BlockT> {
 impl<Block: BlockT> GrandpaJustification<Block> {
 	/// Create a GRANDPA justification from the given commit. This method
 	/// assumes the commit is valid and well-formed.
-	pub(crate) fn from_commit<C>(
+	pub fn from_commit<C>(
 		client: &Arc<C>,
 		round: u64,
 		commit: Commit<Block>,
@@ -119,12 +119,14 @@ mod finality_proof;
 mod import;
 mod justification;
 mod light_import;
+mod notification;
 mod observer;
 mod until_imported;
 mod voting_rule;

 pub use authorities::SharedAuthoritySet;
 pub use finality_proof::{FinalityProofProvider, StorageAndProofProvider};
+pub use notification::{GrandpaJustificationSender, GrandpaJustificationStream};
 pub use import::GrandpaBlockImport;
 pub use justification::GrandpaJustification;
 pub use light_import::{light_block_import, GrandpaLightBlockImport};
@@ -448,6 +450,8 @@ pub struct LinkHalf<Block: BlockT, C, SC> {
 	select_chain: SC,
 	persistent_data: PersistentData<Block>,
 	voter_commands_rx: TracingUnboundedReceiver<VoterCommand<Block::Hash, NumberFor<Block>>>,
+	justification_sender: GrandpaJustificationSender<Block>,
+	justification_stream: GrandpaJustificationStream<Block>,
 }

 impl<Block: BlockT, C, SC> LinkHalf<Block, C, SC> {
@@ -455,6 +459,11 @@ impl<Block: BlockT, C, SC> LinkHalf<Block, C, SC> {
 	pub fn shared_authority_set(&self) -> &SharedAuthoritySet<Block::Hash, NumberFor<Block>> {
 		&self.persistent_data.authority_set
 	}
+
+	/// Get the receiving end of justification notifications.
+	pub fn justification_stream(&self) -> GrandpaJustificationStream<Block> {
+		self.justification_stream.clone()
+	}
 }

 /// Provider for the Grandpa authority set configured on the genesis block.
@@ -553,6 +562,9 @@ where

 	let (voter_commands_tx, voter_commands_rx) = tracing_unbounded("mpsc_grandpa_voter_command");

+	let (justification_sender, justification_stream) =
+		GrandpaJustificationStream::channel();
+
 	// create pending change objects with 0 delay and enacted on finality
 	// (i.e. standard changes) for each authority set hard fork.
 	let authority_set_hard_forks = authority_set_hard_forks
@@ -579,12 +591,15 @@ where
 			voter_commands_tx,
 			persistent_data.consensus_changes.clone(),
 			authority_set_hard_forks,
+			justification_sender.clone(),
 		),
 		LinkHalf {
 			client,
 			select_chain,
 			persistent_data,
 			voter_commands_rx,
+			justification_sender,
+			justification_stream,
 		},
 	))
 }
@@ -719,6 +734,8 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		select_chain,
 		persistent_data,
 		voter_commands_rx,
+		justification_sender,
+		justification_stream: _,
 	} = link;

 	let network = NetworkBridge::new(
@@ -767,6 +784,7 @@ pub fn run_grandpa_voter<Block: BlockT, BE: 'static, C, N, SC, VR>(
 		voter_commands_rx,
 		prometheus_registry,
 		shared_voter_state,
+		justification_sender,
 	);

 	let voter_work = voter_work
@@ -827,6 +845,7 @@ where
 		voter_commands_rx: TracingUnboundedReceiver<VoterCommand<Block::Hash, NumberFor<Block>>>,
 		prometheus_registry: Option<prometheus_endpoint::Registry>,
 		shared_voter_state: SharedVoterState,
+		justification_sender: GrandpaJustificationSender<Block>,
 	) -> Self {
 		let metrics = match prometheus_registry.as_ref().map(Metrics::register) {
 			Some(Ok(metrics)) => Some(metrics),
@@ -850,6 +869,7 @@ where
 			consensus_changes: persistent_data.consensus_changes.clone(),
 			voter_set_state: persistent_data.set_state,
 			metrics: metrics.as_ref().map(|m| m.environment.clone()),
+			justification_sender: Some(justification_sender),
 			_phantom: PhantomData,
 		});

@@ -988,6 +1008,7 @@ where
 					network: self.env.network.clone(),
 					voting_rule: self.env.voting_rule.clone(),
 					metrics: self.env.metrics.clone(),
+					justification_sender: self.env.justification_sender.clone(),
 					_phantom: PhantomData,
 				});

@@ -0,0 +1,102 @@
+// This file is part of Substrate.
+
+// Copyright (C) 2020 Parity Technologies (UK) Ltd.
+// SPDX-License-Identifier: GPL-3.0-or-later WITH Classpath-exception-2.0
+
+// This program is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// This program is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with this program. If not, see <https://www.gnu.org/licenses/>.
+
+use std::sync::Arc;
+use parking_lot::Mutex;
+
+use sp_runtime::traits::Block as BlockT;
+use sp_utils::mpsc::{tracing_unbounded, TracingUnboundedSender, TracingUnboundedReceiver};
+
+use crate::justification::GrandpaJustification;
+
+// Stream of justifications returned when subscribing.
+type JustificationStream<Block> = TracingUnboundedReceiver<GrandpaJustification<Block>>;
+
+// Sending endpoint for notifying about justifications.
+type JustificationSender<Block> = TracingUnboundedSender<GrandpaJustification<Block>>;
+
+// Collection of channel sending endpoints shared with the receiver side so they can register
+// themselves.
+type SharedJustificationSenders<Block> = Arc<Mutex<Vec<JustificationSender<Block>>>>;
+
+/// The sending half of the Grandpa justification channel(s).
+///
+/// Used to send notifications about justifications generated
+/// at the end of a Grandpa round.
+#[derive(Clone)]
+pub struct GrandpaJustificationSender<Block: BlockT> {
+	subscribers: SharedJustificationSenders<Block>
+}
+
+impl<Block: BlockT> GrandpaJustificationSender<Block> {
+	/// The `subscribers` should be shared with a corresponding
+	/// `GrandpaJustificationStream`.
+	fn new(subscribers: SharedJustificationSenders<Block>) -> Self {
+		Self {
+			subscribers,
+		}
+	}
+
+	/// Send out a notification to all subscribers that a new justification
+	/// is available for a block.
+	pub fn notify(&self, notification: GrandpaJustification<Block>) -> Result<(), ()> {
+		self.subscribers.lock().retain(|n| {
+			!n.is_closed() && n.unbounded_send(notification.clone()).is_ok()
+		});
+		Ok(())
+	}
+}
+
+/// The receiving half of the Grandpa justification channel.
+///
+/// Used to receive notifications about justifications generated
+/// at the end of a Grandpa round.
+/// The `GrandpaJustificationStream` entity stores the `SharedJustificationSenders`
+/// so it can be used to add more subscriptions.
+#[derive(Clone)]
+pub struct GrandpaJustificationStream<Block: BlockT> {
+	subscribers: SharedJustificationSenders<Block>
+}
+
+impl<Block: BlockT> GrandpaJustificationStream<Block> {
+	/// Creates a new pair of receiver and sender of justification notifications.
+	pub fn channel() -> (GrandpaJustificationSender<Block>, Self) {
+		let subscribers = Arc::new(Mutex::new(vec![]));
+		let receiver = GrandpaJustificationStream::new(subscribers.clone());
+		let sender = GrandpaJustificationSender::new(subscribers.clone());
+		(sender, receiver)
+	}
+
+	/// Create a new receiver of justification notifications.
+	///
+	/// The `subscribers` should be shared with a corresponding
+	/// `GrandpaJustificationSender`.
+	fn new(subscribers: SharedJustificationSenders<Block>) -> Self {
+		Self {
+			subscribers,
+		}
+	}
+
+	/// Subscribe to a channel through which justifications are sent
+	/// at the end of each Grandpa voting round.
+	pub fn subscribe(&self) -> JustificationStream<Block> {
+		let (sender, receiver) = tracing_unbounded("mpsc_justification_notification_stream");
+		self.subscribers.lock().push(sender);
+		receiver
+	}
+}
@@ -40,6 +40,7 @@ use crate::{
 use crate::authorities::SharedAuthoritySet;
 use crate::communication::{Network as NetworkT, NetworkBridge};
 use crate::consensus_changes::SharedConsensusChanges;
+use crate::notification::GrandpaJustificationSender;
 use sp_finality_grandpa::AuthorityId;
 use std::marker::{PhantomData, Unpin};

@@ -69,6 +70,7 @@ fn grandpa_observer<BE, Block: BlockT, Client, S, F>(
 	authority_set: &SharedAuthoritySet<Block::Hash, NumberFor<Block>>,
 	consensus_changes: &SharedConsensusChanges<Block::Hash, NumberFor<Block>>,
 	voters: &Arc<VoterSet<AuthorityId>>,
+	justification_sender: &Option<GrandpaJustificationSender<Block>>,
 	last_finalized_number: NumberFor<Block>,
 	commits: S,
 	note_round: F,
@@ -85,6 +87,7 @@ fn grandpa_observer<BE, Block: BlockT, Client, S, F>(
 	let consensus_changes = consensus_changes.clone();
 	let client = client.clone();
 	let voters = voters.clone();
+	let justification_sender = justification_sender.clone();

 	let observer = commits.try_fold(last_finalized_number, move |last_finalized_number, global| {
 		let (round, commit, callback) = match global {
@@ -127,6 +130,7 @@ fn grandpa_observer<BE, Block: BlockT, Client, S, F>(
 				finalized_number,
 				(round, commit).into(),
 				false,
+				&justification_sender,
 			) {
 				Ok(_) => {},
 				Err(e) => return future::err(e),
@@ -177,6 +181,7 @@ where
 		select_chain: _,
 		persistent_data,
 		voter_commands_rx,
+		justification_sender,
 		..
 	} = link;

@@ -192,7 +197,8 @@ where
 		network,
 		persistent_data,
 		config.keystore,
-		voter_commands_rx
+		voter_commands_rx,
+		Some(justification_sender),
 	);

 	let observer_work = observer_work
@@ -213,6 +219,7 @@ struct ObserverWork<B: BlockT, BE, Client, N: NetworkT<B>> {
 	persistent_data: PersistentData<B>,
 	keystore: Option<BareCryptoStorePtr>,
 	voter_commands_rx: TracingUnboundedReceiver<VoterCommand<B::Hash, NumberFor<B>>>,
+	justification_sender: Option<GrandpaJustificationSender<B>>,
 	_phantom: PhantomData<BE>,
 }

@@ -230,6 +237,7 @@ where
 		persistent_data: PersistentData<B>,
 		keystore: Option<BareCryptoStorePtr>,
 		voter_commands_rx: TracingUnboundedReceiver<VoterCommand<B::Hash, NumberFor<B>>>,
+		justification_sender: Option<GrandpaJustificationSender<B>>,
 	) -> Self {

 		let mut work = ObserverWork {
@@ -241,6 +249,7 @@ where
 			persistent_data,
 			keystore: keystore.clone(),
 			voter_commands_rx,
+			justification_sender,
 			_phantom: PhantomData,
 		};
 		work.rebuild_observer();
@@ -287,6 +296,7 @@ where
 			&self.persistent_data.authority_set,
 			&self.persistent_data.consensus_changes,
 			&voters,
+			&self.justification_sender,
 			last_finalized_number,
 			global_in,
 			note_round,
@@ -422,12 +432,14 @@ mod tests {
 		).unwrap();

 		let (_tx, voter_command_rx) = tracing_unbounded("");
+
 		let observer = ObserverWork::new(
 			client,
 			tester.net_handle.clone(),
 			persistent_data,
 			None,
 			voter_command_rx,
+			None,
 		);

 		// Trigger a reputation change through the gossip validator.
@@ -1567,6 +1567,7 @@ where
 		network,
 		voting_rule,
 		metrics: None,
+		justification_sender: None,
 		_phantom: PhantomData,
 	}
 }