@ -1,12 +1,16 @@
use tracing ::{ debug , info , info_span , warn } ;
use tracing ::{ instrument ::Instrumented , Instrument } ;
use std ::time ::{ Duration , Instant } ;
use tracing ::{ debug , info , instrument , warn } ;
use hyperlane_core ::{
Indexer , KnownHyperlaneDomain , ListValidity , MailboxIndexer , SyncBlockRangeCursor ,
utils ::fmt_sync_time , Indexer , KnownHyperlaneDomain , ListValidity , MailboxIndexer ,
SyncBlockRangeCursor ,
} ;
use crate ::contract_sync ::last_message ::validate_message_continuity ;
use crate ::{ contract_sync ::schema ::MailboxContractSyncDB , ContractSync } ;
use crate ::{
contract_sync ::{ last_message ::validate_message_continuity , schema ::MailboxContractSyncDB } ,
ContractSync ,
} ;
const MESSAGES_LABEL : & str = "messages" ;
@ -15,37 +19,32 @@ where
I : MailboxIndexer + Clone + ' static ,
{
/// Sync dispatched messages
pub fn sync_dispatched_messages (
& self ,
) -> Instrumented < tokio ::task ::JoinHandle < eyre ::Result < ( ) > > > {
let span = info_span ! ( "MessageContractSync" ) ;
let db = self . db . clone ( ) ;
let indexer = self . indexer . clone ( ) ;
let chain_name = self . domain . to_string ( ) ;
#[ instrument(name = " MessageContractSync " , skip(self)) ]
pub ( crate ) async fn sync_dispatched_messages ( & self ) -> eyre ::Result < ( ) > {
let chain_name = self . domain . as_ref ( ) ;
let indexed_height = self
. metrics
. indexed_height
. with_label_values ( & [ MESSAGES_LABEL , & chain_name ] ) ;
. with_label_values ( & [ MESSAGES_LABEL , chain_name ] ) ;
let stored_messages = self
. metrics
. stored_events
. with_label_values ( & [ MESSAGES_LABEL , & chain_name ] ) ;
. with_label_values ( & [ MESSAGES_LABEL , chain_name ] ) ;
let missed_messages = self
. metrics
. missed_events
. with_label_values ( & [ MESSAGES_LABEL , & chain_name ] ) ;
. with_label_values ( & [ MESSAGES_LABEL , chain_name ] ) ;
let message_nonce = self . metrics . message_nonce . clone ( ) ;
let cursor = {
let config_initial_height = self . index_settings . from ( ) ;
let initial_height = db
let initial_height = self
. db
. retrieve_latest_valid_message_range_start_block ( )
. map_or ( config_initial_height , | b | b + 1 ) ;
create_cursor (
indexer . clone ( ) ,
self . indexer . clone ( ) ,
self . index_settings . chunk_size ( ) ,
initial_height ,
)
@ -98,107 +97,142 @@ where
// Note this means we only handle this case upon observing messages in some
// range [C,D] that indicate a previously indexed range may have
// missed some messages.
tokio ::spawn ( async move {
let mut cursor = cursor . await ? ;
let mut cursor = cursor . await ? ;
let start_block = cursor . current_position ( ) ;
let mut last_valid_range_start_block = start_block ;
info ! (
from = start_block ,
"Resuming indexer from latest valid message range start block"
) ;
indexed_height . set ( start_block as i64 ) ;
let mut last_logged_time : Option < Instant > = None ;
let mut should_log_checkpoint_info = | | {
if last_logged_time . is_none ( )
| | last_logged_time . unwrap ( ) . elapsed ( ) > Duration ::from_secs ( 30 )
{
last_logged_time = Some ( Instant ::now ( ) ) ;
true
} else {
false
}
} ;
loop {
let start_block = cursor . current_position ( ) ;
let mut last_valid_range_start_block = start_block ;
info ! ( from = start_block , "Resuming indexer from latest valid message range start block" ) ;
indexed_height . set ( start_block as i64 ) ;
loop {
let start_block = cursor . current_position ( ) ;
let ( from , to ) = match cursor . next_range ( ) . await {
Ok ( range ) = > range ,
Err ( err ) = > {
warn ! ( error = % err , "Failed to get next block range" ) ;
continue ;
}
} ;
let mut sorted_messages : Vec < _ > = indexer
. fetch_sorted_messages ( from , to )
. await ?
. into_iter ( )
. map ( | ( msg , _ ) | msg )
. collect ( ) ;
debug ! ( from , to , message_count = sorted_messages . len ( ) , "Indexed block range" ) ;
// Get the latest known nonce. All messages whose indices are <= this index
// have been stored in the DB.
let last_nonce = db . retrieve_latest_nonce ( ) ? ;
// Filter out any messages that have already been successfully indexed and stored.
// This is necessary if we're re-indexing blocks in hope of finding missing messages.
if let Some ( min_nonce ) = last_nonce {
sorted_messages . retain ( | m | m . nonce > min_nonce ) ;
}
let Ok ( ( from , to , eta ) ) = cursor . next_range ( ) . await else { continue } ;
let mut sorted_messages : Vec < _ > = self
. indexer
. fetch_sorted_messages ( from , to )
. await ?
. into_iter ( )
. map ( | ( msg , _ ) | msg )
. collect ( ) ;
if should_log_checkpoint_info ( ) {
info ! (
from ,
to ,
distance_from_tip = cursor . distance_from_tip ( ) ,
estimated_time_to_sync = fmt_sync_time ( eta ) ,
message_count = sorted_messages . len ( ) ,
"Indexed block range"
) ;
} else {
debug ! (
from ,
to ,
distance_from_tip = cursor . distance_from_tip ( ) ,
estimated_time_to_sync = fmt_sync_time ( eta ) ,
message_count = sorted_messages . len ( ) ,
"Indexed block range"
) ;
}
debug ! ( from , to , message_count = sorted_messages . len ( ) , "Filtered any messages already indexed" ) ;
// Get the latest known nonce. All messages whose indices are <= this index
// have been stored in the DB.
let last_nonce = self . db . retrieve_latest_nonce ( ) ? ;
// Ensure the sorted messages are a valid continuation of last_nonce
match validate_message_continuity ( last_nonce , & sorted_messages . iter ( ) . collect ::< Vec < _ > > ( ) ) {
ListValidity ::Valid = > {
// Store messages
let max_nonce_of_batch = db . store_messages ( & sorted_messages ) ? ;
// Filter out any messages that have already been successfully indexed and
// stored. This is necessary if we're re-indexing blocks in hope of
// finding missing messages.
if let Some ( min_nonce ) = last_nonce {
sorted_messages . retain ( | m | m . nonce > min_nonce ) ;
}
// Report amount of messages stored into db
stored_messages . inc_by ( sorted_messages . len ( ) as u64 ) ;
debug ! (
from ,
to ,
message_count = sorted_messages . len ( ) ,
"Filtered any messages already indexed"
) ;
// Report latest nonce to gauge by dst
for msg in sorted_messages . iter ( ) {
let dst = KnownHyperlaneDomain ::try_from ( msg . destination ) . map ( | d | d . as_str ( ) ) . unwrap_or ( "unknown" ) ;
message_nonce
. with_label_values ( & [ "dispatch" , & chain_name , dst ] )
. set ( max_nonce_of_batch as i64 ) ;
}
// Ensure the sorted messages are a valid continuation of last_nonce
match validate_message_continuity (
last_nonce ,
& sorted_messages . iter ( ) . collect ::< Vec < _ > > ( ) ,
) {
ListValidity ::Valid = > {
// Store messages
let max_nonce_of_batch = self . db . store_messages ( & sorted_messages ) ? ;
// Report amount of messages stored into db
stored_messages . inc_by ( sorted_messages . len ( ) as u64 ) ;
// Report latest nonce to gauge by dst
for msg in sorted_messages . iter ( ) {
let dst = KnownHyperlaneDomain ::try_from ( msg . destination )
. map ( | d | d . as_str ( ) )
. unwrap_or ( "unknown" ) ;
message_nonce
. with_label_values ( & [ "dispatch" , chain_name , dst ] )
. set ( max_nonce_of_batch as i64 ) ;
}
// Update the latest valid start block.
db . store_latest_valid_message_range_start_block ( from ) ? ;
last_valid_range_start_block = from ;
// Update the latest valid start block.
self . db . store_latest_valid_message_range_start_block ( from ) ? ;
last_valid_range_start_block = from ;
// Move forward to the next height
indexed_height . set ( to as i64 ) ;
}
// The index of the first message in sorted_messages is not the
// `last_nonce+1`.
ListValidity ::InvalidContinuation = > {
missed_messages . inc ( ) ;
warn ! (
last_nonce = ? last_nonce ,
start_block = from ,
end_block = to ,
last_valid_range_start_block ,
"Found invalid continuation in range. Re-indexing from the start block of the last successful range." ,
) ;
cursor . backtrack ( last_valid_range_start_block ) ;
indexed_height . set ( last_valid_range_start_block as i64 ) ;
}
ListValidity ::ContainsGaps = > {
missed_messages . inc ( ) ;
cursor . backtrack ( start_block ) ;
warn ! (
last_nonce = ? last_nonce ,
start_block = from ,
end_block = to ,
"Found gaps in the messages in range, re-indexing the same range." ,
) ;
}
ListValidity ::Empty = > {
// Continue if no messages found.
// We don't update last_valid_range_start_block because we cannot extrapolate
// if the range was correctly indexed if there are no messages to observe their
// indices.
indexed_height . set ( to as i64 ) ;
}
} ;
}
} )
. instrument ( span )
// Move forward to the next height
indexed_height . set ( to as i64 ) ;
}
// The index of the first message in sorted_messages is not the
// `last_nonce+1`.
ListValidity ::InvalidContinuation = > {
missed_messages . inc ( ) ;
warn ! (
last_nonce = ? last_nonce ,
start_block = from ,
end_block = to ,
last_valid_range_start_block ,
"Found invalid continuation in range. Re-indexing from the start block of the last successful range." ,
) ;
cursor . backtrack ( last_valid_range_start_block ) ;
indexed_height . set ( last_valid_range_start_block as i64 ) ;
}
ListValidity ::ContainsGaps = > {
missed_messages . inc ( ) ;
cursor . backtrack ( start_block ) ;
warn ! (
last_nonce = ? last_nonce ,
start_block = from ,
end_block = to ,
"Found gaps in the messages in range, re-indexing the same range." ,
) ;
}
ListValidity ::Empty = > {
// Continue if no messages found.
// We don't update last_valid_range_start_block because we cannot extrapolate
// if the range was correctly indexed if there are no messages to observe their
// indices.
indexed_height . set ( to as i64 ) ;
}
} ;
}
}
}
@ -321,7 +355,11 @@ mod test {
. expect__next_range ( )
. times ( 1 )
. in_sequence ( & mut seq )
. return_once ( | | Box ::pin ( async { Ok ( ( $expected_from , $expected_to ) ) } ) ) ;
. return_once ( | | {
Box ::pin ( async {
Ok ( ( $expected_from , $expected_to , Duration ::from_secs ( 0 ) ) )
} )
} ) ;
mock_indexer
. expect__fetch_sorted_messages ( )
. times ( 1 )
@ -382,7 +420,7 @@ mod test {
// this sleep should be longer than the test timeout since we don't actually
// want to yield any more values at this point.
sleep ( Duration ::from_secs ( 100 ) ) . await ;
Ok ( ( 161 , 161 ) )
Ok ( ( 161 , 161 , Duration ::from_secs ( 0 ) ) )
} )
} ) ;
}
Mattie Conover
2 years ago
committed by
GitHub