: FinalitySourceClient,
P: FinalitySyncPipeline,
{
// search for mandatory header first
let mandatory_source_header_number = find_mandatory_header_in_range(finality_source, range).await?;
// if there are no mandatory headers - we have nothing to do
let mandatory_source_header_number = match mandatory_source_header_number {
Some(mandatory_source_header_number) => mandatory_source_header_number,
None => return Ok(false),
};
// `find_mandatory_header` call may take a while => check if `required_header_number` is still
// less than our `mandatory_source_header_number` before logging anything
let mut required_header_number = required_header_number.lock().await;
if *required_header_number >= mandatory_source_header_number {
return Ok(false);
}
log::trace!(
target: "bridge",
"Too many {} headers missing at target in {} relay ({} vs {}). Going to sync up to the mandatory {}",
SourceChain::NAME,
relay_task_name,
best_finalized_source_header_at_target,
range.1,
mandatory_source_header_number,
);
*required_header_number = mandatory_source_header_number;
Ok(true)
}
/// Read best finalized source block number from source client.
///
/// Returns `None` if we have failed to read the number.
async fn best_finalized_source_header_at_source(
finality_source: &SubstrateFinalitySource,
relay_task_name: &str,
) -> Result
where
SubstrateFinalitySource: FinalitySourceClient,
P: FinalitySyncPipeline,
{
finality_source
.on_chain_best_finalized_block_number()
.await
.map_err(|error| {
log::error!(
target: "bridge",
"Failed to read best finalized source header from source in {} relay: {:?}",
relay_task_name,
error,
);
error
})
}
/// Read best finalized source block number from target client.
///
/// Returns `None` if we have failed to read the number.
async fn best_finalized_source_header_at_target(
finality_target: &SubstrateFinalityTarget,
relay_task_name: &str,
) -> Result as RelayClient>::Error>
where
SubstrateFinalityTarget: FinalityTargetClient,
P: FinalitySyncPipeline,
{
finality_target
.best_finalized_source_block_number()
.await
.map_err(|error| {
log::error!(
target: "bridge",
"Failed to read best finalized source header from target in {} relay: {:?}",
relay_task_name,
error,
);
error
})
}
/// Read first mandatory header in given inclusive range.
///
/// Returns `Ok(None)` if there were no mandatory headers in the range.
async fn find_mandatory_header_in_range(
finality_source: &SubstrateFinalitySource,
range: (SourceChain::BlockNumber, SourceChain::BlockNumber),
) -> Result, relay_substrate_client::Error>
where
SubstrateFinalitySource: FinalitySourceClient,
P: FinalitySyncPipeline,
{
let mut current = range.0;
while current <= range.1 {
let header: SyncHeader = finality_source.client().header_by_number(current).await?.into();
if header.is_mandatory() {
return Ok(Some(current));
}
current += One::one();
}
Ok(None)
}
/// On-demand headers relay task name.
fn on_demand_headers_relay_name() -> String {
format!("on-demand-{}-to-{}", SourceChain::NAME, TargetChain::NAME)
}
#[cfg(test)]
mod tests {
use super::*;
type TestChain = relay_millau_client::Millau;
const AT_SOURCE: Option = Some(10);
const AT_TARGET: Option = Some(1);
#[async_std::test]
async fn mandatory_headers_scan_range_selects_range_if_too_many_headers_are_missing() {
assert_eq!(
mandatory_headers_scan_range::(AT_SOURCE, AT_TARGET, 5, &Arc::new(Mutex::new(0))).await,
Some((AT_TARGET.unwrap() + 1, AT_SOURCE.unwrap())),
);
}
#[async_std::test]
async fn mandatory_headers_scan_range_selects_nothing_if_enough_headers_are_relayed() {
assert_eq!(
mandatory_headers_scan_range::(AT_SOURCE, AT_TARGET, 10, &Arc::new(Mutex::new(0))).await,
None,
);
}
}