feat(storage,consensus): add index_for_key to Archive trait

consensus/src/marshal/actor.rs

@@ -342,16 +342,13 @@ where
                     match message {
                         Message::GetInfo { identifier, response } => {
                             let info = match identifier {
-                                // TODO: Instead of pulling out the entire block, determine the
-                                // height directly from the archive by mapping the commitment to
-                                // the index, which is the same as the height.
                                 BlockID::Commitment(commitment) => self
                                     .finalized_blocks
-                                    .get(ArchiveID::Key(&commitment))
+                                    .get_height(&commitment)
                                     .await
                                     .ok()
                                     .flatten()
-                                    .map(|b| (b.height(), commitment)),
+                                    .map(|height| (height, commitment)),
                                 BlockID::Height(height) => self
                                     .finalizations_by_height
                                     .get(ArchiveID::Index(height))

consensus/src/marshal/store.rs

@@ -120,6 +120,23 @@ pub trait Blocks: Send + Sync + 'static {
         id: Identifier<'_, <Self::Block as Committable>::Commitment>,
     ) -> impl Future<Output = Result<Option<Self::Block>, Self::Error>> + Send;
 
+    /// Retrieve the height (index) for a given block commitment without fetching the full block.
+    ///
+    /// This is more efficient than [Blocks::get] when only the height is needed,
+    /// as it may avoid deserializing the block.
+    ///
+    /// # Arguments
+    ///
+    /// * `commitment`: The block commitment to look up.
+    ///
+    /// # Returns
+    ///
+    /// `Ok(Some(height))` if the block exists, `Ok(None)` if not found, or `Err` on read failure.
+    fn get_height(
+        &self,
+        commitment: &<Self::Block as Committable>::Commitment,
+    ) -> impl Future<Output = Result<Option<u64>, Self::Error>> + Send;
+
     /// Prune the store to the provided minimum height (inclusive).
     ///
     /// # Arguments
@@ -228,6 +245,18 @@ where
         <Self as Archive>::get(self, id).await
     }
 
+    async fn get_height(
+        &self,
+        commitment: &<Self::Block as Committable>::Commitment,
+    ) -> Result<Option<u64>, Self::Error> {
+        // Immutable archive does not support direct key -> index lookup.
+        // Fall back to fetching the full block.
+        match <Self as Archive>::get(self, Identifier::Key(commitment)).await? {
+            Some(block) => Ok(Some(block.height())),
+            None => Ok(None),
+        }
+    }
+
     async fn prune(&mut self, _: u64) -> Result<(), Self::Error> {
         // Pruning is a no-op for immutable archives.
         Ok(())
@@ -299,6 +328,13 @@ where
         <Self as Archive>::get(self, id).await
     }
 
+    async fn get_height(
+        &self,
+        commitment: &<Self::Block as Committable>::Commitment,
+    ) -> Result<Option<u64>, Self::Error> {
+        <Self as Archive>::index_for_key(self, commitment).await
+    }
+
     async fn prune(&mut self, min: u64) -> Result<(), Self::Error> {
         Self::prune(self, min).await
     }

storage/src/archive/immutable/storage.rs

@@ -287,6 +287,14 @@ impl<E: Storage + Metrics + Clock, K: Array, V: Codec> crate::archive::Archive
         }
     }
 
+    async fn index_for_key(&self, _key: &K) -> Result<Option<u64>, Error> {
+        // Immutable archive does not maintain a direct key -> index mapping.
+        // The structure stores: key -> cursor (in Freezer) and index -> cursor (in Ordinal).
+        // There is no efficient reverse lookup from key to index without scanning.
+        // Callers should use `get` and extract the index from the value if needed.
+        Ok(None)
+    }
+
     async fn sync(&mut self) -> Result<(), Error> {
         self.syncs.inc();
 

storage/src/archive/mod.rs

@@ -80,6 +80,12 @@ pub trait Archive {
         identifier: Identifier<'_, Self::Key>,
     ) -> impl Future<Output = Result<bool, Error>>;
 
+    /// Retrieve the index associated with a key without fetching the value.
+    ///
+    /// This is more efficient than [Archive::get] when only the index is needed,
+    /// as it avoids deserializing the value.
+    fn index_for_key(&self, key: &Self::Key) -> impl Future<Output = Result<Option<u64>, Error>>;
+
     /// Retrieve the end of the current range including `index` (inclusive) and
     /// the start of the next range after `index` (if it exists).
     ///

storage/src/archive/prunable/storage.rs

@@ -344,6 +344,29 @@ impl<T: Translator, E: Storage + Metrics, K: Array, V: Codec> crate::archive::Ar
         }
     }
 
+    async fn index_for_key(&self, key: &K) -> Result<Option<u64>, Error> {
+        // Fetch index from in-memory keys map without reading the value from disk
+        let min_allowed = self.oldest_allowed.unwrap_or(0);
+        for index in self.keys.get(key) {
+            // Skip indices that have been pruned
+            if *index < min_allowed {
+                continue;
+            }
+
+            // Verify the key matches by reading the record from disk
+            let offset = *self.indices.get(index).ok_or(Error::RecordCorrupted)?;
+            let section = self.section(*index);
+            let record = self.journal.get(section, offset).await?;
+
+            if record.key.as_ref() == key.as_ref() {
+                return Ok(Some(*index));
+            }
+            self.unnecessary_reads.inc();
+        }
+
+        Ok(None)
+    }
+
     async fn sync(&mut self) -> Result<(), Error> {
         let mut syncs = Vec::with_capacity(self.pending.len());
         for section in self.pending.iter() {