Merge branch 'tomas/storage-read-height-bug' into james+ray/ethbridge…

…/merge-0.9.0

* tomas/storage-read-height-bug:
  changelog: add #706
  config: add `shell.storage_read_past_height_limit`, use it for queries
  storage: fix block heights used in read, write and delete
  rocksdb: fix delete_subspace_val diff write
  rocksdb: fix read_subspace_val_with_height
  test/storage: add a test for reading from arbitrary block heights

# Conflicts:
#	apps/src/lib/node/ledger/shell/mod.rs

.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

@@ -90,6 +90,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.
@@ -137,6 +140,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

@@ -332,6 +332,10 @@ where
     pub(super) vp_wasm_cache: VpCache<WasmCacheRwAccess>,
     /// Tx WASM compilation cache
     pub(super) 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>,
     /// Log of events emitted by `FinalizeBlock` ABCI calls.
@@ -358,6 +362,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");
@@ -449,6 +455,7 @@ where
                 tx_wasm_cache_dir,
                 tx_wasm_compilation_cache as usize,
             ),
+            storage_read_past_height_limit,
             proposal_data: HashSet::new(),
             // TODO: config event log params
             event_log: EventLog::default(),

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

@@ -60,6 +60,7 @@ where
             event_log: self.event_log(),
             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

@@ -54,6 +54,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::*;
@@ -211,4 +214,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,77 @@ 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
+        if let Some(new_val) = self
+            .0
+            .get(new_val_key)
+            .map_err(|e| Error::DBError(e.into_string()))?
+        {
+            return Ok(Some(new_val));
+        } else {
+            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 +736,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) => {
@@ -1034,7 +1080,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

@@ -230,6 +230,7 @@ mod testing {
                 event_log: &self.event_log,
                 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

@@ -1012,6 +1012,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

@@ -125,6 +125,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

@@ -28,6 +28,10 @@ where
     /// Cache of transaction wasm compiled artifacts.
     #[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!()

shared/src/ledger/storage/mod.rs

@@ -189,11 +189,15 @@ pub trait DB: std::fmt::Debug {
     /// Read the latest value for account subspace key from the DB
     fn read_subspace_val(&self, key: &Key) -> Result<Option<Vec<u8>>>;
 
-    /// Read the value for account subspace key at the given height from the DB
+    /// Read the value for account subspace key at the given height from the DB.
+    /// In our `PersistentStorage` (rocksdb), to find a value from arbitrary
+    /// height requires looking for diffs from the given `height`, possibly
+    /// up to the `last_height`.
     fn read_subspace_val_with_height(
         &self,
         key: &Key,
-        _height: BlockHeight,
+        height: BlockHeight,
+        last_height: BlockHeight,
     ) -> Result<Option<Vec<u8>>>;
 
     /// Write the value with the given height and account subspace key to the
@@ -409,10 +413,14 @@ where
         key: &Key,
         height: BlockHeight,
     ) -> Result<(Option<Vec<u8>>, u64)> {
-        if height >= self.get_block_height().0 {
+        if height >= self.last_height {
             self.read(key)
         } else {
-            match self.db.read_subspace_val_with_height(key, height)? {
+            match self.db.read_subspace_val_with_height(
+                key,
+                height,
+                self.last_height,
+            )? {
                 Some(v) => {
                     let gas = key.len() + v.len();
                     Ok((Some(v), gas as _))
@@ -455,7 +463,7 @@ where
         let len = value.len();
         let gas = key.len() + len;
         let size_diff =
-            self.db.write_subspace_val(self.last_height, key, value)?;
+            self.db.write_subspace_val(self.block.height, key, value)?;
         Ok((gas as _, size_diff))
     }
 
@@ -468,7 +476,7 @@ where
         if self.has_key(key)?.0 {
             self.block.tree.delete(key)?;
             deleted_bytes_len =
-                self.db.delete_subspace_val(self.last_height, key)?;
+                self.db.delete_subspace_val(self.block.height, key)?;
         }
         let gas = key.len() + deleted_bytes_len as usize;
         Ok((gas as _, deleted_bytes_len))