fix storage height bugs and rocksdb read from arb. height and delete's diffs

.changelog/unreleased/bug-fixes/706-storage-read-height-bug.md

@@ -0,0 +1,3 @@
+- Fixed storage read from arbitrary height and added an optional config value
+  `shell.storage_read_past_height_limit` to limit how far back storage queries
+  can read from. ([#706](https://github.com/anoma/namada/pull/706))

apps/src/lib/config/mod.rs

@@ -88,6 +88,9 @@ pub struct Shell {
     /// Tx WASM compilation in-memory cache maximum size in bytes.
     /// When not set, defaults to 1/6 of the available memory.
     pub tx_wasm_compilation_cache_bytes: Option<u64>,
+    /// When set, will limit the how many block heights in the past can the
+    /// storage be queried for reading values.
+    pub storage_read_past_height_limit: Option<u64>,
     /// Use the [`Ledger::db_dir()`] method to read the value.
     db_dir: PathBuf,
     /// Use the [`Ledger::tendermint_dir()`] method to read the value.
@@ -135,6 +138,8 @@ impl Ledger {
                 block_cache_bytes: None,
                 vp_wasm_compilation_cache_bytes: None,
                 tx_wasm_compilation_cache_bytes: None,
+                // Default corresponds to 1 hour of past blocks at 1 block/sec
+                storage_read_past_height_limit: Some(3600),
                 db_dir: DB_DIR.into(),
                 tendermint_dir: TENDERMINT_DIR.into(),
             },

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

@@ -211,6 +211,10 @@ where
     vp_wasm_cache: VpCache<WasmCacheRwAccess>,
     /// Tx WASM compilation cache
     tx_wasm_cache: TxCache<WasmCacheRwAccess>,
+    /// Taken from config `storage_read_past_height_limit`. When set, will
+    /// limit the how many block heights in the past can the storage be
+    /// queried for reading values.
+    storage_read_past_height_limit: Option<u64>,
     /// Proposal execution tracking
     pub proposal_data: HashSet<u64>,
 }
@@ -234,6 +238,8 @@ where
         let db_path = config.shell.db_dir(&chain_id);
         let base_dir = config.shell.base_dir;
         let mode = config.tendermint.tendermint_mode;
+        let storage_read_past_height_limit =
+            config.shell.storage_read_past_height_limit;
         if !Path::new(&base_dir).is_dir() {
             std::fs::create_dir(&base_dir)
                 .expect("Creating directory for Anoma should not fail");
@@ -317,6 +323,7 @@ where
                 tx_wasm_cache_dir,
                 tx_wasm_compilation_cache as usize,
             ),
+            storage_read_past_height_limit,
             proposal_data: HashSet::new(),
         }
     }

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

@@ -25,6 +25,7 @@ where
             storage: &self.storage,
             vp_wasm_cache: self.vp_wasm_cache.read_only(),
             tx_wasm_cache: self.tx_wasm_cache.read_only(),
+            storage_read_past_height_limit: self.storage_read_past_height_limit,
         };
 
         // Convert request to domain-type

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

@@ -53,6 +53,9 @@ mod tests {
     use namada::ledger::storage::types;
     use namada::types::chain::ChainId;
     use namada::types::storage::{BlockHash, BlockHeight, Key};
+    use proptest::collection::vec;
+    use proptest::prelude::*;
+    use proptest::test_runner::Config;
     use tempfile::TempDir;
 
     use super::*;
@@ -210,4 +213,110 @@ mod tests {
         assert_eq!(vp.expect("no VP"), vp1);
         assert_eq!(gas, (key.len() + vp1.len()) as u64);
     }
+
+    proptest! {
+        #![proptest_config(Config {
+            cases: 5,
+            .. Config::default()
+        })]
+        #[test]
+        fn test_read_with_height(blocks_write_value in vec(any::<bool>(), 20)) {
+            test_read_with_height_aux(blocks_write_value).unwrap()
+        }
+    }
+
+    /// Test reads at arbitrary block heights.
+    ///
+    /// We generate `blocks_write_value` with random bools as the input to this
+    /// function, then:
+    ///
+    /// 1. For each `blocks_write_value`, write the current block height if true
+    ///    or delete otherwise.
+    /// 2. We try to read from these heights to check that we get back expected
+    ///    value if was written at that block height or `None` if it was
+    ///    deleted.
+    /// 3. We try to read past the last height and we expect the last written
+    ///    value, if any.
+    fn test_read_with_height_aux(
+        blocks_write_value: Vec<bool>,
+    ) -> namada::ledger::storage::Result<()> {
+        let db_path =
+            TempDir::new().expect("Unable to create a temporary DB directory");
+        let mut storage =
+            PersistentStorage::open(db_path.path(), ChainId::default(), None);
+
+        // 1. For each `blocks_write_value`, write the current block height if
+        // true or delete otherwise.
+        // We `.enumerate()` height (starting from `0`)
+        let blocks_write_value = blocks_write_value
+            .into_iter()
+            .enumerate()
+            .map(|(height, write_value)| {
+                println!(
+                    "At height {height} will {}",
+                    if write_value { "write" } else { "delete" }
+                );
+                (BlockHeight::from(height as u64), write_value)
+            });
+
+        let key = Key::parse("key").expect("cannot parse the key string");
+        for (height, write_value) in blocks_write_value.clone() {
+            let hash = BlockHash::default();
+            storage.begin_block(hash, height)?;
+            assert_eq!(
+                height, storage.block.height,
+                "sanity check - height is as expected"
+            );
+
+            if write_value {
+                let value_bytes = types::encode(&storage.block.height);
+                storage.write(&key, value_bytes)?;
+            } else {
+                storage.delete(&key)?;
+            }
+            storage.commit()?;
+        }
+
+        // 2. We try to read from these heights to check that we get back
+        // expected value if was written at that block height or
+        // `None` if it was deleted.
+        for (height, write_value) in blocks_write_value.clone() {
+            let (value_bytes, _gas) = storage.read_with_height(&key, height)?;
+            if write_value {
+                let value_bytes = value_bytes.unwrap_or_else(|| {
+                    panic!("Couldn't read from height {height}")
+                });
+                let value: BlockHeight = types::decode(value_bytes).unwrap();
+                assert_eq!(value, height);
+            } else if value_bytes.is_some() {
+                let value: BlockHeight =
+                    types::decode(value_bytes.unwrap()).unwrap();
+                panic!("Expected no value at height {height}, got {}", value,);
+            }
+        }
+
+        // 3. We try to read past the last height and we expect the last written
+        // value, if any.
+
+        // If height is >= storage.last_height, it should read the latest state.
+        let is_last_write = blocks_write_value.last().unwrap().1;
+
+        // The upper bound is arbitrary.
+        for height in storage.last_height.0..storage.last_height.0 + 10 {
+            let height = BlockHeight::from(height);
+            let (value_bytes, _gas) = storage.read_with_height(&key, height)?;
+            if is_last_write {
+                let value_bytes =
+                    value_bytes.expect("Should have been written");
+                let value: BlockHeight = types::decode(value_bytes).unwrap();
+                assert_eq!(value, storage.last_height);
+            } else if value_bytes.is_some() {
+                let value: BlockHeight =
+                    types::decode(value_bytes.unwrap()).unwrap();
+                panic!("Expected no value at height {height}, got {}", value,);
+            }
+        }
+
+        Ok(())
+    }
 }

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

@@ -614,31 +614,81 @@ impl DB for RocksDB {
         &self,
         key: &Key,
         height: BlockHeight,
+        last_height: BlockHeight,
     ) -> Result<Option<Vec<u8>>> {
-        if self.read_subspace_val(key)?.is_none() {
-            return Ok(None);
+        // Check if the value changed at this height
+        let key_prefix = Key::from(height.to_db_key())
+            .push(&"diffs".to_owned())
+            .map_err(Error::KeyError)?;
+        let new_val_key = key_prefix
+            .push(&"new".to_owned())
+            .map_err(Error::KeyError)?
+            .join(key)
+            .to_string();
+
+        // If it has a "new" val, it was written at this height
+        match self
+            .0
+            .get(new_val_key)
+            .map_err(|e| Error::DBError(e.into_string()))?
+        {
+            Some(new_val) => {
+                return Ok(Some(new_val));
+            }
+            None => {
+                let old_val_key = key_prefix
+                    .push(&"old".to_owned())
+                    .map_err(Error::KeyError)?
+                    .join(key)
+                    .to_string();
+                // If it has an "old" val, it was deleted at this height
+                if self.0.key_may_exist(old_val_key) {
+                    return Ok(None);
+                }
+            }
         }
 
-        let mut height = height.0;
-        while height > 0 {
-            let key_prefix = Key::from(BlockHeight(height).to_db_key())
+        // If the value didn't change at the given height, we try to look for it
+        // at successor heights, up to the `last_height`
+        let mut raw_height = height.0 + 1;
+        loop {
+            // Try to find the next diff on this key
+            let key_prefix = Key::from(BlockHeight(raw_height).to_db_key())
                 .push(&"diffs".to_owned())
                 .map_err(Error::KeyError)?;
-            let new_val_key = key_prefix
-                .push(&"new".to_owned())
+            let old_val_key = key_prefix
+                .push(&"old".to_owned())
                 .map_err(Error::KeyError)?
                 .join(key)
                 .to_string();
-            let val = self
+            let old_val = self
                 .0
-                .get(new_val_key)
+                .get(old_val_key)
                 .map_err(|e| Error::DBError(e.into_string()))?;
-            match val {
+            // If it has an "old" val, it's the one we're looking for
+            match old_val {
                 Some(bytes) => return Ok(Some(bytes)),
-                None => height -= 1,
+                None => {
+                    // Check if the value was created at this height instead,
+                    // which would mean that it wasn't present before
+                    let new_val_key = key_prefix
+                        .push(&"new".to_owned())
+                        .map_err(Error::KeyError)?
+                        .join(key)
+                        .to_string();
+                    if self.0.key_may_exist(new_val_key) {
+                        return Ok(None);
+                    }
+
+                    if raw_height >= last_height.0 {
+                        // Read from latest height
+                        return self.read_subspace_val(key);
+                    } else {
+                        raw_height += 1
+                    }
+                }
             }
         }
-        Ok(None)
     }
 
     fn write_subspace_val(
@@ -690,7 +740,7 @@ impl DB for RocksDB {
         // Check the length of previous value, if any
         let prev_len = match self
             .0
-            .get(key.to_string())
+            .get(subspace_key.to_string())
             .map_err(|e| Error::DBError(e.into_string()))?
         {
             Some(prev_value) => {
@@ -1033,7 +1083,7 @@ mod test {
         db.exec_batch(batch.0).unwrap();
 
         let prev_value = db
-            .read_subspace_val_with_height(&key, BlockHeight(100))
+            .read_subspace_val_with_height(&key, BlockHeight(100), last_height)
             .expect("read should succeed");
         assert_eq!(prev_value, Some(vec![1_u8, 1, 1, 1]));
 

shared/src/ledger/queries/mod.rs

@@ -223,6 +223,7 @@ mod testing {
                 storage: &self.storage,
                 vp_wasm_cache: self.vp_wasm_cache.clone(),
                 tx_wasm_cache: self.tx_wasm_cache.clone(),
+                storage_read_past_height_limit: None,
             };
             let response = self.rpc.handle(ctx, &request).unwrap();
             Ok(response)

shared/src/ledger/queries/router.rs

@@ -1011,6 +1011,7 @@ mod test {
             storage: &client.storage,
             vp_wasm_cache: client.vp_wasm_cache.clone(),
             tx_wasm_cache: client.tx_wasm_cache.clone(),
+            storage_read_past_height_limit: None,
         };
         let result = TEST_RPC.handle(ctx, &request);
         assert!(result.is_err());

shared/src/ledger/queries/shell.rs

@@ -110,6 +110,19 @@ where
     D: 'static + DB + for<'iter> DBIter<'iter> + Sync,
     H: 'static + StorageHasher + Sync,
 {
+    if let Some(past_height_limit) = ctx.storage_read_past_height_limit {
+        if request.height.0 + past_height_limit < ctx.storage.last_height.0 {
+            return Err(storage_api::Error::new(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                format!(
+                    "Cannot query more than {past_height_limit} blocks in the \
+                     past (configured via \
+                     `shell.storage_read_past_height_limit`)."
+                ),
+            )));
+        }
+    }
+
     match ctx
         .storage
         .read_with_height(&storage_key, request.height)

shared/src/ledger/queries/types.rs

@@ -24,6 +24,10 @@ where
     /// tx WASM compilation cache
     #[cfg(feature = "wasm-runtime")]
     pub tx_wasm_cache: TxCache<WasmCacheRoAccess>,
+    /// Taken from config `storage_read_past_height_limit`. When set, will
+    /// limit the how many block heights in the past can the storage be
+    /// queried for reading values.
+    pub storage_read_past_height_limit: Option<u64>,
 }
 
 /// A `Router` handles parsing read-only query requests and dispatching them to

shared/src/ledger/storage/mockdb.rs

@@ -336,6 +336,7 @@ impl DB for MockDB {
         &self,
         _key: &Key,
         _height: BlockHeight,
+        _last_height: BlockHeight,
     ) -> Result<Option<Vec<u8>>> {
         // Mock DB can read only the latest value for now
         unimplemented!()