Messages relay fixes (#2073)

* removed obsolete check that is superseded by the unblock checks below

* if messages race transaction submit has failed, do not restart loop. Instead, wait for new best nonces from target node and retry selection. That's because submit has probably failed because other relayer has submitted same nonces

* reset nonces_to_submit and nonces_submitted if at least one of selected/submitted nonces is already at target

* removed extra check

Author: svyatonik

relays/messages/src/message_race_delivery.rs

@@ -343,38 +343,14 @@ where
 		// There's additional condition in the message delivery race: target would reject messages
 		// if there are too much unconfirmed messages at the inbound lane.
 
-		// The receiving race is responsible to deliver confirmations back to the source chain. So
-		// if there's a lot of unconfirmed messages, let's wait until it'll be able to do its job.
-		let latest_received_nonce_at_target = target_nonces.latest_nonce;
-		let confirmations_missing =
-			latest_received_nonce_at_target.checked_sub(latest_confirmed_nonce_at_source);
-		match confirmations_missing {
-			Some(confirmations_missing)
-				if confirmations_missing >= self.max_unconfirmed_nonces_at_target =>
-			{
-				log::debug!(
-					target: "bridge",
-					"Cannot deliver any more messages from {} to {}. Too many unconfirmed nonces \
-					at target: target.latest_received={:?}, source.latest_confirmed={:?}, max={:?}",
-					MessageDeliveryRace::<P>::source_name(),
-					MessageDeliveryRace::<P>::target_name(),
-					latest_received_nonce_at_target,
-					latest_confirmed_nonce_at_source,
-					self.max_unconfirmed_nonces_at_target,
-				);
-
-				return None
-			},
-			_ => (),
-		}
-
 		// Ok - we may have new nonces to deliver. But target may still reject new messages, because
 		// we haven't notified it that (some) messages have been confirmed. So we may want to
 		// include updated `source.latest_confirmed` in the proof.
 		//
 		// Important note: we're including outbound state lane proof whenever there are unconfirmed
 		// nonces on the target chain. Other strategy is to include it only if it's absolutely
 		// necessary.
+		let latest_received_nonce_at_target = target_nonces.latest_nonce;
 		let latest_confirmed_nonce_at_target = target_nonces.nonces_data.confirmed_nonce;
 		let outbound_state_proof_required =
 			latest_confirmed_nonce_at_target < latest_confirmed_nonce_at_source;
@@ -585,6 +561,11 @@ where
 		self.strategy.source_nonces_updated(at_block, nonces)
 	}
 
+	fn reset_best_target_nonces(&mut self) {
+		self.target_nonces = None;
+		self.strategy.reset_best_target_nonces();
+	}
+
 	fn best_target_nonces_updated<RS: RaceState<SourceHeaderIdOf<P>, TargetHeaderIdOf<P>>>(
 		&mut self,
 		nonces: TargetClientNonces<DeliveryRaceTargetNoncesData>,
@@ -808,22 +789,6 @@ mod tests {
 		);
 	}
 
-	#[async_std::test]
-	async fn message_delivery_strategy_selects_nothing_if_too_many_confirmations_missing() {
-		let (state, mut strategy) = prepare_strategy();
-
-		// if there are already `max_unconfirmed_nonces_at_target` messages on target,
-		// we need to wait until confirmations will be delivered by receiving race
-		strategy.latest_confirmed_nonces_at_source = vec![(
-			header_id(1),
-			strategy.target_nonces.as_ref().unwrap().latest_nonce -
-				strategy.max_unconfirmed_nonces_at_target,
-		)]
-		.into_iter()
-		.collect();
-		assert_eq!(strategy.select_nonces_to_deliver(state).await, None);
-	}
-
 	#[async_std::test]
 	async fn message_delivery_strategy_includes_outbound_state_proof_when_new_nonces_are_available()
 	{

relays/messages/src/message_race_loop.rs

@@ -195,6 +195,9 @@ pub trait RaceStrategy<SourceHeaderId, TargetHeaderId, Proof>: Debug {
 		at_block: SourceHeaderId,
 		nonces: SourceClientNonces<Self::SourceNoncesRange>,
 	);
+	/// Called when we want to wait until next `best_target_nonces_updated` before selecting
+	/// any nonces for delivery.
+	fn reset_best_target_nonces(&mut self);
 	/// Called when best nonces are updated at target node of the race.
 	fn best_target_nonces_updated<RS: RaceState<SourceHeaderId, TargetHeaderId>>(
 		&mut self,
@@ -542,14 +545,22 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 							P::target_name(),
 						);
 
-						race_state.reset_nonces_to_submit();
 						race_state.nonces_submitted = Some(artifacts.nonces);
 						target_tx_tracker.set(artifacts.tx_tracker.wait().fuse());
 					},
 					&mut target_go_offline_future,
 					async_std::task::sleep,
 					|| format!("Error submitting proof {}", P::target_name()),
-				).fail_if_error(FailedClient::Target).map(|_| true)?;
+				).fail_if_connection_error(FailedClient::Target)?;
+
+				// in any case - we don't need to retry submitting the same nonces again until
+				// we read nonces from the target client
+				race_state.reset_nonces_to_submit();
+				// if we have failed to submit transaction AND that is not the connection issue,
+				// then we need to read best target nonces before selecting nonces again
+				if !target_client_is_online {
+					strategy.reset_best_target_nonces();
+				}
 			},
 			target_transaction_status = target_tx_tracker => {
 				match (target_transaction_status, race_state.nonces_submitted.as_ref()) {
@@ -699,7 +710,13 @@ pub async fn run<P: MessageRace, SC: SourceClient<P>, TC: TargetClient<P>>(
 						.fuse(),
 				);
 			} else if let Some(source_required_header) = source_required_header.clone() {
-				log::debug!(target: "bridge", "Going to require {} header {:?} at {}", P::source_name(), source_required_header, P::target_name());
+				log::debug!(
+					target: "bridge",
+					"Going to require {} header {:?} at {}",
+					P::source_name(),
+					source_required_header,
+					P::target_name(),
+				);
 				target_require_source_header
 					.set(race_target.require_source_header(source_required_header).fuse());
 			} else if target_best_nonces_required {

relays/messages/src/message_race_strategy.rs

@@ -254,6 +254,10 @@ impl<
 		)
 	}
 
+	fn reset_best_target_nonces(&mut self) {
+		self.best_target_nonce = None;
+	}
+
 	fn best_target_nonces_updated<
 		RS: RaceState<
 			HeaderId<SourceHeaderHash, SourceHeaderNumber>,
@@ -266,19 +270,37 @@ impl<
 	) {
 		let nonce = nonces.latest_nonce;
 
+		// if **some** of nonces that we have selected to submit already present at the
+		// target chain => select new nonces
 		let need_to_select_new_nonces = race_state
 			.nonces_to_submit()
-			.map(|nonces| *nonces.end() <= nonce)
+			.map(|nonces| nonce >= *nonces.start())
 			.unwrap_or(false);
 		if need_to_select_new_nonces {
+			log::trace!(
+				target: "bridge",
+				"Latest nonce at target is {}. Clearing nonces to submit: {:?}",
+				nonce,
+				race_state.nonces_to_submit(),
+			);
+
 			race_state.reset_nonces_to_submit();
 		}
 
+		// if **some** of nonces that we have submitted already present at the
+		// target chain => select new nonces
 		let need_new_nonces_to_submit = race_state
 			.nonces_submitted()
-			.map(|nonces| *nonces.end() <= nonce)
+			.map(|nonces| nonce >= *nonces.start())
 			.unwrap_or(false);
 		if need_new_nonces_to_submit {
+			log::trace!(
+				target: "bridge",
+				"Latest nonce at target is {}. Clearing submitted nonces: {:?}",
+				nonce,
+				race_state.nonces_submitted(),
+			);
+
 			race_state.reset_nonces_submitted();
 		}
 
@@ -415,21 +437,31 @@ mod tests {
 		let mut state = TestRaceStateImpl::default();
 		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_to_submit = Some((header_id(1), 5..=10, (5..=10, None)));
-		strategy.best_target_nonces_updated(target_nonces(7), &mut state);
+		// we are going to submit 5..=10, so having latest nonce 4 at target is fine
+		strategy.best_target_nonces_updated(target_nonces(4), &mut state);
 		assert!(state.nonces_to_submit.is_some());
-		strategy.best_target_nonces_updated(target_nonces(10), &mut state);
-		assert!(state.nonces_to_submit.is_none());
+		// any nonce larger than 4 invalidates the `nonces_to_submit`
+		for nonce in 5..=11 {
+			state.nonces_to_submit = Some((header_id(1), 5..=10, (5..=10, None)));
+			strategy.best_target_nonces_updated(target_nonces(nonce), &mut state);
+			assert!(state.nonces_to_submit.is_none());
+		}
 	}
 
 	#[test]
 	fn submitted_nonces_are_dropped_on_target_nonce_update() {
 		let mut state = TestRaceStateImpl::default();
 		let mut strategy = BasicStrategy::<TestMessageLane>::new();
 		state.nonces_submitted = Some(5..=10);
-		strategy.best_target_nonces_updated(target_nonces(7), &mut state);
+		// we have submitted 5..=10, so having latest nonce 4 at target is fine
+		strategy.best_target_nonces_updated(target_nonces(4), &mut state);
 		assert!(state.nonces_submitted.is_some());
-		strategy.best_target_nonces_updated(target_nonces(10), &mut state);
-		assert!(state.nonces_submitted.is_none());
+		// any nonce larger than 4 invalidates the `nonces_submitted`
+		for nonce in 5..=11 {
+			state.nonces_submitted = Some(5..=10);
+			strategy.best_target_nonces_updated(target_nonces(nonce), &mut state);
+			assert!(state.nonces_submitted.is_none());
+		}
 	}
 
 	#[async_std::test]