Fix history reads migration

ledger/store/src/helpers/memory/program.rs

@@ -16,7 +16,7 @@
 #![allow(clippy::type_complexity)]
 
 #[cfg(feature = "history")]
-use crate::program::HeightBytes;
+use crate::program::{HeightBytes, migrate_legacy_mapping_updates};
 use crate::{
     CommitteeStorage,
     CommitteeStore,
@@ -52,7 +52,7 @@ pub struct FinalizeMemory<N: Network> {
     /// The historical mapping value map (keyed by big-endian block height).
     #[cfg(feature = "history")]
     mapping_update_map: MemoryMap<(ProgramID<N>, Identifier<N>, Plaintext<N>, HeightBytes), Value<N>>,
-    /// The legacy heights index; present only for keys written before the BE schema change.
+    /// The legacy heights index, retained only to migrate pre-schema-change entries.
     #[cfg(feature = "history")]
     mapping_update_heights_map: MemoryMap<(ProgramID<N>, Identifier<N>, Plaintext<N>), Vec<u32>>,
     /// The current block height.
@@ -87,8 +87,8 @@ impl<N: Network> FinalizeStorage<N> for FinalizeMemory<N> {
         // Returns 0 for a fresh database that has no committee data yet.
         #[cfg(feature = "history")]
         let initial_height = committee_store.current_height().unwrap_or(0);
-        // Return the finalize store.
-        Ok(Self {
+        // Initialize the finalize store.
+        let storage = Self {
             committee_store,
             program_id_map: MemoryMap::default(),
             key_value_map: NestedMemoryMap::default(),
@@ -102,7 +102,11 @@ impl<N: Network> FinalizeStorage<N> for FinalizeMemory<N> {
             #[cfg(feature = "history-staking-rewards")]
             staking_rewards_map: MemoryMap::default(),
             storage_mode: storage,
-        })
+        };
+        // Rewrite legacy history entries before accepting requests.
+        #[cfg(feature = "history")]
+        migrate_legacy_mapping_updates(&storage)?;
+        Ok(storage)
     }
 
     /// Returns the committee store.

ledger/store/src/helpers/rocksdb/internal/map.rs

@@ -369,6 +369,22 @@ impl<
         }
     }
 
+    ///
+    /// Returns the values for the given keys from the map, in the same order as the keys.
+    ///
+    fn get_many_confirmed(&'a self, keys: &[K]) -> Result<Vec<Option<V>>> {
+        let raw_keys = keys.iter().map(|key| self.create_prefixed_key(key)).collect::<Result<Vec<_>>>()?;
+
+        self.database
+            .multi_get_opt(raw_keys, &self.database.default_readopts)
+            .into_iter()
+            .map(|value| match value? {
+                Some(bytes) => Ok(Some(unchecked_deserialize(&bytes)?)),
+                None => Ok(None),
+            })
+            .collect()
+    }
+
     ///
     /// Returns the current value for the given key if it is scheduled
     /// to be inserted as part of an atomic batch.
@@ -868,6 +884,19 @@ mod tests {
         crate::helpers::test_helpers::map::check_remove_and_get_speculative(map);
     }
 
+    #[test]
+    #[traced_test]
+    fn test_get_many_confirmed() {
+        // Initialize a map.
+        let map: DataMap<usize, String> = RocksDB::open_map(0, StorageMode::new_test(None), MapID::Test(TestMap::Test))
+            .expect("Failed to open data map");
+        map.insert(1, "one".to_string()).unwrap();
+        map.insert(2, "two".to_string()).unwrap();
+
+        let values = map.get_many_confirmed(&[2, 3, 1]).unwrap();
+        assert_eq!(values, vec![Some("two".to_string()), None, Some("one".to_string())]);
+    }
+
     #[test]
     #[traced_test]
     fn test_contains_key() {

ledger/store/src/helpers/rocksdb/program.rs

@@ -16,7 +16,7 @@
 #![allow(clippy::type_complexity)]
 
 #[cfg(feature = "history")]
-use crate::program::HeightBytes;
+use crate::program::{HeightBytes, migrate_legacy_mapping_updates};
 use crate::{
     CommitteeStorage,
     CommitteeStore,
@@ -52,7 +52,7 @@ pub struct FinalizeDB<N: Network> {
     /// The historical mapping value map (keyed by big-endian block height).
     #[cfg(feature = "history")]
     mapping_update_map: DataMap<(ProgramID<N>, Identifier<N>, Plaintext<N>, HeightBytes), Value<N>>,
-    /// The legacy heights index; present only for keys written before the BE schema change.
+    /// The legacy heights index, retained only to migrate pre-schema-change entries.
     #[cfg(feature = "history")]
     mapping_update_heights_map: DataMap<(ProgramID<N>, Identifier<N>, Plaintext<N>), Vec<u32>>,
     /// The current block height.
@@ -89,8 +89,8 @@ impl<N: Network> FinalizeStorage<N> for FinalizeDB<N> {
         // Returns 0 for a fresh database that has no committee data yet.
         #[cfg(feature = "history")]
         let initial_height = committee_store.current_height().unwrap_or(0);
-        // Return the finalize storage.
-        Ok(Self {
+        // Initialize the finalize storage.
+        let storage = Self {
             committee_store,
             program_id_map: rocksdb::RocksDB::open_map(N::ID, storage.clone(), MapID::Program(ProgramMap::ProgramID))?,
             key_value_map: rocksdb::RocksDB::open_nested_map(N::ID, storage.clone(), MapID::Program(ProgramMap::KeyValueID))?,
@@ -104,7 +104,11 @@ impl<N: Network> FinalizeStorage<N> for FinalizeDB<N> {
             #[cfg(feature = "history-staking-rewards")]
             staking_rewards_map: rocksdb::RocksDB::open_map(N::ID, storage.clone(), MapID::Program(ProgramMap::StakingRewards))?,
             storage_mode: storage,
-        })
+        };
+        // Rewrite legacy history entries before accepting requests.
+        #[cfg(feature = "history")]
+        migrate_legacy_mapping_updates(&storage)?;
+        Ok(storage)
     }
 
     /// Returns the committee store.

ledger/store/src/helpers/traits/map.rs

@@ -138,6 +138,13 @@ pub trait MapRead<
         K: Borrow<Q>,
         Q: PartialEq + Eq + Hash + Serialize + ?Sized;
 
+    ///
+    /// Returns the values for the given keys from the map, in the same order as the keys.
+    ///
+    fn get_many_confirmed(&'a self, keys: &[K]) -> Result<Vec<Option<V>>> {
+        keys.iter().map(|key| self.get_confirmed(key).map(|value| value.map(Cow::into_owned))).collect()
+    }
+
     ///
     /// Returns the current value for the given key if it is scheduled
     /// to be inserted as part of an atomic batch.