diff --git a/lean_client/Cargo.lock b/lean_client/Cargo.lock index 8f507b9..b52568f 100644 --- a/lean_client/Cargo.lock +++ b/lean_client/Cargo.lock @@ -68,7 +68,7 @@ dependencies = [ "rand 0.9.4", "rapidhash", "ruint", - "rustc-hash", + "rustc-hash 2.1.2", "serde", "sha3 0.10.8", ] @@ -627,6 +627,24 @@ version = "1.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "2af50177e190e07a26ab74f8b1efbfe2ef87da2116221318cb1c2e82baf7de06" +[[package]] +name = "bindgen" +version = "0.70.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f49d8fed880d473ea71efb9bf597651e77201bdd4893efe54c9e5d65ae04ce6f" +dependencies = [ + "bitflags", + "cexpr", + "clang-sys", + "itertools 0.13.0", + "proc-macro2 1.0.106", + "quote 1.0.45", + "regex", + "rustc-hash 1.1.0", + "shlex", + "syn 2.0.117", +] + [[package]] name = "bit-set" version = "0.8.0" @@ -798,16 +816,35 @@ dependencies = [ "serde", ] +[[package]] +name = "bytesize" +version = "2.4.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "49e78e506b9d7633710dab98996f22f95f3d0f488e8f1aa162830556ed9fc14d" +dependencies = [ + "serde_core", +] + [[package]] name = "cc" -version = "1.2.60" +version = "1.2.15" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "43c5703da9466b66a946814e1adf53ea2c90f10063b86290cc9eb67ce3478a20" +checksum = "c736e259eea577f443d5c86c304f9f4ae0295c43f3ba05c21f1d66b5f06001af" dependencies = [ - "find-msvc-tools", + "jobserver", + "libc", "shlex", ] +[[package]] +name = "cexpr" +version = "0.6.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6fac387a98bb7c37292057cffc56d62ecb629900026402633ae9160df93a8766" +dependencies = [ + "nom", +] + [[package]] name = "cfg-if" version = "1.0.4" @@ -882,6 +919,17 @@ dependencies = [ "zeroize", ] +[[package]] +name = "clang-sys" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0b023947811758c97c59bf9d1c188fd619ad4718dcaa767947df1cadb14f39f4" +dependencies = [ + "glob", + "libc", + "libloading", +] + [[package]] name = "clap" version = "4.6.0" @@ -1294,7 +1342,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ab67060fc6b8ef687992d439ca0fa36e7ed17e9a0b16b25b601e8757df720de" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.117", ] [[package]] @@ -1832,12 +1880,6 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" -[[package]] -name = "find-msvc-tools" -version = "0.1.9" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5baebc0774151f905a1a2cc41989300b1e6fbb29aff0ceffa1064fdd3088d582" - [[package]] name = "fixed-hash" version = "0.8.0" @@ -1899,6 +1941,15 @@ dependencies = [ "percent-encoding", ] +[[package]] +name = "fs-err" +version = "3.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "73fde052dbfc920003cfd2c8e2c6e6d4cc7c1091538c3a24226cec0665ab08c0" +dependencies = [ + "autocfg", +] + [[package]] name = "funty" version = "2.0.0" @@ -2865,6 +2916,16 @@ version = "1.0.18" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" +[[package]] +name = "jobserver" +version = "0.1.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9afb3de4395d6b3e67a780b6de64b51c978ecf11cb9a462c66be7d4ca9039d33" +dependencies = [ + "getrandom 0.3.4", + "libc", +] + [[package]] name = "js-sys" version = "0.3.95" @@ -2949,6 +3010,7 @@ dependencies = [ "parking_lot", "reqwest", "ssz", + "storage", "tikv-jemallocator", "tokio", "tracing", @@ -3077,6 +3139,16 @@ version = "0.2.185" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "52ff2c0fe9bc6cb6b14a0592c2ff4fa9ceb83eea9db979b0487cd054946a2b8f" +[[package]] +name = "libloading" +version = "0.8.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d7c4b02199fee7c5d21a5ae7d8cfa79a6ef5bb2fc834d6e9058e89c825efdc55" +dependencies = [ + "cfg-if", + "windows-link", +] + [[package]] name = "libm" version = "0.2.16" @@ -4341,6 +4413,12 @@ dependencies = [ "spki", ] +[[package]] +name = "pkg-config" +version = "0.3.33" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e" + [[package]] name = "polling" version = "3.11.0" @@ -4650,7 +4728,7 @@ dependencies = [ "pin-project-lite", "quinn-proto", "quinn-udp", - "rustc-hash", + "rustc-hash 2.1.2", "rustls", "socket2 0.6.3", "thiserror 2.0.18", @@ -4670,7 +4748,7 @@ dependencies = [ "lru-slab", "rand 0.9.4", "ring", - "rustc-hash", + "rustc-hash 2.1.2", "rustls", "rustls-pki-types", "slab", @@ -5016,6 +5094,31 @@ version = "0.7.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1e061d1b48cb8d38042de4ae0a7a6401009d6143dc80d2e2d6f31f0bdd6470c7" +[[package]] +name = "reth-libmdbx" +version = "1.3.12" +source = "git+https://github.com/paradigmxyz/reth.git?rev=6f8e7258f4733279080e4bd8345ce50538a40d6e#6f8e7258f4733279080e4bd8345ce50538a40d6e" +dependencies = [ + "bitflags", + "byteorder", + "derive_more", + "indexmap 2.14.0", + "parking_lot", + "reth-mdbx-sys", + "smallvec", + "thiserror 2.0.18", + "tracing", +] + +[[package]] +name = "reth-mdbx-sys" +version = "1.3.12" +source = "git+https://github.com/paradigmxyz/reth.git?rev=6f8e7258f4733279080e4bd8345ce50538a40d6e#6f8e7258f4733279080e4bd8345ce50538a40d6e" +dependencies = [ + "bindgen", + "cc", +] + [[package]] name = "rfc6979" version = "0.4.0" @@ -5131,6 +5234,12 @@ version = "1.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "48fd7bd8a6377e15ad9d42a8ec25371b94ddc67abe7c8b9127bec79bebaaae18" +[[package]] +name = "rustc-hash" +version = "1.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "08d43f7aa6b08d49f382cde6a7982047c3426db949b1424bc4b7ec9ae12c6ce2" + [[package]] name = "rustc-hash" version = "2.1.2" @@ -5722,6 +5831,21 @@ dependencies = [ "triomphe", ] +[[package]] +name = "storage" +version = "0.0.0" +dependencies = [ + "anyhow", + "bytesize", + "containers", + "fs-err", + "lz4_flex", + "reth-libmdbx", + "ssz", + "tempfile", + "zstd", +] + [[package]] name = "strength_reduce" version = "0.2.4" @@ -5896,7 +6020,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand", - "getrandom 0.3.4", + "getrandom 0.4.2", "once_cell", "rustix 1.1.4", "windows-sys 0.61.2", @@ -7299,3 +7423,31 @@ name = "zmij" version = "1.0.21" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b8848ee67ecc8aedbaf3e4122217aff892639231befc6a1b58d29fff4c2cabaa" + +[[package]] +name = "zstd" +version = "0.13.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e91ee311a569c327171651566e07972200e76fcfe2242a4fa446149a3881c08a" +dependencies = [ + "zstd-safe", +] + +[[package]] +name = "zstd-safe" +version = "7.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f49c4d5f0abb602a93fb8736af2a4f4dd9512e36f7f570d66e65ff867ed3b9d" +dependencies = [ + "zstd-sys", +] + +[[package]] +name = "zstd-sys" +version = "2.0.16+zstd.1.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91e19ebc2adc8f83e43039e79776e3fda8ca919132d68a1fed6a5faca2683748" +dependencies = [ + "cc", + "pkg-config", +] diff --git a/lean_client/Cargo.toml b/lean_client/Cargo.toml index 5a394ea..ac52e68 100644 --- a/lean_client/Cargo.toml +++ b/lean_client/Cargo.toml @@ -7,7 +7,8 @@ members = [ "http_api", "metrics", "networking", - "spec_test_fixtures", + "spec_test_fixtures", + "storage", "validator", "xmss", ] @@ -229,6 +230,7 @@ http_api = { path = "./http_api" } metrics = { path = "./metrics" } networking = { path = "./networking" } spec_test_fixtures = { path = "./spec_test_fixtures" } +storage = { path = "./storage" } validator = { path = "./validator" } xmss = { path = "./xmss" } @@ -237,12 +239,15 @@ async-trait = "0.1" axum = "0.8.8" bitvec = "1.0.1" bls = { git = "https://github.com/grandinetech/grandine", package = "bls", features = ["blst"], rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } +bytesize = { version = '2', features = ['serde'] } clap = { version = "4", features = ["derive"] } +database = { git = "https://github.com/grandinetech/grandine", package = "database", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } derive_more = "2.1.1" discv5 = "0.10.2" enr = { version = "0.13", features = ["k256"] } eth_ssz = { package = "ethereum_ssz", version = "0.10.0" } ethereum-types = "0.14" +fs-err = "3" futures = "0.3" features = { git = "https://github.com/grandinetech/grandine", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } git-version = "0.3" @@ -255,6 +260,7 @@ rec_aggregation = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e backend = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" } leansig = { git = "https://github.com/leanEthereum/leanSig", branch = "devnet4" } leansig_wrapper = { git = "https://github.com/leanEthereum/leanVM.git", rev = "e2592df4e30fdddbbf8ae26a333116c68cec7026" } +libmdbx = { git = "https://github.com/paradigmxyz/reth.git", package = "reth-libmdbx", rev = "6f8e7258f4733279080e4bd8345ce50538a40d6e" } libp2p = { git = "https://github.com/libp2p/rust-libp2p.git", rev = "91e8931e275bcd1c72791d18b09fea8b77209baf", default-features = false, features = [ 'dns', 'gossipsub', @@ -268,6 +274,7 @@ libp2p = { git = "https://github.com/libp2p/rust-libp2p.git", rev = "91e8931e275 'yamux' ] } libp2p-identity = { version = "0.2.13", features = ["secp256k1"] } +lz4_flex = "0.13" dedicated_executor = { path = "dedicated_executor" } num-bigint = "0.4" num-traits = "0.2" @@ -291,6 +298,7 @@ snap = "1.1" ssz = { git = "https://github.com/grandinetech/grandine", package = "ssz", rev = "64afdee3c6be79fceffb66933dcb69a943f3f1ae" } ssz-types = "0.3" itertools = "0.14" +tempfile = "3" test-generator = "0.3.1" thiserror = "2" tikv-jemallocator = { version = "0.6", features = ["stats", "unprefixed_malloc_on_supported_platforms"] } @@ -305,6 +313,7 @@ try_from_iterator = { git = "https://github.com/grandinetech/grandine", package typenum = "1.19" yamux = "0.12" zeroize = "1.8" +zstd = "0.13.3" [package] name = "lean_client" @@ -330,6 +339,7 @@ networking = { workspace = true } num_cpus = { workspace = true } parking_lot = { workspace = true } ssz = { workspace = true } +storage = { workspace = true } tokio = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true } diff --git a/lean_client/storage/Cargo.toml b/lean_client/storage/Cargo.toml new file mode 100644 index 0000000..2a05808 --- /dev/null +++ b/lean_client/storage/Cargo.toml @@ -0,0 +1,19 @@ +[package] +name = "storage" +edition = { workspace = true } + +[dependencies] +anyhow = { workspace = true } +bytesize = { workspace = true } +containers = { workspace = true } +fs-err = { workspace = true } +libmdbx = { workspace = true } +lz4_flex = { workspace = true } +ssz = { workspace = true } +zstd = { workspace = true } + +[dev-dependencies] +tempfile = { workspace = true } + +[lints] +workspace = true diff --git a/lean_client/storage/src/lib.rs b/lean_client/storage/src/lib.rs new file mode 100644 index 0000000..1943633 --- /dev/null +++ b/lean_client/storage/src/lib.rs @@ -0,0 +1,926 @@ +use std::{borrow::Cow, cell::RefCell, collections::HashSet, path::Path, sync::Arc}; + +use anyhow::Result; +use bytesize::ByteSize; + +use containers::{Block, Checkpoint, Slot, State}; +use ssz::{H256, SszReadDefault as _, SszWrite as _}; + +use libmdbx::{DatabaseFlags, Environment, Geometry, RW, Transaction, TransactionKind, WriteFlags}; + +const BLOCKS_TABLE_NAME: &str = "blocks"; +const STATES_TABLE_NAME: &str = "states"; +const CHECKPOINTS_TABLE_NAME: &str = "checkpoints"; +const SLOT_INDEX_TABLE_NAME: &str = "slot_index"; +const STATE_ROOT_INDEX_TABLE_NAME: &str = "state_root_index"; +const BLOCKS_BY_SLOT_TABLE_NAME: &str = "blocks_by_slot"; +const STATES_BY_SLOT_TABLE_NAME: &str = "states_by_slot"; + +const JUSTIFIED: u8 = 0; +const FINALIZED: u8 = 1; +const HEAD: u8 = 2; +const GENESIS_TIME: u8 = 3; + +pub struct Storage { + environment: Arc, + transaction: RefCell>>, + blocks: Database, + states: Database, + checkpoints: Database, + slot_index: Database, + state_root_index: Database, + blocks_by_slot: Database, + states_by_slot: Database, +} + +impl Storage { + pub fn new(base: impl AsRef) -> Result { + let base = base.as_ref(); + fs_err::create_dir_all(base)?; + + let environment = Arc::new( + Environment::builder() + .set_max_dbs(7) + .set_geometry(Geometry { + size: Some(..usize::try_from(ByteSize::gib(2).as_u64())?), + growth_step: Some(isize::try_from(ByteSize::mib(256).as_u64())?), + shrink_threshold: None, + page_size: None, + }) + .open_with_permissions(base, 0o600)?, + ); + + let txn = environment.begin_rw_txn()?; + for name in [ + BLOCKS_TABLE_NAME, + STATES_TABLE_NAME, + CHECKPOINTS_TABLE_NAME, + SLOT_INDEX_TABLE_NAME, + STATE_ROOT_INDEX_TABLE_NAME, + BLOCKS_BY_SLOT_TABLE_NAME, + STATES_BY_SLOT_TABLE_NAME, + ] { + txn.create_db(Some(name), DatabaseFlags::default())?; + } + txn.commit()?; + + Ok(Self { + blocks: Database::new(BLOCKS_TABLE_NAME, Compression::Lz4), + states: Database::new(STATES_TABLE_NAME, Compression::Zstd), + checkpoints: Database::new(CHECKPOINTS_TABLE_NAME, Compression::None), + slot_index: Database::new(SLOT_INDEX_TABLE_NAME, Compression::None), + state_root_index: Database::new(STATE_ROOT_INDEX_TABLE_NAME, Compression::None), + blocks_by_slot: Database::new(BLOCKS_BY_SLOT_TABLE_NAME, Compression::None), + states_by_slot: Database::new(STATES_BY_SLOT_TABLE_NAME, Compression::None), + environment, + transaction: RefCell::new(None), + }) + } + + pub fn batch_write(&self, f: impl FnOnce() -> Result<()>) -> Result<()> { + let txn = self.environment.begin_rw_txn()?; + *self.transaction.borrow_mut() = Some(txn); + + let result = f(); + + let txn = self + .transaction + .borrow_mut() + .take() + .expect("batch_write installed the ambient transaction"); + + match result { + Ok(()) => { + txn.commit()?; + Ok(()) + } + Err(error) => { + drop(txn); + Err(error) + } + } + } + + fn write(&self, f: impl FnOnce(&Transaction) -> Result) -> Result { + if let Some(txn) = self.transaction.borrow().as_ref() { + return f(txn); + } + let txn = self.environment.begin_rw_txn()?; + let result = f(&txn)?; + txn.commit()?; + Ok(result) + } + + fn read_bytes(&self, db: &Database, key: impl AsRef<[u8]>) -> Result>> { + if let Some(txn) = self.transaction.borrow().as_ref() { + return db.get(txn, key); + } + let txn = self.environment.begin_ro_txn()?; + let bytes = db.get(&txn, key)?; + txn.commit()?; + Ok(bytes) + } + + pub fn get_block(&self, root: H256) -> Result> { + match self.read_bytes(&self.blocks, root)? { + Some(bytes) => Ok(Some(Block::from_ssz_default(&bytes)?)), + None => Ok(None), + } + } + + pub fn put_block(&self, block: Block, root: H256) -> Result<()> { + let block_bytes = block.to_ssz()?; + let index_key = slot_root_key(block.slot, root); + self.write(|txn| { + self.blocks.put(txn, root, block_bytes)?; + self.blocks_by_slot.put(txn, index_key, b"")?; + Ok(()) + }) + } + + pub fn get_state(&self, root: H256) -> Result> { + match self.read_bytes(&self.states, root)? { + Some(bytes) => Ok(Some(State::from_ssz_default(&bytes)?)), + None => Ok(None), + } + } + + pub fn put_state(&self, state: State, root: H256) -> Result<()> { + let state_bytes = state.to_ssz()?; + let index_key = slot_root_key(state.slot, root); + self.write(|txn| { + self.states.put(txn, root, state_bytes)?; + self.states_by_slot.put(txn, index_key, b"")?; + Ok(()) + }) + } + + pub fn get_justified_checkpoint(&self) -> Result> { + match self.read_bytes(&self.checkpoints, [JUSTIFIED])? { + Some(bytes) => Ok(Some(Checkpoint::from_ssz_default(&bytes)?)), + None => Ok(None), + } + } + + pub fn put_justified_checkpoint(&self, checkpoint: Checkpoint) -> Result<()> { + let checkpoint_bytes = checkpoint.to_ssz()?; + self.write(|txn| self.checkpoints.put(txn, [JUSTIFIED], checkpoint_bytes)) + } + + pub fn get_finalized_checkpoint(&self) -> Result> { + match self.read_bytes(&self.checkpoints, [FINALIZED])? { + Some(bytes) => Ok(Some(Checkpoint::from_ssz_default(&bytes)?)), + None => Ok(None), + } + } + + pub fn put_finalized_checkpoint(&self, checkpoint: Checkpoint) -> Result<()> { + let checkpoint_bytes = checkpoint.to_ssz()?; + self.write(|txn| self.checkpoints.put(txn, [FINALIZED], checkpoint_bytes)) + } + + pub fn get_head_root(&self) -> Result> { + match self.read_bytes(&self.checkpoints, [HEAD])? { + Some(bytes) => Ok(Some(H256::from_slice(&bytes))), + None => Ok(None), + } + } + + pub fn put_head_root(&self, root: H256) -> Result<()> { + self.write(|txn| self.checkpoints.put(txn, [HEAD], root)) + } + + pub fn get_block_root_by_slot(&self, slot: Slot) -> Result> { + match self.read_bytes(&self.slot_index, slot.0.to_be_bytes())? { + Some(bytes) => Ok(Some(H256::from_slice(&bytes))), + None => Ok(None), + } + } + + pub fn put_block_root_by_slot(&self, slot: Slot, root: H256) -> Result<()> { + let slot_bytes = slot.0.to_be_bytes(); + self.write(|txn| self.slot_index.put(txn, slot_bytes, root)) + } + + pub fn get_block_root_by_state_root(&self, state_root: H256) -> Result> { + match self.read_bytes(&self.state_root_index, state_root)? { + Some(bytes) => Ok(Some(H256::from_slice(&bytes))), + None => Ok(None), + } + } + + pub fn put_block_root_by_state_root(&self, state_root: H256, block_root: H256) -> Result<()> { + self.write(|txn| self.state_root_index.put(txn, state_root, block_root)) + } + + pub fn get_genesis_time(&self) -> Result> { + match self.read_bytes(&self.checkpoints, [GENESIS_TIME])? { + Some(bytes) => Ok(Some(u64::from_ssz_default(&bytes)?)), + None => Ok(None), + } + } + + pub fn put_genesis_time(&self, genesis_time: u64) -> Result<()> { + let genesis_time_bytes = genesis_time.to_ssz()?; + self.write(|txn| { + self.checkpoints + .put(txn, [GENESIS_TIME], genesis_time_bytes) + }) + } + + pub fn prune_before_slot(&self, slot: Slot, keep_roots: &HashSet) -> Result { + let bound = slot.0; + self.write(|txn| { + let mut block_roots: Vec = Vec::new(); + let mut block_index_keys: Vec> = Vec::new(); + self.blocks_by_slot.for_each(txn, |key, _value| { + if slot_of(key) >= bound { + return Ok(false); + } + let root = H256::from_slice(&key[8..]); + if !keep_roots.contains(&root) { + block_roots.push(root); + block_index_keys.push(key.to_vec()); + } + Ok(true) + })?; + + let mut state_roots: Vec = Vec::new(); + let mut state_index_keys: Vec> = Vec::new(); + self.states_by_slot.for_each(txn, |key, _value| { + if slot_of(key) >= bound { + return Ok(false); + } + let root = H256::from_slice(&key[8..]); + if !keep_roots.contains(&root) { + state_roots.push(root); + state_index_keys.push(key.to_vec()); + } + Ok(true) + })?; + + let mut slot_index_keys: Vec> = Vec::new(); + self.slot_index.for_each(txn, |key, value| { + if slot_of(key) >= bound { + return Ok(false); + } + let canonical_root = H256::from_slice(value); + if !keep_roots.contains(&canonical_root) { + slot_index_keys.push(key.to_vec()); + } + Ok(true) + })?; + + let mut removed = 0; + removed += self.blocks.delete_batch(txn, &block_roots)?; + removed += self.blocks_by_slot.delete_batch(txn, &block_index_keys)?; + removed += self.slot_index.delete_batch(txn, &slot_index_keys)?; + removed += self.states.delete_batch(txn, &state_roots)?; + removed += self.states_by_slot.delete_batch(txn, &state_index_keys)?; + removed += self.state_root_index.delete_batch(txn, &state_roots)?; + + Ok(removed) + }) + } +} + +fn slot_root_key(slot: Slot, root: H256) -> [u8; 40] { + let mut key = [0u8; 40]; + key[..8].copy_from_slice(&slot.0.to_be_bytes()); + key[8..].copy_from_slice(root.as_ref()); + key +} + +fn slot_of(key: &[u8]) -> u64 { + let mut slot = [0u8; 8]; + slot.copy_from_slice(&key[..8]); + u64::from_be_bytes(slot) +} + +struct Database { + name: String, + compression: Compression, +} + +impl Database { + fn new(name: &str, compression: Compression) -> Self { + Self { + name: name.to_owned(), + compression, + } + } + + fn get( + &self, + txn: &Transaction, + key: impl AsRef<[u8]>, + ) -> Result>> { + let db = txn.open_db(Some(&self.name))?; + + txn.get::>(db.dbi(), key.as_ref())? + .map(|compressed| self.compression.decompress(&compressed)) + .transpose() + } + + fn put( + &self, + txn: &Transaction, + key: impl AsRef<[u8]>, + value: impl AsRef<[u8]>, + ) -> Result<()> { + let db = txn.open_db(Some(&self.name))?; + let compressed = self.compression.compress(value.as_ref())?; + txn.put(db.dbi(), key.as_ref(), compressed, WriteFlags::default())?; + Ok(()) + } + + fn for_each( + &self, + txn: &Transaction, + mut f: impl FnMut(&[u8], &[u8]) -> Result, + ) -> Result<()> { + let db = txn.open_db(Some(&self.name))?; + let mut cursor = txn.cursor(&db)?; + + while let Some((key, value)) = cursor.next::, Cow<[u8]>>()? { + let value = self.compression.decompress(&value)?; + if !f(key.as_ref(), &value)? { + break; + } + } + + Ok(()) + } + + fn delete_batch( + &self, + txn: &Transaction, + keys: impl IntoIterator>, + ) -> Result { + let db = txn.open_db(Some(&self.name))?; + let mut cursor = txn.cursor(&db)?; + + let mut deleted = 0; + for key in keys { + if cursor.set::<()>(key.as_ref())?.is_some() { + cursor.del(WriteFlags::default())?; + deleted += 1; + } + } + + Ok(deleted) + } +} + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +enum Compression { + None, + #[default] + Lz4, + Zstd, +} + +impl Compression { + fn compress(self, data: &[u8]) -> Result> { + match self { + Self::None => Ok(data.to_vec()), + Self::Lz4 => Ok(lz4_flex::compress_prepend_size(data)), + Self::Zstd => Ok(zstd::encode_all(data, 3)?), + } + } + + fn decompress(self, data: &[u8]) -> Result> { + match self { + Self::None => Ok(data.to_vec()), + Self::Lz4 => Ok(lz4_flex::decompress_size_prepended(data)?), + Self::Zstd => Ok(zstd::decode_all(data)?), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + use containers::BlockBody; + use ssz::SszWrite; + use tempfile::TempDir; + + fn storage() -> (TempDir, Storage) { + let dir = tempfile::tempdir().expect("failed to create temp dir"); + let storage = Storage::new(dir.path()).expect("failed to open storage"); + (dir, storage) + } + + fn h256(byte: u8) -> H256 { + H256::from_slice(&[byte; 32]) + } + + fn sample_block(seed: u8) -> Block { + Block { + slot: Slot(u64::from(seed)), + proposer_index: u64::from(seed), + parent_root: h256(seed), + state_root: h256(seed.wrapping_add(1)), + body: BlockBody::default(), + } + } + + fn assert_ssz_eq(left: &T, right: &T) { + assert_eq!( + left.to_ssz().expect("left should serialize"), + right.to_ssz().expect("right should serialize"), + ); + } + + #[test] + fn get_block_returns_none_when_absent() { + let (_dir, storage) = storage(); + assert!(storage.get_block(h256(9)).unwrap().is_none()); + } + + #[test] + fn put_then_get_block_roundtrips() { + let (_dir, storage) = storage(); + let root = h256(1); + let block = sample_block(1); + + storage.put_block(block.clone(), root).unwrap(); + + let read = storage + .get_block(root) + .unwrap() + .expect("block should exist"); + assert_ssz_eq(&block, &read); + } + + #[test] + fn put_and_get_multiple_blocks() { + let (_dir, storage) = storage(); + let blocks: Vec<(H256, Block)> = (0..8).map(|i| (h256(i), sample_block(i))).collect(); + + for (root, block) in &blocks { + storage.put_block(block.clone(), *root).unwrap(); + } + + for (root, block) in &blocks { + let read = storage + .get_block(*root) + .unwrap() + .expect("block should exist"); + assert_ssz_eq(block, &read); + } + } + + #[test] + fn put_block_overwrites_existing_root() { + let (_dir, storage) = storage(); + let root = h256(3); + + storage.put_block(sample_block(3), root).unwrap(); + let updated = sample_block(7); + storage.put_block(updated.clone(), root).unwrap(); + + let read = storage + .get_block(root) + .unwrap() + .expect("block should exist"); + assert_ssz_eq(&updated, &read); + } + + #[test] + fn get_state_returns_none_when_absent() { + let (_dir, storage) = storage(); + assert!(storage.get_state(h256(9)).unwrap().is_none()); + } + + #[test] + fn put_then_get_state_roundtrips() { + let (_dir, storage) = storage(); + let root = h256(2); + let state = State::generate_genesis(1_234, 3); + + storage.put_state(state.clone(), root).unwrap(); + + let read = storage + .get_state(root) + .unwrap() + .expect("state should exist"); + assert_ssz_eq(&state, &read); + } + + #[test] + fn put_and_get_multiple_states() { + let (_dir, storage) = storage(); + let states: Vec<(H256, State)> = (0..4) + .map(|i| (h256(i), State::generate_genesis(u64::from(i), u64::from(i)))) + .collect(); + + for (root, state) in &states { + storage.put_state(state.clone(), *root).unwrap(); + } + + for (root, state) in &states { + let read = storage + .get_state(*root) + .unwrap() + .expect("state should exist"); + assert_ssz_eq(state, &read); + } + } + + #[test] + fn justified_checkpoint_roundtrips() { + let (_dir, storage) = storage(); + assert!(storage.get_justified_checkpoint().unwrap().is_none()); + + let checkpoint = Checkpoint { + root: h256(4), + slot: Slot(10), + }; + storage + .put_justified_checkpoint(checkpoint.clone()) + .unwrap(); + + assert_eq!( + storage.get_justified_checkpoint().unwrap().unwrap(), + checkpoint + ); + } + + #[test] + fn finalized_checkpoint_roundtrips() { + let (_dir, storage) = storage(); + assert!(storage.get_finalized_checkpoint().unwrap().is_none()); + + let checkpoint = Checkpoint { + root: h256(5), + slot: Slot(20), + }; + storage + .put_finalized_checkpoint(checkpoint.clone()) + .unwrap(); + + assert_eq!( + storage.get_finalized_checkpoint().unwrap().unwrap(), + checkpoint + ); + } + + #[test] + fn head_root_roundtrips() { + let (_dir, storage) = storage(); + assert!(storage.get_head_root().unwrap().is_none()); + + let root = h256(6); + storage.put_head_root(root).unwrap(); + + assert_eq!(storage.get_head_root().unwrap().unwrap(), root); + } + + #[test] + fn genesis_time_roundtrips() { + let (_dir, storage) = storage(); + assert!(storage.get_genesis_time().unwrap().is_none()); + + storage.put_genesis_time(1_700_000_000).unwrap(); + + assert_eq!(storage.get_genesis_time().unwrap().unwrap(), 1_700_000_000); + } + + #[test] + fn checkpoints_database_keys_do_not_collide() { + let (_dir, storage) = storage(); + let justified = Checkpoint { + root: h256(1), + slot: Slot(1), + }; + let finalized = Checkpoint { + root: h256(2), + slot: Slot(2), + }; + let head = h256(3); + let genesis_time = 42; + + storage.put_justified_checkpoint(justified.clone()).unwrap(); + storage.put_finalized_checkpoint(finalized.clone()).unwrap(); + storage.put_head_root(head).unwrap(); + storage.put_genesis_time(genesis_time).unwrap(); + + assert_eq!( + storage.get_justified_checkpoint().unwrap().unwrap(), + justified + ); + assert_eq!( + storage.get_finalized_checkpoint().unwrap().unwrap(), + finalized + ); + assert_eq!(storage.get_head_root().unwrap().unwrap(), head); + assert_eq!(storage.get_genesis_time().unwrap().unwrap(), genesis_time); + } + + #[test] + fn block_root_by_slot_returns_none_when_absent() { + let (_dir, storage) = storage(); + assert!(storage.get_block_root_by_slot(Slot(0)).unwrap().is_none()); + } + + #[test] + fn block_root_by_slot_roundtrips() { + let (_dir, storage) = storage(); + let root = h256(6); + storage.put_block_root_by_slot(Slot(42), root).unwrap(); + + assert_eq!( + storage.get_block_root_by_slot(Slot(42)).unwrap().unwrap(), + root + ); + } + + #[test] + fn block_root_by_slot_handles_multiple() { + let (_dir, storage) = storage(); + for i in 0..10 { + storage + .put_block_root_by_slot(Slot(u64::from(i)), h256(i)) + .unwrap(); + } + + for i in 0..10 { + assert_eq!( + storage + .get_block_root_by_slot(Slot(u64::from(i))) + .unwrap() + .unwrap(), + h256(i), + ); + } + } + + #[test] + fn block_root_by_state_root_returns_none_when_absent() { + let (_dir, storage) = storage(); + assert!( + storage + .get_block_root_by_state_root(h256(7)) + .unwrap() + .is_none() + ); + } + + #[test] + fn block_root_by_state_root_roundtrips() { + let (_dir, storage) = storage(); + let state_root = h256(7); + let block_root = h256(8); + storage + .put_block_root_by_state_root(state_root, block_root) + .unwrap(); + + assert_eq!( + storage + .get_block_root_by_state_root(state_root) + .unwrap() + .unwrap(), + block_root, + ); + } + + #[test] + fn block_root_by_state_root_handles_multiple() { + let (_dir, storage) = storage(); + let pairs: Vec<(H256, H256)> = (0..10) + .map(|i| (h256(i), h256(i.wrapping_add(100)))) + .collect(); + + for (state_root, block_root) in &pairs { + storage + .put_block_root_by_state_root(*state_root, *block_root) + .unwrap(); + } + + for (state_root, block_root) in &pairs { + assert_eq!( + storage + .get_block_root_by_state_root(*state_root) + .unwrap() + .unwrap(), + *block_root, + ); + } + } + + fn state_at_slot(slot: u64) -> State { + let mut state = State::generate_genesis(1_234, 3); + state.slot = Slot(slot); + state + } + + #[test] + fn prune_removes_blocks_below_slot_and_keeps_the_rest() { + let (_dir, storage) = storage(); + for i in 0..10u8 { + storage.put_block(sample_block(i), h256(i)).unwrap(); + } + + let removed = storage.prune_before_slot(Slot(5), &HashSet::new()).unwrap(); + assert_eq!(removed, 10); + + for i in 0..5u8 { + assert!(storage.get_block(h256(i)).unwrap().is_none()); + } + for i in 5..10u8 { + assert!(storage.get_block(h256(i)).unwrap().is_some()); + } + } + + #[test] + fn prune_keeps_frozen_block_root() { + let (_dir, storage) = storage(); + for i in 0..5u8 { + storage.put_block(sample_block(i), h256(i)).unwrap(); + } + + let mut keep = HashSet::new(); + keep.insert(h256(2)); + + storage.prune_before_slot(Slot(5), &keep).unwrap(); + + assert!(storage.get_block(h256(2)).unwrap().is_some()); + for i in [0u8, 1, 3, 4] { + assert!(storage.get_block(h256(i)).unwrap().is_none()); + } + } + + #[test] + fn prune_removes_forked_blocks_at_same_slot() { + let (_dir, storage) = storage(); + let mut fork = sample_block(2); + fork.proposer_index = 99; + storage.put_block(sample_block(2), h256(20)).unwrap(); + storage.put_block(fork, h256(21)).unwrap(); + + let removed = storage.prune_before_slot(Slot(5), &HashSet::new()).unwrap(); + + assert_eq!(removed, 4); + assert!(storage.get_block(h256(20)).unwrap().is_none()); + assert!(storage.get_block(h256(21)).unwrap().is_none()); + } + + #[test] + fn prune_cleans_slot_index_unless_frozen() { + let (_dir, storage) = storage(); + storage.put_block(sample_block(2), h256(2)).unwrap(); + storage.put_block_root_by_slot(Slot(2), h256(2)).unwrap(); + storage.put_block(sample_block(3), h256(3)).unwrap(); + storage.put_block_root_by_slot(Slot(3), h256(3)).unwrap(); + + let mut keep = HashSet::new(); + keep.insert(h256(3)); + + storage.prune_before_slot(Slot(5), &keep).unwrap(); + + assert!(storage.get_block_root_by_slot(Slot(2)).unwrap().is_none()); + assert_eq!( + storage.get_block_root_by_slot(Slot(3)).unwrap().unwrap(), + h256(3), + ); + } + + #[test] + fn prune_removes_states_below_slot_and_keeps_frozen() { + let (_dir, storage) = storage(); + storage.put_state(state_at_slot(1), h256(1)).unwrap(); + storage.put_state(state_at_slot(2), h256(2)).unwrap(); + storage + .put_block_root_by_state_root(h256(1), h256(101)) + .unwrap(); + storage + .put_block_root_by_state_root(h256(2), h256(102)) + .unwrap(); + + let mut keep = HashSet::new(); + keep.insert(h256(2)); + + storage.prune_before_slot(Slot(5), &keep).unwrap(); + + assert!(storage.get_state(h256(1)).unwrap().is_none()); + assert!(storage.get_state(h256(2)).unwrap().is_some()); + assert!( + storage + .get_block_root_by_state_root(h256(1)) + .unwrap() + .is_none() + ); + assert_eq!( + storage + .get_block_root_by_state_root(h256(2)) + .unwrap() + .unwrap(), + h256(102), + ); + } + + #[test] + fn prune_protection_is_per_entity() { + let (_dir, storage) = storage(); + storage.put_block(sample_block(1), h256(1)).unwrap(); + storage.put_state(state_at_slot(1), h256(1)).unwrap(); + + let mut keep = HashSet::new(); + keep.insert(h256(1)); + + storage.prune_before_slot(Slot(5), &keep).unwrap(); + + assert!(storage.get_block(h256(1)).unwrap().is_some()); + assert!(storage.get_state(h256(1)).unwrap().is_some()); + } + + #[test] + fn prune_below_lowest_slot_is_a_noop() { + let (_dir, storage) = storage(); + for i in 0..5u8 { + storage.put_block(sample_block(i), h256(i)).unwrap(); + } + + let removed = storage.prune_before_slot(Slot(0), &HashSet::new()).unwrap(); + assert_eq!(removed, 0); + for i in 0..5u8 { + assert!(storage.get_block(h256(i)).unwrap().is_some()); + } + } + + #[test] + fn batch_write_commits_all_on_success() { + let (_dir, storage) = storage(); + + storage + .batch_write(|| { + storage.put_block(sample_block(1), h256(1))?; + storage.put_state(state_at_slot(1), h256(2))?; + storage.put_head_root(h256(3))?; + Ok(()) + }) + .unwrap(); + + assert!(storage.get_block(h256(1)).unwrap().is_some()); + assert!(storage.get_state(h256(2)).unwrap().is_some()); + assert_eq!(storage.get_head_root().unwrap().unwrap(), h256(3)); + } + + #[test] + fn batch_write_rolls_back_every_write_on_error() { + let (_dir, storage) = storage(); + + let result = storage.batch_write(|| { + storage.put_block(sample_block(1), h256(1))?; + storage.put_state(state_at_slot(1), h256(2))?; + Err(anyhow::anyhow!("intentional failure")) + }); + + assert!(result.is_err()); + assert!(storage.get_block(h256(1)).unwrap().is_none()); + assert!(storage.get_state(h256(2)).unwrap().is_none()); + } + + #[test] + fn storage_is_usable_after_batch_write() { + let (_dir, storage) = storage(); + + storage + .batch_write(|| storage.put_block(sample_block(1), h256(1))) + .unwrap(); + + storage.put_block(sample_block(2), h256(2)).unwrap(); + + assert!(storage.get_block(h256(1)).unwrap().is_some()); + assert!(storage.get_block(h256(2)).unwrap().is_some()); + } + + #[test] + fn storage_is_usable_after_failed_batch_write() { + let (_dir, storage) = storage(); + + assert!( + storage + .batch_write(|| Err(anyhow::anyhow!("boom"))) + .is_err() + ); + + storage.put_block(sample_block(5), h256(5)).unwrap(); + assert!(storage.get_block(h256(5)).unwrap().is_some()); + } + + #[test] + fn batch_write_reads_see_pending_writes() { + let (_dir, storage) = storage(); + + storage + .batch_write(|| { + storage.put_block(sample_block(7), h256(7)).unwrap(); + assert!(storage.get_block(h256(7)).unwrap().is_some()); + Ok(()) + }) + .unwrap(); + } +}