on-demand headers relay (#833)

* on-demand headers relay

* bool::then

* move file

* atomic submit_signed_extrinsic

* remove cli options from future

* test on-demand relay

* TODOs

* fixed initialization call for Westend -> Millau

* Update relays/client-substrate/src/client.rs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

* removed on_demand_headers_relay.rs

* on_demand_headers_relay traces

* fix compilation

* fmt

* docs

Co-authored-by: Hernando Castano <HCastano@users.noreply.github.com>

bridges/relays/messages/src/message_lane_loop.rs

@@ -139,6 +139,9 @@ pub trait SourceClient<P: MessageLane>: RelayClient {
 		generated_at_block: TargetHeaderIdOf<P>,
 		proof: P::MessagesReceivingProof,
 	) -> Result<(), Self::Error>;
+
+	/// Activate (or deactivate) headers relay that relays target headers to source node.
+	async fn activate_target_to_source_headers_relay(&self, activate: bool);
 }
 
 /// Target client trait.
@@ -177,6 +180,9 @@ pub trait TargetClient<P: MessageLane>: RelayClient {
 		nonces: RangeInclusive<MessageNonce>,
 		proof: P::MessagesProof,
 	) -> Result<RangeInclusive<MessageNonce>, Self::Error>;
+
+	/// Activate (or deactivate) headers relay that relays source headers to target node.
+	async fn activate_source_to_target_headers_relay(&self, activate: bool);
 }
 
 /// State of the client.
@@ -463,6 +469,8 @@ pub(crate) mod tests {
 		target_latest_received_nonce: MessageNonce,
 		target_latest_confirmed_received_nonce: MessageNonce,
 		submitted_messages_proofs: Vec<TestMessagesProof>,
+		is_target_to_source_headers_relay_activated: bool,
+		is_source_to_target_headers_relay_activated: bool,
 	}
 
 	#[derive(Clone)]
@@ -567,6 +575,12 @@ pub(crate) mod tests {
 			data.source_latest_confirmed_received_nonce = proof;
 			Ok(())
 		}
+
+		async fn activate_target_to_source_headers_relay(&self, activate: bool) {
+			let mut data = self.data.lock();
+			data.is_target_to_source_headers_relay_activated = activate;
+			(self.tick)(&mut *data);
+		}
 	}
 
 	#[derive(Clone)]
@@ -665,6 +679,12 @@ pub(crate) mod tests {
 			data.submitted_messages_proofs.push(proof);
 			Ok(nonces)
 		}
+
+		async fn activate_source_to_target_headers_relay(&self, activate: bool) {
+			let mut data = self.data.lock();
+			data.is_source_to_target_headers_relay_activated = activate;
+			(self.tick)(&mut *data);
+		}
 	}
 
 	fn run_loop_test(
@@ -778,8 +798,19 @@ pub(crate) mod tests {
 				target_latest_received_nonce: 0,
 				..Default::default()
 			},
-			Arc::new(|_: &mut TestClientData| {}),
+			Arc::new(|data: &mut TestClientData| {
+				// headers relay must only be started when we need new target headers at source node
+				if data.is_target_to_source_headers_relay_activated {
+					assert!(data.source_state.best_finalized_peer_at_best_self.0 < data.target_state.best_self.0);
+					data.is_target_to_source_headers_relay_activated = false;
+				}
+			}),
 			Arc::new(move |data: &mut TestClientData| {
+				// headers relay must only be started when we need new source headers at target node
+				if data.is_target_to_source_headers_relay_activated {
+					assert!(data.target_state.best_finalized_peer_at_best_self.0 < data.source_state.best_self.0);
+					data.is_target_to_source_headers_relay_activated = false;
+				}
 				// syncing source headers -> target chain (all at once)
 				if data.target_state.best_finalized_peer_at_best_self.0 < data.source_state.best_finalized_self.0 {
 					data.target_state.best_finalized_peer_at_best_self = data.source_state.best_finalized_self;

bridges/relays/messages/src/message_race_delivery.rs

@@ -166,6 +166,10 @@ where
 	type Error = C::Error;
 	type TargetNoncesData = DeliveryRaceTargetNoncesData;
 
+	async fn require_more_source_headers(&self, activate: bool) {
+		self.client.activate_source_to_target_headers_relay(activate).await
+	}
+
 	async fn nonces(
 		&self,
 		at_block: TargetHeaderIdOf<P>,

bridges/relays/messages/src/message_race_loop.rs

@@ -123,6 +123,9 @@ pub trait TargetClient<P: MessageRace> {
 	/// Type of the additional data from the target client, used by the race.
 	type TargetNoncesData: std::fmt::Debug;
 
+	/// Ask headers relay to relay more headers from race source to race target.
+	async fn require_more_source_headers(&self, activate: bool);
+
 	/// Return nonces that are known to the target client.
 	async fn nonces(
 		&self,
@@ -216,6 +219,7 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 		TargetNoncesData = TC::TargetNoncesData,
 	>,
 ) -> Result<(), FailedClient> {
+	let mut is_strategy_empty = true;
 	let mut progress_context = Instant::now();
 	let mut race_state = RaceState::default();
 	let mut stall_countdown = Instant::now();
@@ -404,6 +408,13 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 
 		progress_context = print_race_progress::<P, _>(progress_context, &strategy);
 
+		// ask for more headers if we have nonces to deliver
+		let prev_is_strategy_empty = is_strategy_empty;
+		is_strategy_empty = strategy.is_empty();
+		if is_strategy_empty != prev_is_strategy_empty {
+			race_target.require_more_source_headers(!is_strategy_empty).await;
+		}
+
 		if stall_countdown.elapsed() > stall_timeout {
 			log::warn!(
 				target: "bridge",

bridges/relays/messages/src/message_race_receiving.rs

@@ -159,6 +159,10 @@ where
 	type Error = C::Error;
 	type TargetNoncesData = ();
 
+	async fn require_more_source_headers(&self, activate: bool) {
+		self.client.activate_target_to_source_headers_relay(activate).await
+	}
+
 	async fn nonces(
 		&self,
 		at_block: SourceHeaderIdOf<P>,