feat: add metric for observed block height of checkpoints (#7146)

Author: kamiyaa

rust/main/agents/validator/src/submit.rs

@@ -128,8 +128,7 @@ impl ValidatorSubmitter {
             .await;
 
             self.metrics
-                .latest_checkpoint_observed
-                .set(latest_checkpoint.index as i64);
+                .set_latest_checkpoint_observed(&latest_checkpoint);
 
             if should_log_checkpoint_info() {
                 info!(
@@ -476,6 +475,7 @@ fn tree_exceeds_checkpoint(checkpoint: &Checkpoint, tree: &IncrementalMerkle) ->
 
 #[derive(Clone)]
 pub(crate) struct ValidatorSubmitterMetrics {
+    latest_checkpoint_observed_block_height: IntGauge,
     latest_checkpoint_observed: IntGauge,
     latest_checkpoint_processed: IntGauge,
     backfill_complete: IntGauge,
@@ -486,6 +486,9 @@ impl ValidatorSubmitterMetrics {
     pub fn new(metrics: &CoreMetrics, mailbox_chain: &HyperlaneDomain) -> Self {
         let chain_name = mailbox_chain.name();
         Self {
+            latest_checkpoint_observed_block_height: metrics
+                .latest_checkpoint()
+                .with_label_values(&["validator_observed_block_height", chain_name]),
             latest_checkpoint_observed: metrics
                 .latest_checkpoint()
                 .with_label_values(&["validator_observed", chain_name]),
@@ -498,6 +501,31 @@ impl ValidatorSubmitterMetrics {
                 .with_label_values(&[chain_name]),
         }
     }
+
+    fn set_latest_checkpoint_observed(&self, checkpoint: &CheckpointAtBlock) {
+        let prev_checkpoint_index = self.latest_checkpoint_observed.get();
+
+        if prev_checkpoint_index > checkpoint.index as i64 {
+            tracing::warn!(
+                ?checkpoint,
+                prev_checkpoint_index,
+                checkpoint_index=checkpoint.index, "Observed a checkpoint with index that is lower than previous checkpoint. Did a reorg occur?");
+        }
+        self.latest_checkpoint_observed.set(checkpoint.index as i64);
+
+        if let Some(block_height) = checkpoint.block_height {
+            let block_height = block_height as i64;
+            let prev_block_height = self.latest_checkpoint_observed_block_height.get();
+            if prev_block_height > block_height {
+                tracing::warn!(
+                    ?checkpoint,
+                    prev_block_height,
+                    block_height, "Observed a checkpoint with block height that is lower than previous checkpoint. Did a reorg occur?");
+            }
+            self.latest_checkpoint_observed_block_height
+                .set(block_height);
+        }
+    }
 }
 
 #[cfg(test)]