Merge branch 'tomas/iter-prefix-full-match' (#1642)

* origin/tomas/iter-prefix-full-match:
  changelog: add #1642
  test/storage/rocksdb: check that prefix_iter matches only full segments
  ledger/db: ensure that prefix iter only matches full key segments

.changelog/unreleased/improvements/1642-iter-prefix-full-match.md

@@ -0,0 +1,3 @@
+- Storage: Ensure that prefix iterator only returns key-
+  vals in which the prefix key segments are matched fully.
+  ([\#1642](https://github.com/anoma/namada/pull/1642))

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

@@ -1312,8 +1312,18 @@ fn iter_subspace_prefix<'iter>(
         .get_column_family(SUBSPACE_CF)
         .expect("{SUBSPACE_CF} column family should exist");
     let db_prefix = "".to_owned();
-    let prefix = prefix.map(|k| k.to_string()).unwrap_or_default();
-    iter_prefix(db, subspace_cf, db_prefix, Some(prefix))
+    iter_prefix(
+        db,
+        subspace_cf,
+        db_prefix,
+        prefix.map(|k| {
+            if k == &Key::default() {
+                k.to_string()
+            } else {
+                format!("{k}/")
+            }
+        }),
+    )
 }
 
 fn iter_diffs_prefix(
@@ -1609,28 +1619,24 @@ mod test {
         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 prefix_01 = Key::parse("01").unwrap();
+        let key_01_a = prefix_01.push(&"a".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();
+        let keys_0 = vec![key_0_a, key_0_b, key_0_c];
+        let keys_1 = vec![key_1_a, key_1_b, key_1_c];
+        let keys_01 = vec![key_01_a];
+        let all_keys = vec![keys_0.clone(), keys_01, 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();
+        for key in &all_keys {
+            db.batch_write_subspace_val(&mut batch, height, key, [0_u8])
+                .unwrap();
+        }
         db.exec_batch(batch.0).unwrap();
 
+        // Prefix "0" shouldn't match prefix "01"
         let itered_keys: Vec<Key> = db
             .iter_prefix(Some(&prefix_0))
             .map(|(key, _val, _)| Key::parse(key).unwrap())

core/src/ledger/storage/mockdb.rs

@@ -516,8 +516,20 @@ impl<'iter> DBIter<'iter> for MockDB {
 
     fn iter_prefix(&'iter self, prefix: Option<&Key>) -> MockPrefixIterator {
         let db_prefix = "subspace/".to_owned();
-        let prefix_str = prefix.map(|k| k.to_string()).unwrap_or_default();
-        let prefix = format!("{}{}", db_prefix, prefix_str);
+        let prefix = format!(
+            "{}{}",
+            db_prefix,
+            match prefix {
+                Some(prefix) => {
+                    if prefix == &Key::default() {
+                        prefix.to_string()
+                    } else {
+                        format!("{prefix}/")
+                    }
+                }
+                None => "".to_string(),
+            }
+        );
         let iter = self.0.borrow().clone().into_iter();
         MockPrefixIterator::new(MockIterator { prefix, iter }, db_prefix)
     }