refactor(storage): add tests (WIP)

.gitignore

@@ -8,4 +8,7 @@ target/
 .fleet/
 
 perf.data*
-.scratch
+.scratch
+
+.DS_Store
+storage/tmp/

storage/Cargo.toml

@@ -36,5 +36,12 @@ weedb = "0.1.1"
 tycho-block-util = { path = "../block-util" }
 tycho-util = { path = "../util" }
 
+[dev-dependencies]
+base64 = "0.22.0"
+serde_json = "1.0.114"
+tracing-appender = "0.2.3"
+tracing-subscriber = { version = "0.3", features = ["env-filter"] }
+tracing-test = "0.2"
+
 [lints]
 workspace = true

storage/src/db/kv_db/mod.rs

@@ -10,6 +10,8 @@ use weedb::{Caches, WeeDb};
 pub use weedb::Stats as RocksdbStats;
 pub use weedb::{rocksdb, BoundedCfHandle, ColumnFamily, Table};
 
+pub use self::config::DbOptions;
+
 pub mod refcount;
 pub mod tables;
 
@@ -247,56 +249,3 @@ impl Drop for Db {
         self.raw().cancel_all_background_work(true);
     }
 }
-
-#[derive(Debug, Copy, Clone, Serialize, Deserialize)]
-#[serde(deny_unknown_fields, default)]
-pub struct DbOptions {
-    pub rocksdb_lru_capacity: ByteSize,
-    pub cells_cache_size: ByteSize,
-}
-
-impl Default for DbOptions {
-    fn default() -> Self {
-        // Fetch the currently available memory in bytes
-        let available = {
-            let mut sys = sysinfo::System::new();
-            sys.refresh_memory();
-            sys.available_memory()
-        };
-
-        // Estimated memory usage of components other than cache:
-        // - 2 GiBs for write buffers(4 if we are out of luck and all memtables are being flushed at the same time)
-        // - 2 GiBs for indexer logic
-        // - 10 bits per cell for bloom filter. Realistic case is 100M cells, so 0.25 GiBs
-        // - 1/3 of all available memory is reserved for kernel buffers
-        const WRITE_BUFFERS: ByteSize = ByteSize::gib(2);
-        const INDEXER_LOGIC: ByteSize = ByteSize::gib(2);
-        const BLOOM_FILTER: ByteSize = ByteSize::mib(256);
-        let estimated_memory_usage = WRITE_BUFFERS + INDEXER_LOGIC + BLOOM_FILTER + available / 3;
-
-        // Reduce the available memory by the fixed offset
-        let available = available
-            .checked_sub(estimated_memory_usage.as_u64())
-            .unwrap_or_else(|| {
-                tracing::error!(
-                    "Not enough memory for cache, using 1/4 of all available memory. \
-                    Tweak `db_options` in config to improve performance."
-                );
-                available / 4
-            });
-
-        // We will use 3/4 of available memory for the cells cache (at most 4 GB).
-        let cells_cache_size = std::cmp::min(ByteSize(available * 4 / 3), ByteSize::gib(4));
-
-        // The reset of the memory is used for LRU cache (at least 128 MB)
-        let rocksdb_lru_capacity = std::cmp::max(
-            ByteSize(available.saturating_sub(cells_cache_size.as_u64())),
-            ByteSize::mib(128),
-        );
-
-        Self {
-            rocksdb_lru_capacity,
-            cells_cache_size,
-        }
-    }
-}

storage/src/lib.rs

@@ -23,15 +23,69 @@ pub struct Storage {
 }
 
 impl Storage {
+    pub fn new(
+        db: Arc<Db>,
+        file_db_path: PathBuf,
+        max_cell_cache_size_bytes: u64,
+    ) -> anyhow::Result<Arc<Self>> {
+        let block_handle_storage = Arc::new(BlockHandleStorage::new(db.clone()));
+        let runtime_storage = Arc::new(RuntimeStorage::new(block_handle_storage.clone()));
+        let block_storage = Arc::new(BlockStorage::new(db.clone(), block_handle_storage.clone())?);
+        let shard_state_storage = ShardStateStorage::new(
+            db.clone(),
+            block_handle_storage.clone(),
+            block_storage.clone(),
+            file_db_path.clone(),
+            max_cell_cache_size_bytes,
+        )?;
+        let persistent_state_storage = PersistentStateStorage::new(
+            file_db_path.clone(),
+            db.clone(),
+            block_handle_storage.clone(),
+        )?;
+        let node_state_storage = NodeStateStorage::new(db.clone());
+        let block_connection_storage = BlockConnectionStorage::new(db);
+
+        Ok(Arc::new(Self {
+            file_db_path,
+
+            block_handle_storage,
+            block_storage,
+            shard_state_storage,
+            persistent_state_storage,
+            block_connection_storage,
+            node_state_storage,
+            runtime_storage,
+        }))
+    }
+
+    #[inline(always)]
+    pub fn runtime_storage(&self) -> &RuntimeStorage {
+        &self.runtime_storage
+    }
+
+    #[inline(always)]
+    pub fn persistent_state_storage(&self) -> &PersistentStateStorage {
+        &self.persistent_state_storage
+    }
+
+    #[inline(always)]
     pub fn block_handle_storage(&self) -> &BlockHandleStorage {
         &self.block_handle_storage
     }
 
+    #[inline(always)]
     pub fn block_connection_storage(&self) -> &BlockConnectionStorage {
         &self.block_connection_storage
     }
 
+    #[inline(always)]
     pub fn shard_state_storage(&self) -> &ShardStateStorage {
         &self.shard_state_storage
     }
+
+    #[inline(always)]
+    pub fn node_state(&self) -> &NodeStateStorage {
+        &self.node_state_storage
+    }
 }

storage/src/store/block_connection/mod.rs

@@ -13,8 +13,8 @@ pub struct BlockConnectionStorage {
 }
 
 impl BlockConnectionStorage {
-    pub fn new(db: Arc<Db>) -> Result<Self> {
-        Ok(Self { db })
+    pub fn new(db: Arc<Db>) -> Self {
+        Self { db }
     }
 
     pub fn store_connection(

storage/src/store/block_handle/mod.rs

@@ -16,11 +16,11 @@ pub struct BlockHandleStorage {
 }
 
 impl BlockHandleStorage {
-    pub fn new(db: Arc<Db>) -> Result<Self> {
-        Ok(Self {
+    pub fn new(db: Arc<Db>) -> Self {
+        Self {
             db,
             cache: Arc::new(Default::default()),
-        })
+        }
     }
 
     pub fn store_block_applied(&self, handle: &Arc<BlockHandle>) -> Result<bool> {

storage/src/store/node_state/mod.rs

@@ -15,13 +15,13 @@ pub struct NodeStateStorage {
 }
 
 impl NodeStateStorage {
-    pub fn new(db: Arc<Db>) -> Result<Self> {
-        Ok(Self {
+    pub fn new(db: Arc<Db>) -> Self {
+        Self {
             db,
             last_mc_block_id: (Default::default(), LAST_MC_BLOCK_ID),
             init_mc_block_id: (Default::default(), INIT_MC_BLOCK_ID),
             shards_client_mc_block_id: (Default::default(), SHARDS_CLIENT_MC_BLOCK_ID),
-        })
+        }
     }
 
     pub fn store_historical_sync_start(&self, id: &BlockId) -> Result<()> {

storage/src/store/persistent_state/mod.rs

@@ -139,7 +139,6 @@ impl PersistentStateStorage {
         self.storage_path
             .clone()
             .join(mc_block_id.seqno.to_string())
-            .join(block_id.root_hash.to_string())
     }
 
     pub fn cancel(&self) {

storage/src/store/shard_state/cell_storage.rs

@@ -20,20 +20,20 @@ pub struct CellStorage {
 }
 
 impl CellStorage {
-    pub fn new(db: Arc<Db>, cache_size_bytes: u64) -> Result<Arc<Self>> {
+    pub fn new(db: Arc<Db>, cache_size_bytes: u64) -> Arc<Self> {
         let cells_cache = Default::default();
         let raw_cells_cache = RawCellsCache::new(cache_size_bytes);
 
-        Ok(Arc::new(Self {
+        Arc::new(Self {
             db,
             cells_cache,
             raw_cells_cache,
-        }))
+        })
     }
 
     pub fn store_cell(
         &self,
-        batch: &mut weedb::rocksdb::WriteBatch,
+        batch: &mut rocksdb::WriteBatch,
         root: Cell,
     ) -> Result<usize, CellStorageError> {
         struct CellWithRefs<'a> {

storage/src/store/shard_state/mod.rs

@@ -46,7 +46,7 @@ impl ShardStateStorage {
         cache_size_bytes: u64,
     ) -> Result<Self> {
         let downloads_dir = prepare_file_db_dir(file_db_path, "downloads")?;
-        let cell_storage = CellStorage::new(db.clone(), cache_size_bytes)?;
+        let cell_storage = CellStorage::new(db.clone(), cache_size_bytes);
 
         let res = Self {
             db,

storage/tests/global-config.json

@@ -0,0 +1,22 @@
+{
+  "@type": "config.global",
+  "dht": {
+    "@type": "dht.config.global",
+    "k": 6,
+    "a": 3,
+    "static_nodes": {
+      "@type": "dht.nodes",
+      "nodes": []
+    }
+  },
+  "validator": {
+    "@type": "validator.config.global",
+    "zero_state": {
+      "workchain": -1,
+      "shard": -9223372036854775808,
+      "seqno": 0,
+      "root_hash": "WP/KGheNr/cF3lQhblQzyb0ufYUAcNM004mXhHq56EU=",
+      "file_hash": "0nC4eylStbp9qnCq8KjDYb789NjS25L5ZA1UQwcIOOQ="
+    }
+  }
+}