consensus: handle justification sync for blocks authored locally (par…

…itytech#8698)

* consensus: add trait to control justification sync process

* network: implement JustificationSyncLink for NetworkService

* slots: handle justification sync in slot worker

* babe: fix slot worker instantiation

* aura: fix slot worker instantiation

* pow: handle justification sync in miner

* babe: fix tests

* aura: fix tests

* node: fix compilation

* node-template: fix compilation

* consensus: rename justification sync link parameter

* aura: fix test compilation

* consensus: slots: move JustificationSyncLink out of on_slot

bin/node/cli/src/service.rs

@@ -308,6 +308,7 @@ pub fn new_full_base(
 			env: proposer,
 			block_import,
 			sync_oracle: network.clone(),
+			justification_sync_link: network.clone(),
 			create_inherent_data_providers: move |parent, ()| {
 				let client_clone = client_clone.clone();
 				async move {

client/consensus/aura/src/lib.rs

@@ -109,7 +109,7 @@ fn slot_author<P: Pair>(slot: Slot, authorities: &[AuthorityId<P>]) -> Option<&A
 }
 
 /// Parameters of [`start_aura`].
-pub struct StartAuraParams<C, SC, I, PF, SO, BS, CAW, IDP> {
+pub struct StartAuraParams<C, SC, I, PF, SO, L, CIDP, BS, CAW> {
 	/// The duration of a slot.
 	pub slot_duration: SlotDuration,
 	/// The client to interact with the chain.
@@ -122,8 +122,10 @@ pub struct StartAuraParams<C, SC, I, PF, SO, BS, CAW, IDP> {
 	pub proposer_factory: PF,
 	/// The sync oracle that can give us the current sync status.
 	pub sync_oracle: SO,
+	/// Hook into the sync module to control the justification sync process.
+	pub justification_sync_link: L,
 	/// Something that can create the inherent data providers.
-	pub create_inherent_data_providers: IDP,
+	pub create_inherent_data_providers: CIDP,
 	/// Should we force the authoring of blocks?
 	pub force_authoring: bool,
 	/// The backoff strategy when we miss slots.
@@ -143,46 +145,49 @@ pub struct StartAuraParams<C, SC, I, PF, SO, BS, CAW, IDP> {
 }
 
 /// Start the aura worker. The returned future should be run in a futures executor.
-pub fn start_aura<P, B, C, SC, PF, I, SO, CAW, BS, Error, IDP>(
+pub fn start_aura<P, B, C, SC, I, PF, SO, L, CIDP, BS, CAW, Error>(
 	StartAuraParams {
 		slot_duration,
 		client,
 		select_chain,
 		block_import,
 		proposer_factory,
 		sync_oracle,
+		justification_sync_link,
 		create_inherent_data_providers,
 		force_authoring,
 		backoff_authoring_blocks,
 		keystore,
 		can_author_with,
 		block_proposal_slot_portion,
 		telemetry,
-	}: StartAuraParams<C, SC, I, PF, SO, BS, CAW, IDP>,
+	}: StartAuraParams<C, SC, I, PF, SO, L, CIDP, BS, CAW>,
 ) -> Result<impl Future<Output = ()>, sp_consensus::Error> where
+	P: Pair + Send + Sync,
+	P::Public: AppPublic + Hash + Member + Encode + Decode,
+	P::Signature: TryFrom<Vec<u8>> + Hash + Member + Encode + Decode,
 	B: BlockT,
 	C: ProvideRuntimeApi<B> + BlockOf + ProvideCache<B> + AuxStore + HeaderBackend<B> + Send + Sync,
 	C::Api: AuraApi<B, AuthorityId<P>>,
 	SC: SelectChain<B>,
+	I: BlockImport<B, Transaction = sp_api::TransactionFor<C, B>> + Send + Sync + 'static,
 	PF: Environment<B, Error = Error> + Send + Sync + 'static,
 	PF::Proposer: Proposer<B, Error = Error, Transaction = sp_api::TransactionFor<C, B>>,
-	P: Pair + Send + Sync,
-	P::Public: AppPublic + Hash + Member + Encode + Decode,
-	P::Signature: TryFrom<Vec<u8>> + Hash + Member + Encode + Decode,
-	I: BlockImport<B, Transaction = sp_api::TransactionFor<C, B>> + Send + Sync + 'static,
-	Error: std::error::Error + Send + From<sp_consensus::Error> + 'static,
 	SO: SyncOracle + Send + Sync + Clone,
-	CAW: CanAuthorWith<B> + Send,
+	L: sp_consensus::JustificationSyncLink<B>,
+	CIDP: CreateInherentDataProviders<B, ()> + Send,
+	CIDP::InherentDataProviders: InherentDataProviderExt + Send,
 	BS: BackoffAuthoringBlocksStrategy<NumberFor<B>> + Send + 'static,
-	IDP: CreateInherentDataProviders<B, ()> + Send,
-	IDP::InherentDataProviders: InherentDataProviderExt + Send,
+	CAW: CanAuthorWith<B> + Send,
+	Error: std::error::Error + Send + From<sp_consensus::Error> + 'static,
 {
-	let worker = build_aura_worker::<P, _, _, _, _, _, _, _>(BuildAuraWorkerParams {
+	let worker = build_aura_worker::<P, _, _, _, _, _, _, _, _>(BuildAuraWorkerParams {
 		client: client.clone(),
 		block_import,
 		proposer_factory,
 		keystore,
 		sync_oracle: sync_oracle.clone(),
+		justification_sync_link,
 		force_authoring,
 		backoff_authoring_blocks,
 		telemetry,
@@ -200,7 +205,7 @@ pub fn start_aura<P, B, C, SC, PF, I, SO, CAW, BS, Error, IDP>(
 }
 
 /// Parameters of [`build_aura_worker`].
-pub struct BuildAuraWorkerParams<C, I, PF, SO, BS> {
+pub struct BuildAuraWorkerParams<C, I, PF, SO, L, BS> {
 	/// The client to interact with the chain.
 	pub client: Arc<C>,
 	/// The block import.
@@ -209,6 +214,8 @@ pub struct BuildAuraWorkerParams<C, I, PF, SO, BS> {
 	pub proposer_factory: PF,
 	/// The sync oracle that can give us the current sync status.
 	pub sync_oracle: SO,
+	/// Hook into the sync module to control the justification sync process.
+	pub justification_sync_link: L,
 	/// Should we force the authoring of blocks?
 	pub force_authoring: bool,
 	/// The backoff strategy when we miss slots.
@@ -228,18 +235,19 @@ pub struct BuildAuraWorkerParams<C, I, PF, SO, BS> {
 /// Build the aura worker.
 ///
 /// The caller is responsible for running this worker, otherwise it will do nothing.
-pub fn build_aura_worker<P, B, C, PF, I, SO, BS, Error>(
+pub fn build_aura_worker<P, B, C, PF, I, SO, L, BS, Error>(
 	BuildAuraWorkerParams {
 		client,
 		block_import,
 		proposer_factory,
 		sync_oracle,
+		justification_sync_link,
 		backoff_authoring_blocks,
 		keystore,
 		block_proposal_slot_portion,
 		telemetry,
 		force_authoring,
-	}: BuildAuraWorkerParams<C, I, PF, SO, BS>,
+	}: BuildAuraWorkerParams<C, I, PF, SO, L, BS>,
 ) -> impl sc_consensus_slots::SlotWorker<B, <PF::Proposer as Proposer<B>>::Proof> where
 	B: BlockT,
 	C: ProvideRuntimeApi<B> + BlockOf + ProvideCache<B> + AuxStore + HeaderBackend<B> + Send + Sync,
@@ -252,6 +260,7 @@ pub fn build_aura_worker<P, B, C, PF, I, SO, BS, Error>(
 	I: BlockImport<B, Transaction = sp_api::TransactionFor<C, B>> + Send + Sync + 'static,
 	Error: std::error::Error + Send + From<sp_consensus::Error> + 'static,
 	SO: SyncOracle + Send + Sync + Clone,
+	L: sp_consensus::JustificationSyncLink<B>,
 	BS: BackoffAuthoringBlocksStrategy<NumberFor<B>> + Send + 'static,
 {
 	AuraWorker {
@@ -260,6 +269,7 @@ pub fn build_aura_worker<P, B, C, PF, I, SO, BS, Error>(
 		env: proposer_factory,
 		keystore,
 		sync_oracle,
+		justification_sync_link,
 		force_authoring,
 		backoff_authoring_blocks,
 		telemetry,
@@ -268,21 +278,22 @@ pub fn build_aura_worker<P, B, C, PF, I, SO, BS, Error>(
 	}
 }
 
-struct AuraWorker<C, E, I, P, SO, BS> {
+struct AuraWorker<C, E, I, P, SO, L, BS> {
 	client: Arc<C>,
 	block_import: I,
 	env: E,
 	keystore: SyncCryptoStorePtr,
 	sync_oracle: SO,
+	justification_sync_link: L,
 	force_authoring: bool,
 	backoff_authoring_blocks: Option<BS>,
 	block_proposal_slot_portion: SlotProportion,
 	telemetry: Option<TelemetryHandle>,
 	_key_type: PhantomData<P>,
 }
 
-impl<B, C, E, I, P, Error, SO, BS> sc_consensus_slots::SimpleSlotWorker<B>
-	for AuraWorker<C, E, I, P, SO, BS>
+impl<B, C, E, I, P, Error, SO, L, BS> sc_consensus_slots::SimpleSlotWorker<B>
+	for AuraWorker<C, E, I, P, SO, L, BS>
 where
 	B: BlockT,
 	C: ProvideRuntimeApi<B> + BlockOf + ProvideCache<B> + HeaderBackend<B> + Sync,
@@ -294,11 +305,13 @@ where
 	P::Public: AppPublic + Public + Member + Encode + Decode + Hash,
 	P::Signature: TryFrom<Vec<u8>> + Member + Encode + Decode + Hash + Debug,
 	SO: SyncOracle + Send + Clone,
+	L: sp_consensus::JustificationSyncLink<B>,
 	BS: BackoffAuthoringBlocksStrategy<NumberFor<B>> + Send + 'static,
 	Error: std::error::Error + Send + From<sp_consensus::Error> + 'static,
 {
 	type BlockImport = I;
 	type SyncOracle = SO;
+	type JustificationSyncLink = L;
 	type CreateProposer = Pin<Box<
 		dyn Future<Output = Result<E::Proposer, sp_consensus::Error>> + Send + 'static
 	>>;
@@ -425,6 +438,10 @@ where
 		&mut self.sync_oracle
 	}
 
+	fn justification_sync_link(&mut self) -> &mut Self::JustificationSyncLink {
+		&mut self.justification_sync_link
+	}
+
 	fn proposer(&mut self, block: &B::Header) -> Self::CreateProposer {
 		Box::pin(self.env.init(block).map_err(|e| {
 			sp_consensus::Error::ClientImport(format!("{:?}", e)).into()
@@ -725,13 +742,14 @@ mod tests {
 
 			let slot_duration = slot_duration(&*client).expect("slot duration available");
 
-			aura_futures.push(start_aura::<AuthorityPair, _, _, _, _, _, _, _, _, _, _>(StartAuraParams {
+			aura_futures.push(start_aura::<AuthorityPair, _, _, _, _, _, _, _, _, _, _, _>(StartAuraParams {
 				slot_duration,
 				block_import: client.clone(),
 				select_chain,
 				client,
 				proposer_factory: environ,
 				sync_oracle: DummyOracle,
+				justification_sync_link: (),
 				create_inherent_data_providers: |_, _| async {
 					let timestamp = TimestampInherentDataProvider::from_system_time();
 					let slot = InherentDataProvider::from_timestamp_and_duration(
@@ -804,6 +822,7 @@ mod tests {
 			env: environ,
 			keystore: keystore.into(),
 			sync_oracle: DummyOracle.clone(),
+			justification_sync_link: (),
 			force_authoring: false,
 			backoff_authoring_blocks: Some(BackoffAuthoringOnFinalizedHeadLagging::default()),
 			telemetry: None,
@@ -853,6 +872,7 @@ mod tests {
 			env: environ,
 			keystore: keystore.into(),
 			sync_oracle: DummyOracle.clone(),
+			justification_sync_link: (),
 			force_authoring: false,
 			backoff_authoring_blocks: Option::<()>::None,
 			telemetry: None,
@@ -871,7 +891,7 @@ mod tests {
 				duration: Duration::from_millis(1000),
 				chain_head: head,
 				block_size_limit: None,
-			},
+			}
 		)).unwrap();
 
 		// The returned block should be imported and we should be able to get its header by now.

client/consensus/babe/src/lib.rs

@@ -363,7 +363,7 @@ impl std::ops::Deref for Config {
 }
 
 /// Parameters for BABE.
-pub struct BabeParams<B: BlockT, C, E, I, SO, SC, CAW, BS, IDP> {
+pub struct BabeParams<B: BlockT, C, SC, E, I, SO, L, CIDP, BS, CAW> {
 	/// The keystore that manages the keys of the node.
 	pub keystore: SyncCryptoStorePtr,
 
@@ -384,8 +384,11 @@ pub struct BabeParams<B: BlockT, C, E, I, SO, SC, CAW, BS, IDP> {
 	/// A sync oracle
 	pub sync_oracle: SO,
 
+	/// Hook into the sync module to control the justification sync process.
+	pub justification_sync_link: L,
+
 	/// Something that can create the inherent data providers.
-	pub create_inherent_data_providers: IDP,
+	pub create_inherent_data_providers: CIDP,
 
 	/// Force authoring of blocks even if we are offline
 	pub force_authoring: bool,
@@ -411,40 +414,50 @@ pub struct BabeParams<B: BlockT, C, E, I, SO, SC, CAW, BS, IDP> {
 }
 
 /// Start the babe worker.
-pub fn start_babe<B, C, SC, E, I, SO, CAW, BS, Error, IDP>(BabeParams {
+pub fn start_babe<B, C, SC, E, I, SO, CIDP, BS, CAW, L, Error>(BabeParams {
 	keystore,
 	client,
 	select_chain,
 	env,
 	block_import,
 	sync_oracle,
+	justification_sync_link,
 	create_inherent_data_providers,
 	force_authoring,
 	backoff_authoring_blocks,
 	babe_link,
 	can_author_with,
 	block_proposal_slot_portion,
 	telemetry,
-}: BabeParams<B, C, E, I, SO, SC, CAW, BS, IDP>) -> Result<
+}: BabeParams<B, C, SC, E, I, SO, L, CIDP, BS, CAW>) -> Result<
 	BabeWorker<B>,
 	sp_consensus::Error,
 > where
 	B: BlockT,
-	C: ProvideRuntimeApi<B> + ProvideCache<B> + ProvideUncles<B> + BlockchainEvents<B>
-		+ HeaderBackend<B> + HeaderMetadata<B, Error = ClientError>
-		+ Send + Sync + 'static,
+	C: ProvideRuntimeApi<B>
+		+ ProvideCache<B>
+		+ ProvideUncles<B>
+		+ BlockchainEvents<B>
+		+ HeaderBackend<B>
+		+ HeaderMetadata<B, Error = ClientError>
+		+ Send
+		+ Sync
+		+ 'static,
 	C::Api: BabeApi<B>,
 	SC: SelectChain<B> + 'static,
 	E: Environment<B, Error = Error> + Send + Sync + 'static,
 	E::Proposer: Proposer<B, Error = Error, Transaction = sp_api::TransactionFor<C, B>>,
-	I: BlockImport<B, Error = ConsensusError, Transaction = sp_api::TransactionFor<C, B>> + Send
-		+ Sync + 'static,
-	Error: std::error::Error + Send + From<ConsensusError> + From<I::Error> + 'static,
+	I: BlockImport<B, Error = ConsensusError, Transaction = sp_api::TransactionFor<C, B>>
+		+ Send
+		+ Sync
+		+ 'static,
 	SO: SyncOracle + Send + Sync + Clone + 'static,
-	CAW: CanAuthorWith<B> + Send + Sync + 'static,
+	L: sp_consensus::JustificationSyncLink<B> + 'static,
+	CIDP: CreateInherentDataProviders<B, ()> + Send + Sync + 'static,
+	CIDP::InherentDataProviders: InherentDataProviderExt + Send,
 	BS: BackoffAuthoringBlocksStrategy<NumberFor<B>> + Send + 'static,
-	IDP: CreateInherentDataProviders<B, ()> + Send + Sync + 'static,
-	IDP::InherentDataProviders: InherentDataProviderExt + Send,
+	CAW: CanAuthorWith<B> + Send + Sync + 'static,
+	Error: std::error::Error + Send + From<ConsensusError> + From<I::Error> + 'static,
 {
 	const HANDLE_BUFFER_SIZE: usize = 1024;
 
@@ -456,6 +469,7 @@ pub fn start_babe<B, C, SC, E, I, SO, CAW, BS, Error, IDP>(BabeParams {
 		block_import,
 		env,
 		sync_oracle: sync_oracle.clone(),
+		justification_sync_link,
 		force_authoring,
 		backoff_authoring_blocks,
 		keystore,
@@ -600,11 +614,12 @@ type SlotNotificationSinks<B> = Arc<
 	Mutex<Vec<Sender<(Slot, ViableEpochDescriptor<<B as BlockT>::Hash, NumberFor<B>, Epoch>)>>>
 >;
 
-struct BabeSlotWorker<B: BlockT, C, E, I, SO, BS> {
+struct BabeSlotWorker<B: BlockT, C, E, I, SO, L, BS> {
 	client: Arc<C>,
 	block_import: I,
 	env: E,
 	sync_oracle: SO,
+	justification_sync_link: L,
 	force_authoring: bool,
 	backoff_authoring_blocks: Option<BS>,
 	keystore: SyncCryptoStorePtr,
@@ -615,8 +630,8 @@ struct BabeSlotWorker<B: BlockT, C, E, I, SO, BS> {
 	telemetry: Option<TelemetryHandle>,
 }
 
-impl<B, C, E, I, Error, SO, BS> sc_consensus_slots::SimpleSlotWorker<B>
-	for BabeSlotWorker<B, C, E, I, SO, BS>
+impl<B, C, E, I, Error, SO, L, BS> sc_consensus_slots::SimpleSlotWorker<B>
+	for BabeSlotWorker<B, C, E, I, SO, L, BS>
 where
 	B: BlockT,
 	C: ProvideRuntimeApi<B> +
@@ -628,12 +643,14 @@ where
 	E::Proposer: Proposer<B, Error = Error, Transaction = sp_api::TransactionFor<C, B>>,
 	I: BlockImport<B, Transaction = sp_api::TransactionFor<C, B>> + Send + Sync + 'static,
 	SO: SyncOracle + Send + Clone,
+	L: sp_consensus::JustificationSyncLink<B>,
 	BS: BackoffAuthoringBlocksStrategy<NumberFor<B>>,
 	Error: std::error::Error + Send + From<ConsensusError> + From<I::Error> + 'static,
 {
 	type EpochData = ViableEpochDescriptor<B::Hash, NumberFor<B>, Epoch>;
 	type Claim = (PreDigest, AuthorityId);
 	type SyncOracle = SO;
+	type JustificationSyncLink = L;
 	type CreateProposer = Pin<Box<
 		dyn Future<Output = Result<E::Proposer, sp_consensus::Error>> + Send + 'static
 	>>;
@@ -798,6 +815,10 @@ where
 		&mut self.sync_oracle
 	}
 
+	fn justification_sync_link(&mut self) -> &mut Self::JustificationSyncLink {
+		&mut self.justification_sync_link
+	}
+
 	fn proposer(&mut self, block: &B::Header) -> Self::CreateProposer {
 		Box::pin(self.env.init(block).map_err(|e| {
 			sp_consensus::Error::ClientImport(format!("{:?}", e))

client/consensus/babe/src/tests.rs

@@ -390,9 +390,7 @@ fn rejects_empty_block() {
 	})
 }
 
-fn run_one_test(
-	mutator: impl Fn(&mut TestHeader, Stage) + Send + Sync + 'static,
-) {
+fn run_one_test(mutator: impl Fn(&mut TestHeader, Stage) + Send + Sync + 'static) {
 	sp_tracing::try_init_simple();
 	let mutator = Arc::new(mutator) as Mutator;
 
@@ -473,6 +471,7 @@ fn run_one_test(
 			babe_link: data.link.clone(),
 			keystore,
 			can_author_with: sp_consensus::AlwaysCanAuthor,
+			justification_sync_link: (),
 			block_proposal_slot_portion: SlotProportion::new(0.5),
 			telemetry: None,
 		}).expect("Starts babe"));