Fix issue with on-demand headers relay not starting (#921)

* require lane source -> target headers when weknow all headers that have produced messages, but are unable to deliver all messages because of unrewarded relayers vec capacity

* Update relays/messages/src/message_race_delivery.rs

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

Co-authored-by: Tomasz Drwięga <tomusdrw@users.noreply.github.com>

bridges/relays/messages/src/message_lane_loop.rs

@@ -582,6 +582,9 @@ pub(crate) mod tests {
 		) -> Result<(), TestError> {
 			let mut data = self.data.lock();
 			(self.tick)(&mut *data);
+			data.source_state.best_self =
+				HeaderId(data.source_state.best_self.0 + 1, data.source_state.best_self.1 + 1);
+			data.source_state.best_finalized_self = data.source_state.best_self;
 			data.submitted_messages_receiving_proofs.push(proof);
 			data.source_latest_confirmed_received_nonce = proof;
 			Ok(())
@@ -684,6 +687,7 @@ pub(crate) mod tests {
 			}
 			data.target_state.best_self =
 				HeaderId(data.target_state.best_self.0 + 1, data.target_state.best_self.1 + 1);
+			data.target_state.best_finalized_self = data.target_state.best_self;
 			data.target_latest_received_nonce = *proof.0.end();
 			if let Some(target_latest_confirmed_received_nonce) = proof.1 {
 				data.target_latest_confirmed_received_nonce = target_latest_confirmed_received_nonce;
@@ -812,37 +816,37 @@ pub(crate) mod tests {
 				..Default::default()
 			},
 			Arc::new(|data: &mut TestClientData| {
+				// blocks are produced on every tick
+				data.source_state.best_self =
+					HeaderId(data.source_state.best_self.0 + 1, data.source_state.best_self.1 + 1);
+				data.source_state.best_finalized_self = data.source_state.best_self;
 				// headers relay must only be started when we need new target headers at source node
 				if data.target_to_source_header_required.is_some() {
 					assert!(data.source_state.best_finalized_peer_at_best_self.0 < data.target_state.best_self.0);
 					data.target_to_source_header_required = None;
 				}
+				// syncing target headers -> source chain
+				if let Some(last_requirement) = data.target_to_source_header_requirements.last() {
+					if *last_requirement != data.source_state.best_finalized_peer_at_best_self {
+						data.source_state.best_finalized_peer_at_best_self = *last_requirement;
+					}
+				}
 			}),
 			Arc::new(move |data: &mut TestClientData| {
+				// blocks are produced on every tick
+				data.target_state.best_self =
+					HeaderId(data.target_state.best_self.0 + 1, data.target_state.best_self.1 + 1);
+				data.target_state.best_finalized_self = data.target_state.best_self;
 				// headers relay must only be started when we need new source headers at target node
 				if data.source_to_target_header_required.is_some() {
 					assert!(data.target_state.best_finalized_peer_at_best_self.0 < data.source_state.best_self.0);
 					data.source_to_target_header_required = None;
 				}
-				// syncing source headers -> target chain (all at once)
+				// syncing source headers -> target chain
-				if data.target_state.best_finalized_peer_at_best_self.0 < data.source_state.best_finalized_self.0 {
+				if let Some(last_requirement) = data.source_to_target_header_requirements.last() {
-					data.target_state.best_finalized_peer_at_best_self = data.source_state.best_finalized_self;
+					if *last_requirement != data.target_state.best_finalized_peer_at_best_self {
-				}
+						data.target_state.best_finalized_peer_at_best_self = *last_requirement;
-				// syncing source headers -> target chain (all at once)
+					}
-				if data.source_state.best_finalized_peer_at_best_self.0 < data.target_state.best_finalized_self.0 {
-					data.source_state.best_finalized_peer_at_best_self = data.target_state.best_finalized_self;
-				}
-				// if target has received messages batch => increase blocks so that confirmations may be sent
-				if data.target_latest_received_nonce == 4
-					|| data.target_latest_received_nonce == 8
-					|| data.target_latest_received_nonce == 10
-				{
-					data.target_state.best_self =
-						HeaderId(data.target_state.best_self.0 + 1, data.target_state.best_self.0 + 1);
-					data.target_state.best_finalized_self = data.target_state.best_self;
-					data.source_state.best_self =
-						HeaderId(data.source_state.best_self.0 + 1, data.source_state.best_self.0 + 1);
-					data.source_state.best_finalized_self = data.source_state.best_self;
 				}
 				// if source has received all messages receiving confirmations => stop
 				if data.source_latest_confirmed_received_nonce == 10 {

bridges/relays/messages/src/message_race_delivery.rs

@@ -292,7 +292,16 @@ impl<P: MessageLane> RaceStrategy<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>, P::M
 	}
 
 	fn required_source_header_at_target(&self, current_best: &SourceHeaderIdOf<P>) -> Option<SourceHeaderIdOf<P>> {
-		self.strategy.required_source_header_at_target(current_best)
+		let header_required_for_messages_delivery = self.strategy.required_source_header_at_target(current_best);
+		let header_required_for_reward_confirmations_delivery =
+			self.latest_confirmed_nonces_at_source.back().map(|(id, _)| id.clone());
+		match (
+			header_required_for_messages_delivery,
+			header_required_for_reward_confirmations_delivery,
+		) {
+			(Some(id1), Some(id2)) => Some(if id1.0 > id2.0 { id1 } else { id2 }),
+			(a, b) => a.or(b),
+		}
 	}
 
 	fn best_at_source(&self) -> Option<MessageNonce> {
@@ -876,4 +885,46 @@ mod tests {
 			Some(((20..=23), proof_parameters(true, 4)))
 		);
 	}
+
+	#[test]
+	fn source_header_is_requied_when_confirmations_are_required() {
+		// let's prepare situation when:
+		// - all messages [20; 23] have been generated at source block#1;
+		let (mut state, mut strategy) = prepare_strategy();
+		// - messages [20; 21] have been delivered, but messages [11; 20] can't be delivered because of unrewarded
+		//   relayers vector capacity;
+		strategy.max_unconfirmed_nonces_at_target = 2;
+		assert_eq!(
+			strategy.select_nonces_to_deliver(&state),
+			Some(((20..=21), proof_parameters(false, 2)))
+		);
+		strategy.finalized_target_nonces_updated(
+			TargetClientNonces {
+				latest_nonce: 21,
+				nonces_data: DeliveryRaceTargetNoncesData {
+					confirmed_nonce: 19,
+					unrewarded_relayers: UnrewardedRelayersState {
+						unrewarded_relayer_entries: 2,
+						messages_in_oldest_entry: 2,
+						total_messages: 2,
+					},
+				},
+			},
+			&mut state,
+		);
+		assert_eq!(strategy.select_nonces_to_deliver(&state), None);
+		// - messages [1; 10] receiving confirmation has been delivered at source block#2;
+		strategy.source_nonces_updated(
+			header_id(2),
+			SourceClientNonces {
+				new_nonces: BTreeMap::new(),
+				confirmed_nonce: Some(21),
+			},
+		);
+		// - so now we'll need to relay source block#11 to be able to accept messages [11; 20].
+		assert_eq!(
+			strategy.required_source_header_at_target(&header_id(1)),
+			Some(header_id(2))
+		);
+	}
 }