Rebase mariari/key parsing fix

.changelog/unreleased/bug-fixes/1345-key-parsing-fix.md

@@ -0,0 +1,2 @@
+- Correctly handle parsing storage key if they are empty.
+  ([#1345](https://github.com/anoma/namada/pull/1345))

apps/src/lib/node/ledger/shell/finalize_block.rs

@@ -895,7 +895,6 @@ fn pos_votes_from_abci(
 #[cfg(test)]
 mod test_finalize_block {
     use std::collections::{BTreeMap, BTreeSet};
-    use std::str::FromStr;
 
     use data_encoding::HEXUPPER;
     use namada::ledger::parameters::EpochDuration;
@@ -1340,12 +1339,11 @@ mod test_finalize_block {
 
         // Collect all storage key-vals into a sorted map
         let store_block_state = |shell: &TestShell| -> BTreeMap<_, _> {
-            let prefix: Key = FromStr::from_str("").unwrap();
             shell
                 .wl_storage
                 .storage
                 .db
-                .iter_prefix(&prefix)
+                .iter_optional_prefix(None)
                 .map(|(key, val, _gas)| (key, val))
                 .collect()
         };

apps/src/lib/node/ledger/shell/init_chain.rs

@@ -459,11 +459,9 @@ where
 #[cfg(test)]
 mod test {
     use std::collections::BTreeMap;
-    use std::str::FromStr;
 
     use namada::ledger::storage::DBIter;
     use namada::types::chain::ChainId;
-    use namada::types::storage;
 
     use crate::facade::tendermint_proto::abci::RequestInitChain;
     use crate::facade::tendermint_proto::google::protobuf::Timestamp;
@@ -477,12 +475,11 @@ mod test {
 
         // Collect all storage key-vals into a sorted map
         let store_block_state = |shell: &TestShell| -> BTreeMap<_, _> {
-            let prefix: storage::Key = FromStr::from_str("").unwrap();
             shell
                 .wl_storage
                 .storage
                 .db
-                .iter_prefix(&prefix)
+                .iter_optional_prefix(None)
                 .map(|(key, val, _gas)| (key, val))
                 .collect()
         };

apps/src/lib/node/ledger/storage/rocksdb.rs

@@ -413,9 +413,8 @@ impl RocksDB {
         let batch = Mutex::new(batch);
 
         tracing::info!("Restoring previous hight subspace diffs");
-        self.iter_prefix(&Key::default())
-            .par_bridge()
-            .try_for_each(|(key, _value, _gas)| -> Result<()> {
+        self.iter_optional_prefix(None).par_bridge().try_for_each(
+            |(key, _value, _gas)| -> Result<()> {
                 // Restore previous height diff if present, otherwise delete the
                 // subspace key
                 let subspace_cf = self.get_column_family(SUBSPACE_CF)?;
@@ -433,7 +432,8 @@ impl RocksDB {
                 }
 
                 Ok(())
-            })?;
+            },
+        )?;
 
         tracing::info!("Deleting keys prepended with the last height");
         let mut batch = batch.into_inner().unwrap();

core/src/ledger/storage/mod.rs

@@ -600,7 +600,10 @@ where
         &self,
         prefix: &Key,
     ) -> (<D as DBIter<'_>>::PrefixIter, u64) {
-        (self.db.iter_prefix(prefix), prefix.len() as _)
+        (
+            self.db.iter_optional_prefix(Some(prefix)),
+            prefix.len() as _,
+        )
     }
 
     /// Returns a prefix iterator and the gas cost

core/src/ledger/storage/wl_storage.rs

@@ -236,7 +236,7 @@ where
     D: DB + for<'iter_> DBIter<'iter_>,
     H: StorageHasher,
 {
-    let storage_iter = storage.db.iter_prefix(prefix).peekable();
+    let storage_iter = storage.db.iter_optional_prefix(Some(prefix)).peekable();
     let write_log_iter = write_log.iter_prefix_pre(prefix).peekable();
     (
         PrefixIter {
@@ -261,7 +261,7 @@ where
     D: DB + for<'iter_> DBIter<'iter_>,
     H: StorageHasher,
 {
-    let storage_iter = storage.db.iter_prefix(prefix).peekable();
+    let storage_iter = storage.db.iter_optional_prefix(Some(prefix)).peekable();
     let write_log_iter = write_log.iter_prefix_post(prefix).peekable();
     (
         PrefixIter {

core/src/types/storage.rs

@@ -480,11 +480,16 @@ impl Value for TreeBytes {
 impl Key {
     /// Parses string and returns a key
     pub fn parse(string: impl AsRef<str>) -> Result<Self> {
-        let mut segments = Vec::new();
-        for s in string.as_ref().split(KEY_SEGMENT_SEPARATOR) {
-            segments.push(DbKeySeg::parse(s.to_owned())?);
+        let string = string.as_ref();
+        if string.is_empty() {
+            Err(Error::ParseKeySeg(string.to_string()))
+        } else {
+            let mut segments = Vec::new();
+            for s in string.split(KEY_SEGMENT_SEPARATOR) {
+                segments.push(DbKeySeg::parse(s.to_owned())?);
+            }
+            Ok(Key { segments })
         }
-        Ok(Key { segments })
     }
 
     /// Returns a new key with segments of `Self` and the given segment