Ensure validator writes latest index to checkpoint syncer

rust/agents/validator/src/submit.rs

@@ -136,6 +136,10 @@ impl ValidatorSubmitter {
             self.submit_checkpoints_until_correctness_checkpoint(&mut tree, &latest_checkpoint)
                 .await?;
 
+            self.checkpoint_syncer
+                .write_latest_index(latest_checkpoint.index)
+                .await?;
+
             self.metrics
                 .latest_checkpoint_processed
                 .set(latest_checkpoint.index as i64);

rust/hyperlane-base/src/traits/checkpoint_syncer.rs

@@ -10,6 +10,8 @@ use hyperlane_core::{SignedAnnouncement, SignedCheckpointWithMessageId};
 pub trait CheckpointSyncer: Debug + Send + Sync {
     /// Read the highest index of this Syncer
     async fn latest_index(&self) -> Result<Option<u32>>;
+    /// Writes the highest index of this Syncer
+    async fn write_latest_index(&self, index: u32) -> Result<()>;
     /// Attempt to fetch the signed (checkpoint, messageId) tuple at this index
     async fn fetch_checkpoint(&self, index: u32) -> Result<Option<SignedCheckpointWithMessageId>>;
     /// Write the signed (checkpoint, messageId) tuple to this syncer

rust/hyperlane-base/src/types/local_storage.rs

@@ -62,6 +62,14 @@ impl CheckpointSyncer for LocalStorage {
         }
     }
 
+    async fn write_latest_index(&self, index: u32) -> Result<()> {
+        let path = self.latest_index_file_path();
+        tokio::fs::write(&path, index.to_string())
+            .await
+            .with_context(|| format!("Writing index to {path:?}"))?;
+        Ok(())
+    }
+
     async fn fetch_checkpoint(&self, index: u32) -> Result<Option<SignedCheckpointWithMessageId>> {
         let Ok(data) = tokio::fs::read(self.checkpoint_file_path(index)).await else {
             return Ok(None);

rust/hyperlane-base/src/types/s3_storage.rs

@@ -131,7 +131,7 @@ impl S3Storage {
         format!("checkpoint_{index}_with_id.json")
     }
 
-    fn index_key() -> String {
+    fn latest_index_key() -> String {
         "checkpoint_latest_index.json".to_owned()
     }
 
@@ -144,7 +144,7 @@ impl S3Storage {
 impl CheckpointSyncer for S3Storage {
     async fn latest_index(&self) -> Result<Option<u32>> {
         let ret = self
-            .anonymously_read_from_bucket(S3Storage::index_key())
+            .anonymously_read_from_bucket(S3Storage::latest_index_key())
             .await?
             .map(|data| serde_json::from_slice(&data))
             .transpose()
@@ -159,6 +159,13 @@ impl CheckpointSyncer for S3Storage {
         ret
     }
 
+    async fn write_latest_index(&self, index: u32) -> Result<()> {
+        let serialized_index = serde_json::to_string(&index)?;
+        self.write_to_bucket(S3Storage::latest_index_key(), &serialized_index)
+            .await?;
+        Ok(())
+    }
+
     async fn fetch_checkpoint(&self, index: u32) -> Result<Option<SignedCheckpointWithMessageId>> {
         self.anonymously_read_from_bucket(S3Storage::checkpoint_key(index))
             .await?