feat(validator): record reorg logs (#7238)

mbreithecker · web-flow · commit c71e2525d70e · 2025-10-24T07:31:13.000Z
diff --git a/rust/main/agents/validator/src/reorg_reporter.rs b/rust/main/agents/validator/src/reorg_reporter.rs
@@ -3,14 +3,16 @@ use std::fmt::Debug;
 use std::sync::Arc;
 
 use async_trait::async_trait;
+use ethers::utils::keccak256;
 use futures_util::future::join_all;
-use tracing::info;
+use serde::Serialize;
+use tracing::{info, warn};
 use url::Url;
 
 use hyperlane_base::settings::ChainConnectionConf;
-use hyperlane_base::CoreMetrics;
+use hyperlane_base::{CheckpointSyncer, CoreMetrics};
 use hyperlane_core::rpc_clients::call_and_retry_indefinitely;
-use hyperlane_core::{HyperlaneDomain, MerkleTreeHook, ReorgPeriod};
+use hyperlane_core::{CheckpointAtBlock, HyperlaneDomain, MerkleTreeHook, ReorgPeriod, H256};
 use hyperlane_ethereum::RpcConnectionConf;
 
 use crate::settings::ValidatorSettings;
@@ -26,9 +28,49 @@ pub struct LatestCheckpointReorgReporter {
     merkle_tree_hooks: HashMap<Url, Arc<dyn MerkleTreeHook>>,
 }
 
+#[derive(Serialize)]
+struct ReorgReportRpcResponse {
+    rpc_url_hash: H256,
+    rpc_host_hash: H256,
+    height: Option<u64>,
+    reorg_period: Option<ReorgPeriod>,
+    merkle_root_index: u32,
+    merkle_root_hash: H256,
+    timestamp: String,
+}
+
+impl ReorgReportRpcResponse {
+    fn new(
+        url: Url,
+        latest_checkpoint: CheckpointAtBlock,
+        height: Option<u64>,
+        reorg_period: Option<ReorgPeriod>,
+    ) -> Self {
+        ReorgReportRpcResponse {
+            rpc_host_hash: H256::from_slice(&keccak256(url.host_str().unwrap_or("").as_bytes())),
+            rpc_url_hash: H256::from_slice(&keccak256(url.as_str().as_bytes())),
+            height,
+            reorg_period,
+            merkle_root_hash: latest_checkpoint.checkpoint.root,
+            merkle_root_index: latest_checkpoint.checkpoint.index,
+            timestamp: chrono::Utc::now().to_rfc3339(),
+        }
+    }
+}
+
 #[async_trait]
 impl ReorgReporter for LatestCheckpointReorgReporter {
     async fn report_at_block(&self, height: u64) {
+        self.report_at_block(height).await;
+    }
+
+    async fn report_with_reorg_period(&self, reorg_period: &ReorgPeriod) {
+        self.report_with_reorg_period(reorg_period).await;
+    }
+}
+
+impl LatestCheckpointReorgReporter {
+    async fn report_at_block(&self, height: u64) -> Vec<ReorgReportRpcResponse> {
         info!(?height, "Reporting latest checkpoint on reorg");
         let mut futures = vec![];
         for (url, merkle_tree_hook) in &self.merkle_tree_hooks {
@@ -42,15 +84,19 @@ impl ReorgReporter for LatestCheckpointReorgReporter {
                 .await;
 
                 info!(url = ?url.clone(), ?height, ?latest_checkpoint, "Report latest checkpoint on reorg");
+                ReorgReportRpcResponse::new(url.clone(), latest_checkpoint, Some(height), None)
             };
 
             futures.push(future);
         }
 
-        join_all(futures).await;
+        join_all(futures).await
     }
 
-    async fn report_with_reorg_period(&self, reorg_period: &ReorgPeriod) {
+    async fn report_with_reorg_period(
+        &self,
+        reorg_period: &ReorgPeriod,
+    ) -> Vec<ReorgReportRpcResponse> {
         info!(?reorg_period, "Reporting latest checkpoint on reorg");
         let mut futures = vec![];
         for (url, merkle_tree_hook) in &self.merkle_tree_hooks {
@@ -63,12 +109,18 @@ impl ReorgReporter for LatestCheckpointReorgReporter {
                 .await;
 
                 info!(url = ?url.clone(), ?reorg_period, ?latest_checkpoint, "Report latest checkpoint on reorg");
+                ReorgReportRpcResponse::new(
+                    url.clone(),
+                    latest_checkpoint,
+                    None,
+                    Some(reorg_period.clone()),
+                )
             };
 
             futures.push(future);
         }
 
-        join_all(futures).await;
+        join_all(futures).await
     }
 }
 
@@ -142,13 +194,11 @@ impl LatestCheckpointReorgReporter {
                     Starknet(updated_conn)
                 })
             }
-            Radix(conn) => {
-                Self::map_urls_to_connections(conn.gateway.clone(), conn, |conn, url| {
-                    let mut updated_conn = conn.clone();
-                    updated_conn.gateway = vec![url];
-                    Radix(updated_conn)
-                })
-            }
+            Radix(conn) => Self::map_urls_to_connections(conn.core.clone(), conn, |conn, url| {
+                let mut updated_conn = conn.clone();
+                updated_conn.core = vec![url];
+                Radix(updated_conn)
+            }),
         };
 
         chain_conn_confs
@@ -180,3 +230,62 @@ impl LatestCheckpointReorgReporter {
             .collect()
     }
 }
+
+#[derive(Debug)]
+pub struct LatestCheckpointReorgReporterWithStorageWriter {
+    /// `LatestCheckpointReorgReporterWithStorageWriter` is an extension to
+    /// `LatestCheckpointReorgReporter`
+    latest_checkpoint_reorg_reporter: LatestCheckpointReorgReporter,
+
+    /// Currently, the storage abstraction is tied to the checkpoint syncer, which is why
+    /// it is used here.
+    storage_writer: Arc<dyn CheckpointSyncer>,
+}
+
+#[async_trait]
+impl ReorgReporter for LatestCheckpointReorgReporterWithStorageWriter {
+    async fn report_at_block(&self, height: u64) {
+        let logs = self
+            .latest_checkpoint_reorg_reporter
+            .report_at_block(height)
+            .await;
+        self.submit_to_storage_writer(&logs).await;
+    }
+
+    async fn report_with_reorg_period(&self, reorg_period: &ReorgPeriod) {
+        let logs = self
+            .latest_checkpoint_reorg_reporter
+            .report_with_reorg_period(reorg_period)
+            .await;
+        self.submit_to_storage_writer(&logs).await;
+    }
+}
+
+impl LatestCheckpointReorgReporterWithStorageWriter {
+    pub(crate) async fn from_settings_with_storage_writer(
+        settings: &ValidatorSettings,
+        metrics: &CoreMetrics,
+        storage_writer: Arc<dyn CheckpointSyncer>,
+    ) -> eyre::Result<Self> {
+        Ok(LatestCheckpointReorgReporterWithStorageWriter {
+            latest_checkpoint_reorg_reporter: LatestCheckpointReorgReporter::from_settings(
+                settings, metrics,
+            )
+            .await?,
+            storage_writer,
+        })
+    }
+
+    async fn submit_to_storage_writer(&self, storage_logs_entries: &Vec<ReorgReportRpcResponse>) {
+        let json_string = serde_json::to_string_pretty(storage_logs_entries).unwrap_or_else(|e| {
+            warn!("Error serializing json: {}", e);
+            String::from("{\"error\": \"Error formatting the string\"}")
+        });
+        self.storage_writer
+            .write_reorg_rpc_responses(json_string)
+            .await
+            .unwrap_or_else(|e| {
+                warn!("Error writing checkpoint syncer to reorg log: {}", e);
+            });
+    }
+}
diff --git a/rust/main/agents/validator/src/validator.rs b/rust/main/agents/validator/src/validator.rs
@@ -10,6 +10,7 @@ use itertools::Itertools;
 use serde::Serialize;
 use tokio::{task::JoinHandle, time::sleep};
 use tracing::{error, info, info_span, warn, Instrument};
+use url::Url;
 
 use hyperlane_base::{
     db::{HyperlaneDb, HyperlaneRocksDB, DB},
@@ -27,7 +28,9 @@ use hyperlane_core::{
 };
 use hyperlane_ethereum::{Signers, SingletonSigner, SingletonSignerHandle};
 
-use crate::reorg_reporter::{LatestCheckpointReorgReporter, ReorgReporter};
+use crate::reorg_reporter::{
+    LatestCheckpointReorgReporter, LatestCheckpointReorgReporterWithStorageWriter, ReorgReporter,
+};
 use crate::server::{self as validator_server, merkle_tree_insertions};
 use crate::{
     settings::ValidatorSettings,
@@ -66,9 +69,14 @@ pub struct Validator {
 #[derive(Debug, Serialize)]
 pub struct ValidatorMetadata {
     git_sha: String,
-    rpcs: Vec<H256>,
+    rpcs: Vec<ValidatorMetadataRpcEntry>,
     allows_public_rpcs: bool,
 }
+#[derive(Debug, Serialize)]
+pub struct ValidatorMetadataRpcEntry {
+    url_hash: H256,
+    host_hash: H256,
+}
 
 impl MetadataFromSettings<ValidatorSettings> for ValidatorMetadata {
     /// Create a new instance of the agent metadata from the settings
@@ -77,7 +85,15 @@ impl MetadataFromSettings<ValidatorSettings> for ValidatorMetadata {
         let rpcs = settings
             .rpcs
             .iter()
-            .map(|rpc| H256::from_slice(&keccak256(&rpc.url)))
+            .map(|rpc| ValidatorMetadataRpcEntry {
+                url_hash: H256::from_slice(&keccak256(&rpc.url)),
+                host_hash: H256::from_slice(&keccak256(
+                    Url::parse(&rpc.url)
+                        .ok()
+                        .and_then(|url| url.host_str().map(str::to_string))
+                        .unwrap_or("".to_string()),
+                )),
+            })
             .collect();
         ValidatorMetadata {
             git_sha: git_sha(),
@@ -140,10 +156,21 @@ impl BaseAgent for Validator {
 
         // Be extra sure to panic when checkpoint syncer fails, which indicates
         // a fatal startup error.
-        let checkpoint_syncer = checkpoint_syncer_result
+        let checkpoint_syncer: Arc<dyn CheckpointSyncer> = checkpoint_syncer_result
             .expect("Failed to build checkpoint syncer")
             .into();
 
+        // If checkpoint syncer initialization was successful, use a reorg-reporter which
+        // writes to the storage location in addition to the logs.
+        let reorg_reporter_with_storage_writer =
+            LatestCheckpointReorgReporterWithStorageWriter::from_settings_with_storage_writer(
+                &settings,
+                &metrics,
+                checkpoint_syncer.clone(),
+            )
+            .await?;
+        let reorg_reporter = Arc::new(reorg_reporter_with_storage_writer) as Arc<dyn ReorgReporter>;
+
         let origin_chain_conf = core.settings.chain_setup(&settings.origin_chain)?.clone();
 
         let mailbox = origin_chain_conf.build_mailbox(&metrics).await?;
diff --git a/rust/main/hyperlane-base/src/traits/checkpoint_syncer.rs b/rust/main/hyperlane-base/src/traits/checkpoint_syncer.rs
@@ -1,7 +1,7 @@
 use std::fmt::Debug;
 
 use async_trait::async_trait;
-use eyre::Result;
+use eyre::{Report, Result};
 
 use hyperlane_core::{ReorgEvent, SignedAnnouncement, SignedCheckpointWithMessageId};
 
@@ -43,6 +43,11 @@ pub trait CheckpointSyncer: Debug + Send + Sync {
     /// the validator agent to stop publishing checkpoints. Once any remediation is done, this flag can be reset
     /// to resume operation.
     async fn write_reorg_status(&self, reorg_event: &ReorgEvent) -> Result<()>;
+    /// Writes the provided log message to the storage destination.
+    /// This log is publicly available. It must not contain sensitive information.
+    async fn write_reorg_rpc_responses(&self, _log: String) -> Result<()> {
+        Err(Report::msg("Destination does not support log writing."))
+    }
     /// Read the reorg status of the chain being validated
     async fn reorg_status(&self) -> Result<Option<ReorgEvent>>;
 }
diff --git a/rust/main/hyperlane-base/src/types/gcs_storage.rs b/rust/main/hyperlane-base/src/types/gcs_storage.rs
@@ -17,6 +17,7 @@ const LATEST_INDEX_KEY: &str = "gcsLatestIndexKey";
 const METADATA_KEY: &str = "gcsMetadataKey";
 const ANNOUNCEMENT_KEY: &str = "gcsAnnouncementKey";
 const REORG_FLAG_KEY: &str = "gcsReorgFlagKey";
+const REORG_RPC_RESPONSES_KEY: &str = "gcsReorgRpcResponsesKey";
 
 /// Path to GCS users_secret file
 pub const GCS_USER_SECRET: &str = "GCS_USER_SECRET";
@@ -272,6 +273,12 @@ impl CheckpointSyncer for GcsStorageClient {
         self.upload_and_log(object_name, data).await
     }
 
+    #[instrument(skip(self, log))]
+    async fn write_reorg_rpc_responses(&self, log: String) -> Result<()> {
+        let object_name = REORG_RPC_RESPONSES_KEY;
+        self.upload_and_log(object_name, log.into_bytes()).await
+    }
+
     /// Read the reorg status from this syncer
     #[instrument(skip(self))]
     async fn reorg_status(&self) -> Result<Option<ReorgEvent>> {
diff --git a/rust/main/hyperlane-base/src/types/local_storage.rs b/rust/main/hyperlane-base/src/types/local_storage.rs
@@ -45,6 +45,10 @@ impl LocalStorage {
         self.path.join("reorg_flag.json")
     }
 
+    fn reorg_rpc_responses_path(&self) -> PathBuf {
+        self.path.join("reorg_rpc_responses.json")
+    }
+
     fn metadata_file_path(&self) -> PathBuf {
         self.path.join("metadata_latest.json")
     }
@@ -136,4 +140,12 @@ impl CheckpointSyncer for LocalStorage {
         let reorg = serde_json::from_slice(&data)?;
         Ok(Some(reorg))
     }
+
+    async fn write_reorg_rpc_responses(&self, log: String) -> Result<()> {
+        let path = self.reorg_rpc_responses_path();
+        tokio::fs::write(&path, &log)
+            .await
+            .with_context(|| format!("Writing log to {path:?}"))?;
+        Ok(())
+    }
 }
diff --git a/rust/main/hyperlane-base/src/types/s3_storage.rs b/rust/main/hyperlane-base/src/types/s3_storage.rs
@@ -201,6 +201,10 @@ impl S3Storage {
     fn reorg_flag_key() -> String {
         "reorg_flag.json".to_owned()
     }
+
+    fn reorg_rpc_responses_key() -> String {
+        "reorg_rpc_responses.json".to_owned()
+    }
 }
 
 #[async_trait]
@@ -279,6 +283,12 @@ impl CheckpointSyncer for S3Storage {
         Ok(())
     }
 
+    async fn write_reorg_rpc_responses(&self, reorg_log: String) -> Result<()> {
+        self.write_to_bucket(S3Storage::reorg_rpc_responses_key(), &reorg_log)
+            .await?;
+        Ok(())
+    }
+
     async fn reorg_status(&self) -> Result<Option<ReorgEvent>> {
         self.anonymously_read_from_bucket(S3Storage::reorg_flag_key())
             .await?

Original file line number	Diff line number	Diff line change
`@@ -45,6 +45,10 @@ impl LocalStorage {`
`45`	`45`	`self.path.join("reorg_flag.json")`
`46`	`46`	`}`
`47`	`47`
	`48`	`+ fn reorg_rpc_responses_path(&self) -> PathBuf {`
	`49`	`+ self.path.join("reorg_rpc_responses.json")`
	`50`	`+ }`
	`51`	`+`
`48`	`52`	`fn metadata_file_path(&self) -> PathBuf {`
`49`	`53`	`self.path.join("metadata_latest.json")`
`50`	`54`	`}`
`@@ -136,4 +140,12 @@ impl CheckpointSyncer for LocalStorage {`
`136`	`140`	`let reorg = serde_json::from_slice(&data)?;`
`137`	`141`	`Ok(Some(reorg))`
`138`	`142`	`}`
	`143`	`+`
	`144`	`+ async fn write_reorg_rpc_responses(&self, log: String) -> Result<()> {`
	`145`	`+ let path = self.reorg_rpc_responses_path();`
	`146`	`+ tokio::fs::write(&path, &log)`
	`147`	`+ .await`
	`148`	`+ .with_context(\|\| format!("Writing log to {path:?}"))?;`
	`149`	`+ Ok(())`
	`150`	`+ }`
`139`	`151`	`}`