Bridge: added free headers submission support to the substrate-relay (#4157)

Original PR:
https://github.com/paritytech/parity-bridges-common/pull/2884. Since
chain-specific code lives in the `parity-bridges-common` repo, some
parts of original PR will require another PR

---------

Co-authored-by: Adrian Catangiu <adrian@parity.io>

bridges/relays/finality/src/finality_loop.rs

@@ -29,7 +29,7 @@ use crate::{
 use async_trait::async_trait;
 use backoff::{backoff::Backoff, ExponentialBackoff};
 use futures::{future::Fuse, select, Future, FutureExt};
-use num_traits::Saturating;
+use num_traits::{Saturating, Zero};
 use relay_utils::{
 	metrics::MetricsParams, relay_loop::Client as RelayClient, retry_backoff, FailedClient,
 	HeaderId, MaybeConnectionError, TrackedTransactionStatus, TransactionTracker,
@@ -39,6 +39,17 @@ use std::{
 	time::{Duration, Instant},
 };
 
+/// Type of headers that we relay.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub enum HeadersToRelay {
+	/// Relay all headers.
+	All,
+	/// Relay only mandatory headers.
+	Mandatory,
+	/// Relay only free (including mandatory) headers.
+	Free,
+}
+
 /// Finality proof synchronization loop parameters.
 #[derive(Debug, Clone)]
 pub struct FinalitySyncParams {
@@ -63,7 +74,7 @@ pub struct FinalitySyncParams {
 	/// Timeout before we treat our transactions as lost and restart the whole sync process.
 	pub stall_timeout: Duration,
 	/// If true, only mandatory headers are relayed.
-	pub only_mandatory_headers: bool,
+	pub headers_to_relay: HeadersToRelay,
 }
 
 /// Source client used in finality synchronization loop.
@@ -90,11 +101,16 @@ pub trait TargetClient<P: FinalitySyncPipeline>: RelayClient {
 		&self,
 	) -> Result<HeaderId<P::Hash, P::Number>, Self::Error>;
 
+	/// Get free source headers submission interval, if it is configured in the
+	/// target runtime.
+	async fn free_source_headers_interval(&self) -> Result<Option<P::Number>, Self::Error>;
+
 	/// Submit header finality proof.
 	async fn submit_finality_proof(
 		&self,
 		header: P::Header,
 		proof: P::FinalityProof,
+		is_free_execution_expected: bool,
 	) -> Result<Self::TransactionTracker, Self::Error>;
 }
 
@@ -104,9 +120,13 @@ pub fn metrics_prefix<P: FinalitySyncPipeline>() -> String {
 	format!("{}_to_{}_Sync", P::SOURCE_NAME, P::TARGET_NAME)
 }
 
+/// Finality sync information.
 pub struct SyncInfo<P: FinalitySyncPipeline> {
+	/// Best finalized header at the source client.
 	pub best_number_at_source: P::Number,
+	/// Best source header, known to the target client.
 	pub best_number_at_target: P::Number,
+	/// Whether the target client follows the same fork as the source client do.
 	pub is_using_same_fork: bool,
 }
 
@@ -183,6 +203,7 @@ impl<Tracker: TransactionTracker, Number: Debug + PartialOrd> Transaction<Tracke
 		target_client: &TC,
 		header: P::Header,
 		justification: P::FinalityProof,
+		is_free_execution_expected: bool,
 	) -> Result<Self, TC::Error> {
 		let header_number = header.number();
 		log::debug!(
@@ -193,7 +214,9 @@ impl<Tracker: TransactionTracker, Number: Debug + PartialOrd> Transaction<Tracke
 			P::TARGET_NAME,
 		);
 
-		let tracker = target_client.submit_finality_proof(header, justification).await?;
+		let tracker = target_client
+			.submit_finality_proof(header, justification, is_free_execution_expected)
+			.await?;
 		Ok(Transaction { tracker, header_number })
 	}
 
@@ -292,6 +315,7 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 	pub async fn select_header_to_submit(
 		&mut self,
 		info: &SyncInfo<P>,
+		free_headers_interval: Option<P::Number>,
 	) -> Result<Option<JustifiedHeader<P>>, Error<P, SC::Error, TC::Error>> {
 		// to see that the loop is progressing
 		log::trace!(
@@ -302,9 +326,15 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 		);
 
 		// read missing headers
-		let selector = JustifiedHeaderSelector::new::<SC, TC>(&self.source_client, info).await?;
+		let selector = JustifiedHeaderSelector::new::<SC, TC>(
+			&self.source_client,
+			info,
+			self.sync_params.headers_to_relay,
+			free_headers_interval,
+		)
+		.await?;
 		// if we see that the header schedules GRANDPA change, we need to submit it
-		if self.sync_params.only_mandatory_headers {
+		if self.sync_params.headers_to_relay == HeadersToRelay::Mandatory {
 			return Ok(selector.select_mandatory())
 		}
 
@@ -312,7 +342,12 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 		// => even if we have already selected some header and its persistent finality proof,
 		// we may try to select better header by reading non-persistent proofs from the stream
 		self.finality_proofs_buf.fill(&mut self.finality_proofs_stream);
-		let maybe_justified_header = selector.select(&self.finality_proofs_buf);
+		let maybe_justified_header = selector.select(
+			info,
+			self.sync_params.headers_to_relay,
+			free_headers_interval,
+			&self.finality_proofs_buf,
+		);
 
 		// remove obsolete 'recent' finality proofs + keep its size under certain limit
 		let oldest_finality_proof_to_keep = maybe_justified_header
@@ -329,6 +364,7 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 
 	pub async fn run_iteration(
 		&mut self,
+		free_headers_interval: Option<P::Number>,
 	) -> Result<
 		Option<Transaction<TC::TransactionTracker, P::Number>>,
 		Error<P, SC::Error, TC::Error>,
@@ -345,12 +381,16 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 		}
 
 		// submit new header if we have something new
-		match self.select_header_to_submit(&info).await? {
+		match self.select_header_to_submit(&info, free_headers_interval).await? {
 			Some(header) => {
-				let transaction =
-					Transaction::submit(&self.target_client, header.header, header.proof)
-						.await
-						.map_err(Error::Target)?;
+				let transaction = Transaction::submit(
+					&self.target_client,
+					header.header,
+					header.proof,
+					self.sync_params.headers_to_relay == HeadersToRelay::Free,
+				)
+				.await
+				.map_err(Error::Target)?;
 				self.best_submitted_number = Some(transaction.header_number);
 				Ok(Some(transaction))
 			},
@@ -378,9 +418,11 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 		let exit_signal = exit_signal.fuse();
 		futures::pin_mut!(exit_signal, proof_submission_tx_tracker);
 
+		let free_headers_interval = free_headers_interval(&self.target_client).await?;
+
 		loop {
 			// run loop iteration
-			let next_tick = match self.run_iteration().await {
+			let next_tick = match self.run_iteration(free_headers_interval).await {
 				Ok(Some(tx)) => {
 					proof_submission_tx_tracker
 						.set(tx.track::<P, SC, _>(self.target_client.clone()).fuse());
@@ -433,6 +475,52 @@ impl<P: FinalitySyncPipeline, SC: SourceClient<P>, TC: TargetClient<P>> Finality
 	}
 }
 
+async fn free_headers_interval<P: FinalitySyncPipeline>(
+	target_client: &impl TargetClient<P>,
+) -> Result<Option<P::Number>, FailedClient> {
+	match target_client.free_source_headers_interval().await {
+		Ok(Some(free_headers_interval)) if !free_headers_interval.is_zero() => {
+			log::trace!(
+				target: "bridge",
+				"Free headers interval for {} headers at {} is: {:?}",
+				P::SOURCE_NAME,
+				P::TARGET_NAME,
+				free_headers_interval,
+			);
+			Ok(Some(free_headers_interval))
+		},
+		Ok(Some(_free_headers_interval)) => {
+			log::trace!(
+				target: "bridge",
+				"Free headers interval for {} headers at {} is zero. Not submitting any free headers",
+				P::SOURCE_NAME,
+				P::TARGET_NAME,
+			);
+			Ok(None)
+		},
+		Ok(None) => {
+			log::trace!(
+				target: "bridge",
+				"Free headers interval for {} headers at {} is None. Not submitting any free headers",
+				P::SOURCE_NAME,
+				P::TARGET_NAME,
+			);
+
+			Ok(None)
+		},
+		Err(e) => {
+			log::error!(
+				target: "bridge",
+				"Failed to read free headers interval for {} headers at {}: {:?}",
+				P::SOURCE_NAME,
+				P::TARGET_NAME,
+				e,
+			);
+			Err(FailedClient::Target)
+		},
+	}
+}
+
 /// Run finality proofs synchronization loop.
 pub async fn run<P: FinalitySyncPipeline>(
 	source_client: impl SourceClient<P>,
@@ -509,7 +597,7 @@ mod tests {
 			tick: Duration::from_secs(0),
 			recent_finality_proofs_limit: 1024,
 			stall_timeout: Duration::from_secs(1),
-			only_mandatory_headers: false,
+			headers_to_relay: HeadersToRelay::All,
 		}
 	}
 
@@ -593,8 +681,8 @@ mod tests {
 		);
 	}
 
-	fn run_only_mandatory_headers_mode_test(
-		only_mandatory_headers: bool,
+	fn run_headers_to_relay_mode_test(
+		headers_to_relay: HeadersToRelay,
 		has_mandatory_headers: bool,
 	) -> Option<JustifiedHeader<TestFinalitySyncPipeline>> {
 		let (exit_sender, _) = futures::channel::mpsc::unbounded();
@@ -619,7 +707,7 @@ mod tests {
 					tick: Duration::from_secs(0),
 					recent_finality_proofs_limit: 0,
 					stall_timeout: Duration::from_secs(0),
-					only_mandatory_headers,
+					headers_to_relay,
 				},
 				None,
 			);
@@ -628,16 +716,22 @@ mod tests {
 				best_number_at_target: 5,
 				is_using_same_fork: true,
 			};
-			finality_loop.select_header_to_submit(&info).await.unwrap()
+			finality_loop.select_header_to_submit(&info, Some(3)).await.unwrap()
 		})
 	}
 
 	#[test]
-	fn select_header_to_submit_skips_non_mandatory_headers_when_only_mandatory_headers_are_required(
-	) {
-		assert_eq!(run_only_mandatory_headers_mode_test(true, false), None);
+	fn select_header_to_submit_may_select_non_mandatory_header() {
+		assert_eq!(run_headers_to_relay_mode_test(HeadersToRelay::Mandatory, false), None);
 		assert_eq!(
-			run_only_mandatory_headers_mode_test(false, false),
+			run_headers_to_relay_mode_test(HeadersToRelay::Free, false),
+			Some(JustifiedHeader {
+				header: TestSourceHeader(false, 10, 10),
+				proof: TestFinalityProof(10)
+			}),
+		);
+		assert_eq!(
+			run_headers_to_relay_mode_test(HeadersToRelay::All, false),
 			Some(JustifiedHeader {
 				header: TestSourceHeader(false, 10, 10),
 				proof: TestFinalityProof(10)
@@ -646,17 +740,23 @@ mod tests {
 	}
 
 	#[test]
-	fn select_header_to_submit_selects_mandatory_headers_when_only_mandatory_headers_are_required()
-	{
+	fn select_header_to_submit_may_select_mandatory_header() {
 		assert_eq!(
-			run_only_mandatory_headers_mode_test(true, true),
+			run_headers_to_relay_mode_test(HeadersToRelay::Mandatory, true),
 			Some(JustifiedHeader {
 				header: TestSourceHeader(true, 8, 8),
 				proof: TestFinalityProof(8)
 			}),
 		);
 		assert_eq!(
-			run_only_mandatory_headers_mode_test(false, true),
+			run_headers_to_relay_mode_test(HeadersToRelay::Free, true),
+			Some(JustifiedHeader {
+				header: TestSourceHeader(true, 8, 8),
+				proof: TestFinalityProof(8)
+			}),
+		);
+		assert_eq!(
+			run_headers_to_relay_mode_test(HeadersToRelay::All, true),
 			Some(JustifiedHeader {
 				header: TestSourceHeader(true, 8, 8),
 				proof: TestFinalityProof(8)
@@ -690,7 +790,7 @@ mod tests {
 				test_sync_params(),
 				Some(metrics_sync.clone()),
 			);
-			finality_loop.run_iteration().await.unwrap()
+			finality_loop.run_iteration(None).await.unwrap()
 		});
 
 		assert!(!metrics_sync.is_using_same_fork());

bridges/relays/finality/src/headers.rs

@@ -16,10 +16,11 @@
 
 use crate::{
 	finality_loop::SyncInfo, finality_proofs::FinalityProofsBuf, Error, FinalitySyncPipeline,
-	SourceClient, SourceHeader, TargetClient,
+	HeadersToRelay, SourceClient, SourceHeader, TargetClient,
 };
 
 use bp_header_chain::FinalityProof;
+use num_traits::Saturating;
 use std::cmp::Ordering;
 
 /// Unjustified headers container. Ordered by header number.
@@ -50,9 +51,13 @@ pub enum JustifiedHeaderSelector<P: FinalitySyncPipeline> {
 }
 
 impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
+	/// Selects last header with persistent justification, missing from the target and matching
+	/// the `headers_to_relay` criteria.
 	pub(crate) async fn new<SC: SourceClient<P>, TC: TargetClient<P>>(
 		source_client: &SC,
 		info: &SyncInfo<P>,
+		headers_to_relay: HeadersToRelay,
+		free_headers_interval: Option<P::Number>,
 	) -> Result<Self, Error<P, SC::Error, TC::Error>> {
 		let mut unjustified_headers = Vec::new();
 		let mut maybe_justified_header = None;
@@ -70,12 +75,19 @@ impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
 					return Ok(Self::Mandatory(JustifiedHeader { header, proof }))
 				},
 				(true, None) => return Err(Error::MissingMandatoryFinalityProof(header.number())),
-				(false, Some(proof)) => {
+				(false, Some(proof))
+					if need_to_relay::<P>(
+						info,
+						headers_to_relay,
+						free_headers_interval,
+						&header,
+					) =>
+				{
 					log::trace!(target: "bridge", "Header {:?} has persistent finality proof", header_number);
 					unjustified_headers.clear();
 					maybe_justified_header = Some(JustifiedHeader { header, proof });
 				},
-				(false, None) => {
+				_ => {
 					unjustified_headers.push(header);
 				},
 			}
@@ -97,6 +109,7 @@ impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
 		})
 	}
 
+	/// Returns selected mandatory header if we have seen one. Otherwise returns `None`.
 	pub fn select_mandatory(self) -> Option<JustifiedHeader<P>> {
 		match self {
 			JustifiedHeaderSelector::Mandatory(header) => Some(header),
@@ -104,7 +117,15 @@ impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
 		}
 	}
 
-	pub fn select(self, buf: &FinalityProofsBuf<P>) -> Option<JustifiedHeader<P>> {
+	/// Tries to improve previously selected header using ephemeral
+	/// justifications stream.
+	pub fn select(
+		self,
+		info: &SyncInfo<P>,
+		headers_to_relay: HeadersToRelay,
+		free_headers_interval: Option<P::Number>,
+		buf: &FinalityProofsBuf<P>,
+	) -> Option<JustifiedHeader<P>> {
 		let (unjustified_headers, maybe_justified_header) = match self {
 			JustifiedHeaderSelector::Mandatory(justified_header) => return Some(justified_header),
 			JustifiedHeaderSelector::Regular(unjustified_headers, justified_header) =>
@@ -122,7 +143,14 @@ impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
 			(maybe_finality_proof, maybe_unjustified_header)
 		{
 			match finality_proof.target_header_number().cmp(&unjustified_header.number()) {
-				Ordering::Equal => {
+				Ordering::Equal
+					if need_to_relay::<P>(
+						info,
+						headers_to_relay,
+						free_headers_interval,
+						&unjustified_header,
+					) =>
+				{
 					log::trace!(
 						target: "bridge",
 						"Managed to improve selected {} finality proof {:?} to {:?}.",
@@ -135,6 +163,10 @@ impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
 						proof: finality_proof.clone(),
 					})
 				},
+				Ordering::Equal => {
+					maybe_finality_proof = finality_proofs_iter.next();
+					maybe_unjustified_header = unjustified_headers_iter.next();
+				},
 				Ordering::Less => maybe_unjustified_header = unjustified_headers_iter.next(),
 				Ordering::Greater => {
 					maybe_finality_proof = finality_proofs_iter.next();
@@ -152,6 +184,27 @@ impl<P: FinalitySyncPipeline> JustifiedHeaderSelector<P> {
 	}
 }
 
+/// Returns true if we want to relay header `header_number`.
+fn need_to_relay<P: FinalitySyncPipeline>(
+	info: &SyncInfo<P>,
+	headers_to_relay: HeadersToRelay,
+	free_headers_interval: Option<P::Number>,
+	header: &P::Header,
+) -> bool {
+	match headers_to_relay {
+		HeadersToRelay::All => true,
+		HeadersToRelay::Mandatory => header.is_mandatory(),
+		HeadersToRelay::Free =>
+			header.is_mandatory() ||
+				free_headers_interval
+					.map(|free_headers_interval| {
+						header.number().saturating_sub(info.best_number_at_target) >=
+							free_headers_interval
+					})
+					.unwrap_or(false),
+	}
+}
+
 #[cfg(test)]
 mod tests {
 	use super::*;
@@ -159,13 +212,22 @@ mod tests {
 
 	#[test]
 	fn select_better_recent_finality_proof_works() {
+		let info = SyncInfo {
+			best_number_at_source: 10,
+			best_number_at_target: 5,
+			is_using_same_fork: true,
+		};
+
 		// if there are no unjustified headers, nothing is changed
 		let finality_proofs_buf =
 			FinalityProofsBuf::<TestFinalitySyncPipeline>::new(vec![TestFinalityProof(5)]);
 		let justified_header =
 			JustifiedHeader { header: TestSourceHeader(false, 2, 2), proof: TestFinalityProof(2) };
 		let selector = JustifiedHeaderSelector::Regular(vec![], justified_header.clone());
-		assert_eq!(selector.select(&finality_proofs_buf), Some(justified_header));
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::All, None, &finality_proofs_buf),
+			Some(justified_header)
+		);
 
 		// if there are no buffered finality proofs, nothing is changed
 		let finality_proofs_buf = FinalityProofsBuf::<TestFinalitySyncPipeline>::new(vec![]);
@@ -175,7 +237,10 @@ mod tests {
 			vec![TestSourceHeader(false, 5, 5)],
 			justified_header.clone(),
 		);
-		assert_eq!(selector.select(&finality_proofs_buf), Some(justified_header));
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::All, None, &finality_proofs_buf),
+			Some(justified_header)
+		);
 
 		// if there's no intersection between recent finality proofs and unjustified headers,
 		// nothing is changed
@@ -189,7 +254,10 @@ mod tests {
 			vec![TestSourceHeader(false, 9, 9), TestSourceHeader(false, 10, 10)],
 			justified_header.clone(),
 		);
-		assert_eq!(selector.select(&finality_proofs_buf), Some(justified_header));
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::All, None, &finality_proofs_buf),
+			Some(justified_header)
+		);
 
 		// if there's intersection between recent finality proofs and unjustified headers, but there
 		// are no proofs in this intersection, nothing is changed
@@ -207,7 +275,10 @@ mod tests {
 			],
 			justified_header.clone(),
 		);
-		assert_eq!(selector.select(&finality_proofs_buf), Some(justified_header));
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::All, None, &finality_proofs_buf),
+			Some(justified_header)
+		);
 
 		// if there's intersection between recent finality proofs and unjustified headers and
 		// there's a proof in this intersection:
@@ -228,11 +299,63 @@ mod tests {
 			justified_header,
 		);
 		assert_eq!(
-			selector.select(&finality_proofs_buf),
+			selector.select(&info, HeadersToRelay::All, None, &finality_proofs_buf),
 			Some(JustifiedHeader {
 				header: TestSourceHeader(false, 9, 9),
 				proof: TestFinalityProof(9)
 			})
 		);
+
+		// when only free headers needs to be relayed and there are no free headers
+		let finality_proofs_buf = FinalityProofsBuf::<TestFinalitySyncPipeline>::new(vec![
+			TestFinalityProof(7),
+			TestFinalityProof(9),
+		]);
+		let selector = JustifiedHeaderSelector::None(vec![
+			TestSourceHeader(false, 8, 8),
+			TestSourceHeader(false, 9, 9),
+			TestSourceHeader(false, 10, 10),
+		]);
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::Free, Some(7), &finality_proofs_buf),
+			None,
+		);
+
+		// when only free headers needs to be relayed, mandatory header may be selected
+		let finality_proofs_buf = FinalityProofsBuf::<TestFinalitySyncPipeline>::new(vec![
+			TestFinalityProof(6),
+			TestFinalityProof(9),
+		]);
+		let selector = JustifiedHeaderSelector::None(vec![
+			TestSourceHeader(false, 8, 8),
+			TestSourceHeader(true, 9, 9),
+			TestSourceHeader(false, 10, 10),
+		]);
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::Free, Some(7), &finality_proofs_buf),
+			Some(JustifiedHeader {
+				header: TestSourceHeader(true, 9, 9),
+				proof: TestFinalityProof(9)
+			})
+		);
+
+		// when only free headers needs to be relayed and there is free header
+		let finality_proofs_buf = FinalityProofsBuf::<TestFinalitySyncPipeline>::new(vec![
+			TestFinalityProof(7),
+			TestFinalityProof(9),
+			TestFinalityProof(14),
+		]);
+		let selector = JustifiedHeaderSelector::None(vec![
+			TestSourceHeader(false, 7, 7),
+			TestSourceHeader(false, 10, 10),
+			TestSourceHeader(false, 14, 14),
+		]);
+		assert_eq!(
+			selector.select(&info, HeadersToRelay::Free, Some(7), &finality_proofs_buf),
+			Some(JustifiedHeader {
+				header: TestSourceHeader(false, 14, 14),
+				proof: TestFinalityProof(14)
+			})
+		);
 	}
 }

bridges/relays/finality/src/lib.rs

@@ -21,7 +21,9 @@
 
 pub use crate::{
 	base::{FinalityPipeline, SourceClientBase},
-	finality_loop::{metrics_prefix, run, FinalitySyncParams, SourceClient, TargetClient},
+	finality_loop::{
+		metrics_prefix, run, FinalitySyncParams, HeadersToRelay, SourceClient, TargetClient,
+	},
 	finality_proofs::{FinalityProofsBuf, FinalityProofsStream},
 	sync_loop_metrics::SyncLoopMetrics,
 };

bridges/relays/finality/src/mock.rs

@@ -198,10 +198,15 @@ impl TargetClient<TestFinalitySyncPipeline> for TestTargetClient {
 		Ok(data.target_best_block_id)
 	}
 
+	async fn free_source_headers_interval(&self) -> Result<Option<TestNumber>, TestError> {
+		Ok(Some(3))
+	}
+
 	async fn submit_finality_proof(
 		&self,
 		header: TestSourceHeader,
 		proof: TestFinalityProof,
+		_is_free_execution_expected: bool,
 	) -> Result<TestTransactionTracker, TestError> {
 		let mut data = self.data.lock();
 		(self.on_method_call)(&mut data);