feat(provider): Add StorageProvider impl, table changes (#172)

* feat(provider): Add StorageProvider impl, table changes

* unwind header numbers by walker (#174)

* readme, fmt

* fix tests

* Update crates/interfaces/src/provider/db_provider/storage.rs

Co-authored-by: Roman Krasiuk <rokrassyuk@gmail.com>

* Update crates/interfaces/src/provider/db_provider/storage.rs

Co-authored-by: Roman Krasiuk <rokrassyuk@gmail.com>

Co-authored-by: Roman Krasiuk <rokrassyuk@gmail.com>

README.md

@@ -8,7 +8,7 @@
 
 ## Build
 
-To build this project we are currently using Rust nightly for GAT support, that is planed to release in rust 1.65 (4th Nov 2022). GAT's are used for the `Database` trait in `reth-interface`. 
+Rust minimum required version to build this project is 1.65.0 published 02.11.2022
 
 ## Docs
 

crates/common/rlp/src/decode.rs

@@ -42,7 +42,7 @@ mod alloc_impl {
     }
 }
 
-#[derive(Clone, Copy, Debug, PartialEq)]
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum DecodeError {
     Overflow,
     LeadingZero,

crates/db/src/kv/cursor.rs

@@ -45,10 +45,6 @@ impl<'tx, K: TransactionKind, T: Table> DbCursorRO<'tx, T> for Cursor<'tx, K, T>
         decode!(self.inner.first())
     }
 
-    fn seek(&mut self, key: <T as Table>::SeekKey) -> reth_interfaces::db::PairResult<T> {
-        decode!(self.inner.set_range(key.encode().as_ref()))
-    }
-
     fn seek_exact(&mut self, key: <T as Table>::Key) -> reth_interfaces::db::PairResult<T> {
         decode!(self.inner.set_key(key.encode().as_ref()))
     }
@@ -87,6 +83,10 @@ impl<'tx, K: TransactionKind, T: Table> DbCursorRO<'tx, T> for Cursor<'tx, K, T>
 }
 
 impl<'tx, K: TransactionKind, T: DupSort> DbDupCursorRO<'tx, T> for Cursor<'tx, K, T> {
+    fn seek(&mut self, key: <T as DupSort>::SubKey) -> reth_interfaces::db::PairResult<T> {
+        decode!(self.inner.set_range(key.encode().as_ref()))
+    }
+
     /// Returns the next `(key, value)` pair of a DUPSORT table.
     fn next_dup(&mut self) -> PairResult<T> {
         decode!(self.inner.next_dup())

crates/db/src/kv/mod.rs

@@ -138,10 +138,13 @@ pub mod test_utils {
 #[cfg(test)]
 mod tests {
     use super::{test_utils, Env, EnvKind};
-    use reth_interfaces::db::{
+    use reth_interfaces::{
-        models::ShardedKey,
+        db::{
-        tables::{AccountHistory, Headers, PlainAccountState, PlainStorageState},
+            models::ShardedKey,
-        Database, DbCursorRO, DbDupCursorRO, DbTx, DbTxMut,
+            tables::{AccountHistory, Headers, PlainAccountState, PlainStorageState},
+            Database, DbCursorRO, DbDupCursorRO, DbTx, DbTxMut,
+        },
+        provider::{ProviderImpl, StateProviderFactory},
     };
     use reth_libmdbx::{NoWriteMap, WriteMap};
     use reth_primitives::{Account, Address, Header, IntegerList, StorageEntry, H256, U256};
@@ -315,6 +318,13 @@ mod tests {
             assert_eq!(list200, list);
         }
     }
+
+    #[test]
+    fn common_history_provider() {
+        let db = test_utils::create_test_db::<WriteMap>(EnvKind::RW);
+        let provider = ProviderImpl::new(db);
+        let _ = provider.latest();
+    }
 }
 
 #[cfg(test)]

crates/db/src/kv/tx.rs

@@ -55,12 +55,12 @@ impl<'a, K: TransactionKind, E: EnvironmentKind> DbTxMutGAT<'a> for Tx<'_, K, E>
 
 impl<'tx, K: TransactionKind, E: EnvironmentKind> DbTx<'tx> for Tx<'tx, K, E> {
     // Iterate over read only values in database.
-    fn cursor<T: Table>(&'tx self) -> Result<<Self as DbTxGAT<'tx>>::Cursor<T>, Error> {
+    fn cursor<T: Table>(&self) -> Result<<Self as DbTxGAT<'_>>::Cursor<T>, Error> {
         self.new_cursor()
     }
 
     /// Iterate over read only values in database.
-    fn cursor_dup<T: DupSort>(&'tx self) -> Result<<Self as DbTxGAT<'tx>>::DupCursor<T>, Error> {
+    fn cursor_dup<T: DupSort>(&self) -> Result<<Self as DbTxGAT<'_>>::DupCursor<T>, Error> {
         self.new_cursor()
     }
 

crates/executor/src/executor.rs

@@ -2,7 +2,10 @@ use crate::{
     revm_wrap::{self, State, SubState},
     Config,
 };
-use reth_interfaces::executor::{BlockExecutor, Error, ExecutorDb};
+use reth_interfaces::{
+    executor::{BlockExecutor, Error},
+    provider::StateProvider,
+};
 use reth_primitives::BlockLocked;
 use revm::{AnalysisKind, SpecId, EVM};
 
@@ -19,7 +22,7 @@ impl Executor {
     }
 
     /// Verify block. Execute all transaction and compare results.
-    pub fn verify<DB: ExecutorDb>(&self, block: &BlockLocked, db: DB) -> Result<(), Error> {
+    pub fn verify<DB: StateProvider>(&self, block: &BlockLocked, db: DB) -> Result<(), Error> {
         let db = SubState::new(State::new(db));
         let mut evm = EVM::new();
         evm.database(db);

crates/executor/src/revm_wrap.rs

@@ -1,18 +1,17 @@
-use reth_interfaces::executor::ExecutorDb;
+use reth_interfaces::{provider::StateProvider, Error};
 use reth_primitives::{BlockLocked, Transaction, TransactionKind, H160, H256, U256};
 use revm::{
     db::{CacheDB, DatabaseRef},
     BlockEnv, TransactTo, TxEnv,
 };
-use std::convert::Infallible;
 
 /// SubState of database. Uses revm internal cache with binding to reth DbExecutor trait.
 pub type SubState<DB> = CacheDB<State<DB>>;
 
 /// Wrapper around ExeuctorDb that implements revm database trait
-pub struct State<DB: ExecutorDb>(DB);
+pub struct State<DB: StateProvider>(DB);
 
-impl<DB: ExecutorDb> State<DB> {
+impl<DB: StateProvider> State<DB> {
     /// Create new State with generic ExecutorDb.
     pub fn new(db: DB) -> Self {
         Self(db)
@@ -34,11 +33,11 @@ impl<DB: ExecutorDb> State<DB> {
     }
 }
 
-impl<DB: ExecutorDb> DatabaseRef for State<DB> {
+impl<DB: StateProvider> DatabaseRef for State<DB> {
-    type Error = Infallible;
+    type Error = Error;
 
     fn basic(&self, address: H160) -> Result<Option<revm::AccountInfo>, Self::Error> {
-        Ok(self.0.basic_account(address).map(|account| revm::AccountInfo {
+        Ok(self.0.basic_account(address)?.map(|account| revm::AccountInfo {
             balance: account.balance,
             nonce: account.nonce,
             code_hash: account.bytecode_hash,
@@ -47,19 +46,19 @@ impl<DB: ExecutorDb> DatabaseRef for State<DB> {
     }
 
     fn code_by_hash(&self, code_hash: H256) -> Result<revm::Bytecode, Self::Error> {
-        let (bytecode, size) = self.0.bytecode_by_hash(code_hash).unwrap_or_default();
+        let bytecode = self.0.bytecode_by_hash(code_hash)?.unwrap_or_default();
-        Ok(unsafe { revm::Bytecode::new_checked(bytecode.0, size, Some(code_hash)) })
+        Ok(revm::Bytecode::new_raw(bytecode.0))
     }
 
     fn storage(&self, address: H160, index: U256) -> Result<U256, Self::Error> {
         let mut h_index = H256::zero();
         index.to_big_endian(h_index.as_bytes_mut());
 
-        Ok(U256::from_big_endian(self.0.storage(address, h_index).unwrap_or_default().as_ref()))
+        Ok(self.0.storage(address, h_index)?.unwrap_or_default())
     }
 
     fn block_hash(&self, number: U256) -> Result<H256, Self::Error> {
-        Ok(self.0.block_hash(number).unwrap_or_default())
+        Ok(self.0.block_hash(number)?.unwrap_or_default())
     }
 }
 

crates/interfaces/src/db/mock.rs

@@ -107,10 +107,6 @@ impl<'tx, T: Table> DbCursorRO<'tx, T> for CursorMock {
         todo!()
     }
 
-    fn seek(&mut self, _key: T::SeekKey) -> super::PairResult<T> {
-        todo!()
-    }
-
     fn seek_exact(&mut self, _key: T::Key) -> super::PairResult<T> {
         todo!()
     }
@@ -147,6 +143,10 @@ impl<'tx, T: DupSort> DbDupCursorRO<'tx, T> for CursorMock {
         todo!()
     }
 
+    fn seek(&mut self, _key: T::SubKey) -> super::PairResult<T> {
+        todo!()
+    }
+
     fn next_no_dup(&mut self) -> super::PairResult<T> {
         todo!()
     }

crates/interfaces/src/db/mod.rs

@@ -101,9 +101,9 @@ pub trait DbTx<'tx>: for<'a> DbTxGAT<'a> {
     /// freeing of memory pages
     fn commit(self) -> Result<bool, Error>;
     /// Iterate over read only values in table.
-    fn cursor<T: Table>(&'tx self) -> Result<<Self as DbTxGAT<'tx>>::Cursor<T>, Error>;
+    fn cursor<T: Table>(&self) -> Result<<Self as DbTxGAT<'_>>::Cursor<T>, Error>;
     /// Iterate over read only values in dup sorted table.
-    fn cursor_dup<T: DupSort>(&'tx self) -> Result<<Self as DbTxGAT<'tx>>::DupCursor<T>, Error>;
+    fn cursor_dup<T: DupSort>(&self) -> Result<<Self as DbTxGAT<'_>>::DupCursor<T>, Error>;
 }
 
 /// Read write transaction that allows writing to database
@@ -115,11 +115,11 @@ pub trait DbTxMut<'tx>: for<'a> DbTxMutGAT<'a> {
     /// Clears database.
     fn clear<T: Table>(&self) -> Result<(), Error>;
     /// Cursor mut
-    fn cursor_mut<T: Table>(&self) -> Result<<Self as DbTxMutGAT<'tx>>::CursorMut<T>, Error>;
+    fn cursor_mut<T: Table>(&self) -> Result<<Self as DbTxMutGAT<'_>>::CursorMut<T>, Error>;
     /// DupCursor mut.
     fn cursor_dup_mut<T: DupSort>(
         &self,
-    ) -> Result<<Self as DbTxMutGAT<'tx>>::DupCursorMut<T>, Error>;
+    ) -> Result<<Self as DbTxMutGAT<'_>>::DupCursorMut<T>, Error>;
 }
 
 /// Alias type for a `(key, value)` result coming from a cursor.
@@ -134,9 +134,6 @@ pub trait DbCursorRO<'tx, T: Table> {
     /// First item in table
     fn first(&mut self) -> PairResult<T>;
 
-    /// Seeks for a `(key, value)` pair greater or equal than `key`.
-    fn seek(&mut self, key: T::SeekKey) -> PairResult<T>;
-
     /// Seeks for the exact `(key, value)` pair with `key`.
     fn seek_exact(&mut self, key: T::Key) -> PairResult<T>;
 
@@ -164,6 +161,9 @@ pub trait DbCursorRO<'tx, T: Table> {
 
 /// Read only curor over DupSort table.
 pub trait DbDupCursorRO<'tx, T: DupSort> {
+    /// Seeks for a `(key, value)` pair greater or equal than `key`.
+    fn seek(&mut self, key: T::SubKey) -> PairResult<T>;
+
     /// Returns the next `(key, value)` pair of a DUPSORT table.
     fn next_dup(&mut self) -> PairResult<T>;
 

crates/interfaces/src/db/models/sharded_key.rs

@@ -12,7 +12,7 @@ use reth_primitives::TxNumber;
 /// `Address | 200` -> data is from transaction 0 to 200.
 ///
 /// `Address | 300` -> data is from transaction 201 to 300.
-#[derive(Debug, Default, Clone, PartialEq)]
+#[derive(Debug, Default, Clone, Eq, PartialEq)]
 pub struct ShardedKey<T> {
     /// The key for this type.
     pub key: T,

crates/interfaces/src/db/table.rs

@@ -62,8 +62,6 @@ pub trait Table: Send + Sync + Debug + 'static {
     type Key: Key;
     /// Value element of `Table`.
     type Value: Value;
-    /// Seek Key element of `Table`.
-    type SeekKey: Key;
 }
 
 /// DupSort allows for keys not to be repeated in the database,

crates/interfaces/src/db/tables.rs

@@ -9,7 +9,8 @@ use crate::db::{
     DupSort,
 };
 use reth_primitives::{
-    Account, Address, BlockNumber, Header, IntegerList, Receipt, StorageEntry, TxNumber, H256,
+    Account, Address, BlockHash, BlockNumber, Header, IntegerList, Receipt, StorageEntry, TxNumber,
+    H256,
 };
 
 /// Enum for the type of table present in libmdbx.
@@ -22,7 +23,7 @@ pub enum TableType {
 }
 
 /// Default tables that should be present inside database.
-pub const TABLES: [(TableType, &str); 19] = [
+pub const TABLES: [(TableType, &str); 20] = [
     (TableType::Table, CanonicalHeaders::const_name()),
     (TableType::Table, HeaderTD::const_name()),
     (TableType::Table, HeaderNumbers::const_name()),
@@ -35,6 +36,7 @@ pub const TABLES: [(TableType, &str); 19] = [
     (TableType::Table, Logs::const_name()),
     (TableType::Table, PlainAccountState::const_name()),
     (TableType::DupSort, PlainStorageState::const_name()),
+    (TableType::Table, Bytecodes::const_name()),
     (TableType::Table, AccountHistory::const_name()),
     (TableType::Table, StorageHistory::const_name()),
     (TableType::DupSort, AccountChangeSet::const_name()),
@@ -59,7 +61,6 @@ macro_rules! table {
             const NAME: &'static str = $name::const_name();
             type Key = $key;
             type Value = $value;
-            type SeekKey = $seek;
         }
 
         impl $name {
@@ -89,7 +90,7 @@ macro_rules! dupsort {
             $(#[$docs])+
             ///
             #[doc = concat!("`DUPSORT` table with subkey being: [`", stringify!($subkey), "`].")]
-            $name => $key => $value
+            $name => $key => $value => $subkey
         );
         impl DupSort for $name {
             type SubKey = $subkey;
@@ -111,7 +112,7 @@ table!(
 
 table!(
     /// Stores the block number corresponding to an header.
-    HeaderNumbers => BlockNumHash => BlockNumber);
+    HeaderNumbers => BlockHash => BlockNumber);
 
 table!(
     /// Stores header bodies.
@@ -145,6 +146,10 @@ table!(
     /// Stores the current state of an Account.
     PlainAccountState => Address => Account);
 
+table!(
+    /// Stores all smart contract bytecodes.
+    Bytecodes => H256 => Bytecode);
+
 dupsort!(
     /// Stores the current value of a storage key.
     PlainStorageState => Address => [H256] StorageEntry);
@@ -156,10 +161,10 @@ table!(
     /// use reth_primitives::{Address, IntegerList};
     /// use reth_interfaces::db::{DbTx, DbTxMut, DbCursorRO, Database, models::ShardedKey, tables::AccountHistory};
     /// use reth_db::{kv::{EnvKind, Env, test_utils}, mdbx::WriteMap};
-    /// use std::str::FromStr;
+    /// use std::{str::FromStr,sync::Arc};
     /// 
     /// fn main() {
-    ///     let db: Env<WriteMap> = test_utils::create_test_db(EnvKind::RW);
+    ///     let db: Arc<Env<WriteMap>> = test_utils::create_test_db(EnvKind::RW);
     ///     let account = Address::from_str("0xa2c122be93b0074270ebee7f6b7292c7deb45047").unwrap();
     /// 
     ///     // Setup if each shard can only take 1 transaction.
@@ -242,3 +247,5 @@ pub type RlpTotalDifficulty = Vec<u8>;
 pub type RlpTxBody = Vec<u8>;
 /// Temporary placeholder type for DB.
 pub type AddressStorageKey = Vec<u8>;
+/// Temporary placeholder type for DB.
+pub type Bytecode = Vec<u8>;

crates/interfaces/src/error.rs

@@ -13,4 +13,7 @@ pub enum Error {
 
     #[error(transparent)]
     Database(#[from] crate::db::Error),
+
+    #[error(transparent)]
+    Provider(#[from] crate::provider::Error),
 }

crates/interfaces/src/executor.rs

@@ -1,7 +1,5 @@
 use async_trait::async_trait;
-use reth_primitives::{
+use reth_primitives::Block;
-    Account, Address, Block, BlockNumber, Bytes, StorageKey, StorageValue, H256, U256,
-};
 use thiserror::Error;
 
 /// Takes block and executes it, returns error
@@ -20,21 +18,3 @@ pub enum Error {
     #[error("Example of error.")]
     VerificationFailed,
 }
-
-/// Function needed for executor
-pub trait ExecutorDb {
-    /// Get Block by BlockNumber.
-    fn block(&self, _height: BlockNumber) -> Option<Block>;
-
-    /// Get storage.
-    fn storage(&self, account: Address, storage_key: StorageKey) -> Option<StorageValue>;
-
-    /// Get basic account information.
-    fn basic_account(&self, adderss: Address) -> Option<Account>;
-
-    /// Get account code by its hash
-    fn bytecode_by_hash(&self, code_hash: H256) -> Option<(Bytes, usize)>;
-
-    /// Get block hash by number.
-    fn block_hash(&self, number: U256) -> Option<H256>;
-}

crates/interfaces/src/provider/db_provider.rs

@@ -1,30 +1,23 @@
-use crate::{
+//! Provider that wraps around database traits.
-    db::{tables, Database, DbTx},
+//! to provide higher level abstraction over database tables.
-    provider::HeaderProvider,
+
-};
+mod block;
+mod storage;
+use std::sync::Arc;
+
+pub use storage::{StateProviderImplHistory, StateProviderImplLatest};
+
+use crate::db::Database;
 
 /// Provider
-pub struct DbProvider<DB: Database> {
+pub struct ProviderImpl<DB: Database> {
     /// Database
-    db: DB,
+    db: Arc<DB>,
 }
 
-impl<DB: Database> DbProvider<DB> {
+impl<DB: Database> ProviderImpl<DB> {
     /// create new database provider
-    pub fn new(db: DB) -> Self {
+    pub fn new(db: Arc<DB>) -> Self {
         Self { db }
     }
 }
-
-impl<DB: Database> HeaderProvider for DbProvider<DB> {
-    fn header(
-        &self,
-        block_hash: &reth_primitives::BlockHash,
-    ) -> crate::Result<Option<reth_primitives::Header>> {
-        self.db.view(|tx| tx.get::<tables::Headers>((0, *block_hash).into()))?.map_err(Into::into)
-    }
-
-    fn is_known(&self, block_hash: &reth_primitives::BlockHash) -> crate::Result<bool> {
-        self.header(block_hash).map(|header| header.is_some())
-    }
-}

crates/interfaces/src/provider/db_provider/block.rs

@@ -0,0 +1,17 @@
+use crate::{
+    db::{tables, Database, DbTx},
+    provider::{HeaderProvider, ProviderImpl},
+};
+
+impl<DB: Database> HeaderProvider for ProviderImpl<DB> {
+    fn header(
+        &self,
+        block_hash: &reth_primitives::BlockHash,
+    ) -> crate::Result<Option<reth_primitives::Header>> {
+        self.db.view(|tx| tx.get::<tables::Headers>((0, *block_hash).into()))?.map_err(Into::into)
+    }
+
+    fn is_known(&self, block_hash: &reth_primitives::BlockHash) -> crate::Result<bool> {
+        self.header(block_hash).map(|header| header.is_some())
+    }
+}

crates/interfaces/src/provider/db_provider/storage.rs

@@ -0,0 +1,162 @@
+use super::ProviderImpl;
+use crate::{
+    db::{tables, Database, DatabaseGAT, DbCursorRO, DbDupCursorRO, DbTx},
+    provider::{Error, StateProvider, StateProviderFactory},
+    Result,
+};
+use reth_primitives::{
+    Account, Address, BlockHash, BlockNumber, Bytes, StorageKey, StorageValue, TxNumber, H256, U256,
+};
+use std::marker::PhantomData;
+
+impl<DB: Database> StateProviderFactory for ProviderImpl<DB> {
+    type HistorySP<'a> = StateProviderImplHistory<'a,<DB as DatabaseGAT<'a>>::TX> where Self: 'a;
+    type LatestSP<'a> = StateProviderImplLatest<'a,<DB as DatabaseGAT<'a>>::TX> where Self: 'a;
+    /// Storage provider for latest block
+    fn latest(&self) -> Result<Self::LatestSP<'_>> {
+        Ok(StateProviderImplLatest::new(self.db.tx()?))
+    }
+
+    fn history_by_block_number(&self, block_number: BlockNumber) -> Result<Self::HistorySP<'_>> {
+        let tx = self.db.tx()?;
+        // get block hash
+        let block_hash = tx
+            .get::<tables::CanonicalHeaders>(block_number)?
+            .ok_or(Error::BlockNumberNotExists { block_number })?;
+
+        // get transaction number
+        let block_num_hash = (block_number, block_hash);
+        let transaction_number = tx
+            .get::<tables::CumulativeTxCount>(block_num_hash.into())?
+            .ok_or(Error::BlockTxNumberNotExists { block_hash })?;
+
+        Ok(StateProviderImplHistory::new(tx, transaction_number))
+    }
+
+    fn history_by_block_hash(&self, block_hash: BlockHash) -> Result<Self::HistorySP<'_>> {
+        let tx = self.db.tx()?;
+        // get block number
+        let block_number = tx
+            .get::<tables::HeaderNumbers>(block_hash)?
+            .ok_or(Error::BlockHashNotExist { block_hash })?;
+
+        // get transaction number
+        let block_num_hash = (block_number, block_hash);
+        let transaction_number = tx
+            .get::<tables::CumulativeTxCount>(block_num_hash.into())?
+            .ok_or(Error::BlockTxNumberNotExists { block_hash })?;
+
+        Ok(StateProviderImplHistory::new(tx, transaction_number))
+    }
+}
+
+/// State provider with given hash
+pub struct StateProviderImplHistory<'a, TX: DbTx<'a>> {
+    /// Transaction
+    db: TX,
+    /// Transaction number is main indexer of account and storage changes
+    transaction_number: TxNumber,
+    _phantom: PhantomData<&'a TX>,
+}
+
+impl<'a, TX: DbTx<'a>> StateProviderImplHistory<'a, TX> {
+    /// Create new StateProvider from history transaction number
+    pub fn new(db: TX, transaction_number: TxNumber) -> Self {
+        Self { db, transaction_number, _phantom: PhantomData {} }
+    }
+}
+
+impl<'a, TX: DbTx<'a>> StateProvider for StateProviderImplHistory<'a, TX> {
+    /// Get storage.
+    fn storage(&self, account: Address, storage_key: StorageKey) -> Result<Option<StorageValue>> {
+        // TODO when StorageHistory is defined
+        let transaction_number =
+            self.db.get::<tables::StorageHistory>(Vec::new())?.map(|_integer_list|
+            // TODO select integer that is one less from transaction_number
+            self.transaction_number);
+
+        if transaction_number.is_none() {
+            return Ok(None)
+        }
+        let num = transaction_number.unwrap();
+        let mut cursor = self.db.cursor_dup::<tables::StorageChangeSet>()?;
+
+        if let Some((_, entry)) = cursor.seek_exact((num, account).into())? {
+            if entry.key == storage_key {
+                return Ok(Some(entry.value))
+            }
+
+            if let Some((_, entry)) = cursor.seek(storage_key)? {
+                if entry.key == storage_key {
+                    return Ok(Some(entry.value))
+                }
+            }
+        }
+        Ok(None)
+    }
+
+    /// Get basic account information.
+    fn basic_account(&self, _address: Address) -> Result<Option<Account>> {
+        // TODO add when AccountHistory is defined
+        Ok(None)
+    }
+
+    /// Get account code by its hash
+    fn bytecode_by_hash(&self, code_hash: H256) -> Result<Option<Bytes>> {
+        self.db.get::<tables::Bytecodes>(code_hash).map_err(Into::into).map(|r| r.map(Bytes::from))
+    }
+
+    /// Get block hash by number.
+    fn block_hash(&self, number: U256) -> Result<Option<H256>> {
+        self.db.get::<tables::CanonicalHeaders>(number.as_u64()).map_err(Into::into)
+    }
+}
+
+/// State Provider over latest state
+pub struct StateProviderImplLatest<'a, TX: DbTx<'a>> {
+    /// database transaction
+    db: TX,
+    /// Phantom data over lifetime
+    phantom: PhantomData<&'a TX>,
+}
+
+impl<'a, TX: DbTx<'a>> StateProviderImplLatest<'a, TX> {
+    /// Create new state provider
+    pub fn new(db: TX) -> Self {
+        Self { db, phantom: PhantomData {} }
+    }
+}
+
+impl<'a, TX: DbTx<'a>> StateProvider for StateProviderImplLatest<'a, TX> {
+    /// Get storage.
+    fn storage(&self, account: Address, storage_key: StorageKey) -> Result<Option<StorageValue>> {
+        let mut cursor = self.db.cursor_dup::<tables::PlainStorageState>()?;
+        if let Some((_, entry)) = cursor.seek_exact(account)? {
+            if entry.key == storage_key {
+                return Ok(Some(entry.value))
+            }
+
+            if let Some((_, entry)) = cursor.seek(storage_key)? {
+                if entry.key == storage_key {
+                    return Ok(Some(entry.value))
+                }
+            }
+        }
+        Ok(None)
+    }
+
+    /// Get basic account information.
+    fn basic_account(&self, address: Address) -> Result<Option<Account>> {
+        self.db.get::<tables::PlainAccountState>(address).map_err(Into::into)
+    }
+
+    /// Get account code by its hash
+    fn bytecode_by_hash(&self, code_hash: H256) -> Result<Option<Bytes>> {
+        self.db.get::<tables::Bytecodes>(code_hash).map_err(Into::into).map(|r| r.map(Bytes::from))
+    }
+
+    /// Get block hash by number.
+    fn block_hash(&self, number: U256) -> Result<Option<H256>> {
+        self.db.get::<tables::CanonicalHeaders>(number.as_u64()).map_err(Into::into)
+    }
+}

crates/interfaces/src/provider/error.rs

@@ -0,0 +1,13 @@
+use reth_primitives::{BlockHash, BlockNumber};
+
+/// KV error type. They are using u32 to represent error code.
+#[allow(missing_docs)]
+#[derive(Debug, thiserror::Error, PartialEq, Eq, Clone)]
+pub enum Error {
+    #[error("Block Number {block_number:?} does not exist in database")]
+    BlockNumberNotExists { block_number: BlockNumber },
+    #[error("Block tx cumulative number for hash {block_hash:?} does not exist in database")]
+    BlockTxNumberNotExists { block_hash: BlockHash },
+    #[error("Block hash {block_hash:?} does not exists in Headers table")]
+    BlockHashNotExist { block_hash: BlockHash },
+}

crates/interfaces/src/provider/mod.rs

@@ -1,7 +1,9 @@
 mod block;
-mod db_provider;
+pub mod db_provider;
+mod error;
 mod storage;
 
 pub use block::{BlockProvider, HeaderProvider};
-pub use db_provider::DbProvider;
+pub use db_provider::{self as db, ProviderImpl};
-pub use storage::StorageProvider;
+pub use error::Error;
+pub use storage::{StateProvider, StateProviderFactory, StorageProvider};

crates/interfaces/src/provider/storage.rs

@@ -1,8 +1,46 @@
 use crate::Result;
-use reth_primitives::{rpc::BlockId, Address, H256, U256};
+use reth_primitives::{
+    rpc::BlockId, Account, Address, BlockHash, BlockNumber, Bytes, StorageKey, StorageValue, H256,
+    U256,
+};
 
 /// Provides access to storage data
 pub trait StorageProvider: Send + Sync {
     /// Returns the value from a storage position at a given address and `BlockId`
     fn storage_at(&self, address: Address, index: U256, at: BlockId) -> Result<Option<H256>>;
 }
+
+/// Function needed for executor.
+pub trait StateProvider: Send + Sync {
+    /// Get storage.
+    fn storage(&self, account: Address, storage_key: StorageKey) -> Result<Option<StorageValue>>;
+
+    /// Get basic account information.
+    fn basic_account(&self, address: Address) -> Result<Option<Account>>;
+
+    /// Get account code by its hash
+    fn bytecode_by_hash(&self, code_hash: H256) -> Result<Option<Bytes>>;
+
+    /// Get block hash by number.
+    fn block_hash(&self, number: U256) -> Result<Option<H256>>;
+}
+
+/// Light wrapper that creates StateProvider.
+pub trait StateProviderFactory: Send + Sync {
+    /// History State provider.
+    type HistorySP<'a>: StateProvider
+    where
+        Self: 'a;
+    /// Latest state provider.
+    type LatestSP<'a>: StateProvider
+    where
+        Self: 'a;
+    /// Storage provider for latest block
+    fn latest(&self) -> Result<Self::LatestSP<'_>>;
+
+    /// History provider indexed by block number
+    fn history_by_block_number(&self, block: BlockNumber) -> Result<Self::HistorySP<'_>>;
+
+    /// History provider indexed by block hash
+    fn history_by_block_hash(&self, block: BlockHash) -> Result<Self::HistorySP<'_>>;
+}

crates/primitives/src/lib.rs

@@ -51,7 +51,7 @@ pub type TxNumber = u64;
 pub type StorageKey = H256;
 
 /// Storage value
-pub type StorageValue = H256;
+pub type StorageValue = U256;
 
 pub use ethers_core::{
     types as rpc,

crates/stages/src/stages/headers.rs

@@ -1,5 +1,5 @@
 use crate::{
-    util::unwind::{unwind_table_by_num, unwind_table_by_num_hash},
+    util::unwind::{unwind_table_by_num, unwind_table_by_num_hash, unwind_table_by_walker},
     DatabaseIntegrityError, ExecInput, ExecOutput, Stage, StageError, StageId, UnwindInput,
     UnwindOutput,
 };
@@ -91,7 +91,6 @@ impl<DB: Database, D: Downloader, C: Consensus, H: HeadersClient> Stage<DB>
                 }
             },
         };
-
         let stage_progress = self.write_headers::<DB>(tx, headers).await?.unwrap_or(last_block_num);
         Ok(ExecOutput { stage_progress, reached_tip: true, done: true })
     }
@@ -104,8 +103,13 @@ impl<DB: Database, D: Downloader, C: Consensus, H: HeadersClient> Stage<DB>
     ) -> Result<UnwindOutput, Box<dyn std::error::Error + Send + Sync>> {
         // TODO: handle bad block
         let tx = db.get_mut();
+
+        unwind_table_by_walker::<DB, tables::CanonicalHeaders, tables::HeaderNumbers>(
+            tx,
+            input.unwind_to + 1,
+        )?;
+
         unwind_table_by_num::<DB, tables::CanonicalHeaders>(tx, input.unwind_to)?;
-        unwind_table_by_num_hash::<DB, tables::HeaderNumbers>(tx, input.unwind_to)?;
         unwind_table_by_num_hash::<DB, tables::Headers>(tx, input.unwind_to)?;
         unwind_table_by_num_hash::<DB, tables::HeaderTD>(tx, input.unwind_to)?;
         Ok(UnwindOutput { stage_progress: input.unwind_to })
@@ -143,7 +147,6 @@ impl<D: Downloader, C: Consensus, H: HeadersClient> HeaderStage<D, C, H> {
         tx: &mut <DB as DatabaseGAT<'_>>::TXMut,
         headers: Vec<SealedHeader>,
     ) -> Result<Option<BlockNumber>, StageError> {
-        let mut cursor_header_number = tx.cursor_mut::<tables::HeaderNumbers>()?;
         let mut cursor_header = tx.cursor_mut::<tables::Headers>()?;
         let mut cursor_canonical = tx.cursor_mut::<tables::CanonicalHeaders>()?;
         let mut cursor_td = tx.cursor_mut::<tables::HeaderTD>()?;
@@ -157,14 +160,16 @@ impl<D: Downloader, C: Consensus, H: HeadersClient> HeaderStage<D, C, H> {
                 continue
             }
 
-            let key: BlockNumHash = (header.number, header.hash()).into();
+            let block_hash = header.hash();
+            let key: BlockNumHash = (header.number, block_hash).into();
             let header = header.unseal();
             latest = Some(header.number);
 
             td += header.difficulty;
 
             // TODO: investigate default write flags
-            cursor_header_number.append(key, header.number)?;
+            // NOTE: HeaderNumbers are not sorted and can't be inserted with cursor.
+            tx.put::<tables::HeaderNumbers>(block_hash, header.number)?;
             cursor_header.append(key, header)?;
             cursor_canonical.append(key.number(), key.hash())?;
             cursor_td.append(key, H256::from_uint(&td).as_bytes().to_vec())?;
@@ -352,7 +357,7 @@ mod tests {
             .expect("failed to check cannonical headers");
         runner
             .db()
-            .check_no_entry_above::<tables::HeaderNumbers, _>(unwind_to, |key| key.number())
+            .check_no_entry_above_by_value::<tables::HeaderNumbers, _>(unwind_to, |val| val)
             .expect("failed to check header numbers");
         runner
             .db()
@@ -445,9 +450,8 @@ mod tests {
                 I: Iterator<Item = &'a SealedHeader>,
             {
                 let headers = headers.collect::<Vec<_>>();
-                self.db.map_put::<tables::HeaderNumbers, _, _>(&headers, |h| {
+                self.db
-                    (BlockNumHash((h.number, h.hash())), h.number)
+                    .map_put::<tables::HeaderNumbers, _, _>(&headers, |h| (h.hash(), h.number))?;
-                })?;
                 self.db.map_put::<tables::Headers, _, _>(&headers, |h| {
                     (BlockNumHash((h.number, h.hash())), h.deref().clone().unseal())
                 })?;
@@ -475,7 +479,7 @@ mod tests {
                 let tx = db.get();
                 let key: BlockNumHash = (header.number, header.hash()).into();
 
-                let db_number = tx.get::<tables::HeaderNumbers>(key)?;
+                let db_number = tx.get::<tables::HeaderNumbers>(header.hash())?;
                 assert_eq!(db_number, Some(header.number));
 
                 let db_header = tx.get::<tables::Headers>(key)?;

crates/stages/src/util.rs

@@ -116,6 +116,24 @@ pub(crate) mod unwind {
         }
         Ok(())
     }
+
+    /// Unwind a table forward by a [Walker] on another table
+    pub(crate) fn unwind_table_by_walker<DB, T1, T2>(
+        tx: &mut <DB as DatabaseGAT<'_>>::TXMut,
+        start_at: T1::Key,
+    ) -> Result<(), Error>
+    where
+        DB: Database,
+        T1: Table,
+        T2: Table<Key = T1::Value>,
+    {
+        let mut cursor = tx.cursor_mut::<T1>()?;
+        let mut walker = cursor.walk(start_at)?;
+        while let Some((_, value)) = walker.next().transpose()? {
+            tx.delete::<T2>(value, None)?;
+        }
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -216,8 +234,9 @@ pub(crate) mod test_utils {
             Ok(())
         }
 
-        /// Check there there is no table entry above a given block
+        /// Check that there is no table entry above a given
-        pub(crate) fn check_no_entry_above<T: Table, F>(
+        /// block by [Table::Key]
+        pub(crate) fn check_no_entry_above<T, F>(
             &self,
             block: BlockNumber,
             mut selector: F,
@@ -236,6 +255,28 @@ pub(crate) mod test_utils {
 
             Ok(())
         }
+
+        /// Check that there is no table entry above a given
+        /// block by [Table::Value]
+        pub(crate) fn check_no_entry_above_by_value<T, F>(
+            &self,
+            block: BlockNumber,
+            mut selector: F,
+        ) -> Result<(), Error>
+        where
+            T: Table,
+            F: FnMut(T::Value) -> BlockNumber,
+        {
+            let db = self.container();
+            let tx = db.get();
+
+            let mut cursor = tx.cursor::<T>()?;
+            if let Some((_, value)) = cursor.last()? {
+                assert!(selector(value) <= block);
+            }
+
+            Ok(())
+        }
     }
 
     /// A generic test runner for stages.

rust-toolchain.toml

@@ -1,2 +0,0 @@
-[toolchain]
-channel = "nightly"