fix optional prefix iter

.changelog/unreleased/bug-fixes/1615-fix-optional-prefix-iter.md

@@ -0,0 +1,2 @@
+- Storage: Fix iterator without a prefix.
+  ([\#1615](https://github.com/anoma/namada/pull/1615))

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

@@ -1289,11 +1289,7 @@ fn iter_subspace_prefix<'iter>(
         .get_column_family(SUBSPACE_CF)
         .expect("{SUBSPACE_CF} column family should exist");
     let db_prefix = "".to_owned();
-    let prefix_string = match prefix {
-        Some(prefix) => prefix.to_string(),
-        None => "".to_string(),
-    };
-    iter_prefix(db, subspace_cf, db_prefix, prefix_string)
+    iter_prefix(db, subspace_cf, db_prefix, prefix.map(|k| k.to_string()))
 }
 
 fn iter_diffs_prefix(
@@ -1307,20 +1303,23 @@ fn iter_diffs_prefix(
     let prefix = if is_old { "old" } else { "new" };
     let db_prefix = format!("{}/{}/", height.0.raw(), prefix);
     // get keys without a prefix
-    iter_prefix(db, diffs_cf, db_prefix.clone(), db_prefix)
+    iter_prefix(db, diffs_cf, db_prefix.clone(), Some(db_prefix))
 }
 
 fn iter_prefix<'a>(
     db: &'a RocksDB,
     cf: &'a ColumnFamily,
     db_prefix: String,
-    prefix: String,
+    prefix: Option<String>,
 ) -> PersistentPrefixIterator<'a> {
-    let read_opts = make_iter_read_opts(Some(prefix.clone()));
+    let read_opts = make_iter_read_opts(prefix.clone());
     let iter = db.0.iterator_cf_opt(
         cf,
         read_opts,
-        IteratorMode::From(prefix.as_bytes(), Direction::Forward),
+        IteratorMode::From(
+            prefix.unwrap_or_default().as_bytes(),
+            Direction::Forward,
+        ),
     );
     PersistentPrefixIterator(PrefixIterator::new(iter, db_prefix))
 }
@@ -1364,8 +1363,8 @@ fn make_iter_read_opts(prefix: Option<String>) -> ReadOptions {
         let mut upper_prefix = prefix.into_bytes();
         if let Some(last) = upper_prefix.pop() {
             upper_prefix.push(last + 1);
+            read_opts.set_iterate_upper_bound(upper_prefix);
         }
-        read_opts.set_iterate_upper_bound(upper_prefix);
     }
 
     read_opts
@@ -1600,4 +1599,58 @@ mod test {
             db.read_subspace_val(&key).expect("read should succeed");
         assert_eq!(latest_value, None);
     }
+
+    #[test]
+    fn test_prefix_iter() {
+        let dir = tempdir().unwrap();
+        let mut db = open(dir.path(), None).unwrap();
+
+        let prefix_0 = Key::parse("0").unwrap();
+        let key_0_a = prefix_0.push(&"a".to_string()).unwrap();
+        let key_0_b = prefix_0.push(&"b".to_string()).unwrap();
+        let key_0_c = prefix_0.push(&"c".to_string()).unwrap();
+        let prefix_1 = Key::parse("1").unwrap();
+        let key_1_a = prefix_1.push(&"a".to_string()).unwrap();
+        let key_1_b = prefix_1.push(&"b".to_string()).unwrap();
+        let key_1_c = prefix_1.push(&"c".to_string()).unwrap();
+
+        let keys_0 = vec![key_0_a.clone(), key_0_b.clone(), key_0_c.clone()];
+        let keys_1 = vec![key_1_a.clone(), key_1_b.clone(), key_1_c.clone()];
+        let all_keys = vec![keys_0.clone(), keys_1.clone()].concat();
+
+        // Write the keys
+        let mut batch = RocksDB::batch();
+        let height = BlockHeight(1);
+        db.batch_write_subspace_val(&mut batch, height, &key_0_a, [0_u8])
+            .unwrap();
+        db.batch_write_subspace_val(&mut batch, height, &key_0_b, [0_u8])
+            .unwrap();
+        db.batch_write_subspace_val(&mut batch, height, &key_0_c, [0_u8])
+            .unwrap();
+        db.batch_write_subspace_val(&mut batch, height, &key_1_a, [0_u8])
+            .unwrap();
+        db.batch_write_subspace_val(&mut batch, height, &key_1_b, [0_u8])
+            .unwrap();
+        db.batch_write_subspace_val(&mut batch, height, &key_1_c, [0_u8])
+            .unwrap();
+        db.exec_batch(batch.0).unwrap();
+
+        let itered_keys: Vec<Key> = db
+            .iter_optional_prefix(Some(&prefix_0))
+            .map(|(key, _val, _)| Key::parse(key).unwrap())
+            .collect();
+        itertools::assert_equal(keys_0, itered_keys);
+
+        let itered_keys: Vec<Key> = db
+            .iter_optional_prefix(Some(&prefix_1))
+            .map(|(key, _val, _)| Key::parse(key).unwrap())
+            .collect();
+        itertools::assert_equal(keys_1, itered_keys);
+
+        let itered_keys: Vec<Key> = db
+            .iter_optional_prefix(None)
+            .map(|(key, _val, _)| Key::parse(key).unwrap())
+            .collect();
+        itertools::assert_equal(all_keys, itered_keys);
+    }
 }