diff --git a/lean_client/Cargo.lock b/lean_client/Cargo.lock index 276680bf..9c8ea193 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" @@ -800,14 +818,24 @@ dependencies = [ [[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 +910,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 +1333,19 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "7ab67060fc6b8ef687992d439ca0fa36e7ed17e9a0b16b25b601e8757df720de" dependencies = [ "data-encoding", - "syn 1.0.109", + "syn 2.0.117", +] + +[[package]] +name = "database" +version = "0.0.0" +dependencies = [ + "anyhow", + "containers", + "lz4_flex", + "reth-libmdbx", + "ssz", + "zstd", ] [[package]] @@ -1832,12 +1883,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" @@ -1875,6 +1920,7 @@ dependencies = [ "anyhow", "bls", "containers", + "database", "env-config", "indexmap 2.14.0", "metrics", @@ -2369,6 +2415,7 @@ dependencies = [ "axum", "clap", "containers", + "database", "fork_choice", "futures", "hex", @@ -2865,6 +2912,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" @@ -2935,6 +2992,7 @@ dependencies = [ "chain", "clap", "containers", + "database", "dedicated_executor", "ethereum-types", "features", @@ -3077,6 +3135,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 +4409,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 +4724,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 +4744,7 @@ dependencies = [ "lru-slab", "rand 0.9.4", "ring", - "rustc-hash", + "rustc-hash 2.1.2", "rustls", "rustls-pki-types", "slab", @@ -5016,6 +5090,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 +5230,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" @@ -5896,7 +6001,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 +7404,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 83eb13e9..aa95194d 100644 --- a/lean_client/Cargo.toml +++ b/lean_client/Cargo.toml @@ -2,6 +2,7 @@ members = [ "chain", "containers", + "database", "env-config", "fork_choice", "http_api", @@ -225,6 +226,7 @@ private_intra_doc_links = 'allow' [workspace.dependencies] chain = { path = "./chain" } containers = { path = "./containers" } +database = { path = "./database" } env-config = { path = "./env-config" } fork_choice = { path = "./fork_choice" } http_api = { path = "./http_api" } @@ -239,12 +241,14 @@ 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"] } 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" @@ -257,6 +261,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', @@ -270,6 +275,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" @@ -293,6 +299,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"] } @@ -307,6 +314,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" @@ -319,6 +327,7 @@ bls = { workspace = true } chain = { workspace = true } clap = { workspace = true } containers = { workspace = true } +database = { workspace = true } dedicated_executor = { workspace = true } ethereum-types = { workspace = true } features = { workspace = true } diff --git a/lean_client/database/Cargo.toml b/lean_client/database/Cargo.toml new file mode 100644 index 00000000..729b8efd --- /dev/null +++ b/lean_client/database/Cargo.toml @@ -0,0 +1,14 @@ +[package] +name = "database" +edition = { workspace = true } + +[dependencies] +anyhow = { workspace = true } +containers = { workspace = true } +libmdbx = { workspace = true } +lz4_flex = { workspace = true } +ssz = { workspace = true } +zstd = { workspace = true } + +[lints] +workspace = true diff --git a/lean_client/database/src/lib.rs b/lean_client/database/src/lib.rs new file mode 100644 index 00000000..9b33d034 --- /dev/null +++ b/lean_client/database/src/lib.rs @@ -0,0 +1,584 @@ +use std::{ + borrow::Cow, + collections::HashMap, + marker::PhantomData, + ops::{Bound, RangeBounds}, + path::PathBuf, + sync::{Arc, Mutex, OnceLock, Weak}, +}; + +use anyhow::{Ok, Result}; + +use containers::Slot; +use ssz::{H256, SszReadDefault, SszWrite}; + +use libmdbx::{ + DatabaseFlags, Environment, EnvironmentFlags, EnvironmentKind, Geometry, Mode, SyncMode, + WriteFlags, +}; + +const GIB: usize = 1 << 30; +const MIB: usize = 1 << 20; + +pub const BLOCKS_TABLE_NAME: &str = "blocks"; +pub const GENESIS_STATE_TABLE_NAME: &str = "genesis_state"; + +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub 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)?), + } + } +} + +pub struct EnvironmentBuilder { + path: PathBuf, + max_size: usize, +} + +impl EnvironmentBuilder { + pub fn new(path: impl Into, gib: usize) -> Self { + Self { + path: path.into(), + max_size: gib * GIB, + } + } + + pub fn build(&self) -> Result> { + static ENVIRONMENTS: OnceLock>>> = OnceLock::new(); + + std::fs::create_dir_all(&self.path)?; + let canonical = self.path.canonicalize()?; + + let mut environments = ENVIRONMENTS + .get_or_init(|| Mutex::new(HashMap::new())) + .lock() + .unwrap_or_else(std::sync::PoisonError::into_inner); + + if let Some(env) = environments.get(&canonical).and_then(Weak::upgrade) { + return Ok(env); + } + + let env = Environment::builder() + .set_max_dbs(2) + .set_kind(EnvironmentKind::WriteMap) + .set_flags(EnvironmentFlags { + mode: Mode::ReadWrite { + sync_mode: SyncMode::SafeNoSync, + }, + no_meminit: true, + coalesce: true, + liforeclaim: true, + ..Default::default() + }) + .set_geometry(Geometry { + size: Some(0..self.max_size), + growth_step: Some((256 * MIB) as isize), + shrink_threshold: None, + page_size: None, + }) + .open(&canonical)?; + + let env = Arc::new(env); + environments.insert(canonical, Arc::downgrade(&env)); + Ok(env) + } +} + +#[derive(Debug, Clone)] +pub struct Database { + env: Arc, + name: &'static str, + compression: Compression, + _marker: PhantomData V>, +} + +impl Database { + pub fn new( + env: Arc, + name: &'static str, + compression: Compression, + ) -> Result { + let txn = env.begin_rw_txn()?; + txn.create_db(Some(name), DatabaseFlags::default())?; + txn.commit()?; + Ok(Self { + env, + name, + compression, + _marker: PhantomData, + }) + } + + pub fn get(&self, key: &K) -> Result> { + let txn = self.env.begin_ro_txn()?; + let db = txn.open_db(Some(self.name))?; + + let key_bytes = key.encode(); + let Some(compressed) = txn.get::>(db.dbi(), &key_bytes)? else { + return Ok(None); + }; + + let bytes = self.compression.decompress(&compressed)?; + Ok(Some(V::decode(&bytes)?)) + } + + pub fn put(&self, key: &K, value: &V) -> Result<()> { + let txn = self.env.begin_rw_txn()?; + let db = txn.open_db(Some(self.name))?; + + let compressed = self.compression.compress(&value.encode()?)?; + txn.put(db.dbi(), key.encode(), compressed, WriteFlags::default())?; + + txn.commit()?; + Ok(()) + } + + pub fn range(&self, bounds: impl RangeBounds

) -> Result> { + let txn = self.env.begin_ro_txn()?; + let db = txn.open_db(Some(self.name))?; + let mut cursor = txn.cursor(&db)?; + + let start = bounds.start_bound().map(|p| p.encode()); + let end = bounds.end_bound().map(|p| p.encode()); + + let mut entry = match &start { + Bound::Included(s) | Bound::Excluded(s) => { + cursor.set_range::, Cow<[u8]>>(s)? + } + Bound::Unbounded => cursor.first::, Cow<[u8]>>()?, + }; + + let mut out = Vec::new(); + while let Some((key, value)) = entry { + if let Bound::Excluded(s) = &start { + if key.len() >= s.len() && &key[..s.len()] == s.as_slice() { + entry = cursor.next::, Cow<[u8]>>()?; + continue; + } + } + let stop = match &end { + Bound::Included(e) => key.len() >= e.len() && &key[..e.len()] > e.as_slice(), + Bound::Excluded(e) => key.len() >= e.len() && &key[..e.len()] >= e.as_slice(), + Bound::Unbounded => false, + }; + if stop { + break; + } + + out.push(V::decode(&self.compression.decompress(&value)?)?); + entry = cursor.next::, Cow<[u8]>>()?; + } + Ok(out) + } + + pub fn delete_range(&self, bounds: impl RangeBounds

) -> Result { + let txn = self.env.begin_rw_txn()?; + let db = txn.open_db(Some(self.name))?; + + let start = bounds.start_bound().map(|p| p.encode()); + let end = bounds.end_bound().map(|p| p.encode()); + + let mut keys: Vec> = Vec::new(); + { + let mut cursor = txn.cursor(&db)?; + let mut entry = match &start { + Bound::Included(s) | Bound::Excluded(s) => cursor.set_range::, ()>(s)?, + Bound::Unbounded => cursor.first::, ()>()?, + }; + while let Some((key, ())) = entry { + if let Bound::Excluded(s) = &start { + if key.len() >= s.len() && &key[..s.len()] == s.as_slice() { + entry = cursor.next::, ()>()?; + continue; + } + } + let stop = match &end { + Bound::Included(e) => key.len() >= e.len() && &key[..e.len()] > e.as_slice(), + Bound::Excluded(e) => key.len() >= e.len() && &key[..e.len()] >= e.as_slice(), + Bound::Unbounded => false, + }; + if stop { + break; + } + + keys.push(key.into_owned()); + entry = cursor.next::, ()>()?; + } + } + + let mut cursor = txn.cursor(&db)?; + let mut deleted = 0; + for key in &keys { + if cursor.set::<()>(key)?.is_some() { + cursor.del(WriteFlags::default())?; + deleted += 1; + } + } + + txn.commit()?; + Ok(deleted) + } +} + +pub trait Key { + fn encode(&self) -> Vec; +} + +pub trait KeyPrefix { + fn encode(&self) -> Vec; +} + +pub trait Value: Sized { + fn encode(&self) -> Result>; + fn decode(bytes: &[u8]) -> Result; +} + +impl Value for T { + fn encode(&self) -> Result> { + Ok(self.to_ssz()?) + } + + fn decode(bytes: &[u8]) -> Result { + Ok(Self::from_ssz_default(bytes)?) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct BlockKey { + pub slot: Slot, + pub root: H256, +} + +impl Key for BlockKey { + fn encode(&self) -> Vec { + let mut b = Vec::with_capacity(40); + b.extend_from_slice(&self.slot.0.to_be_bytes()); + b.extend_from_slice(self.root.as_ref()); + b + } +} + +#[derive(Clone, Copy, Debug)] +pub struct BySlot(pub Slot); + +impl KeyPrefix for BySlot { + fn encode(&self) -> Vec { + self.0.0.to_be_bytes().to_vec() + } +} + +const STATE_KEY_PREFIX: &[u8] = b"state"; + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub struct StateKey(pub Slot); + +impl Key for StateKey { + fn encode(&self) -> Vec { + let mut b = Vec::with_capacity(STATE_KEY_PREFIX.len() + 8); + b.extend_from_slice(STATE_KEY_PREFIX); + b.extend_from_slice(&self.0.0.to_be_bytes()); + b + } +} + +impl KeyPrefix for StateKey { + fn encode(&self) -> Vec { + Key::encode(self) + } +} + +#[derive(Clone, Copy, Debug)] +pub struct StateKeyPrefix; + +impl KeyPrefix for StateKeyPrefix { + fn encode(&self) -> Vec { + STATE_KEY_PREFIX.to_vec() + } +} + +#[cfg(test)] +mod tests { + use containers::{Block, BlockBody, State, Validator}; + + use super::*; + + fn test_db_path(name: &str) -> PathBuf { + let path = std::env::temp_dir().join(format!( + "lean_database_test_{}_{}", + std::process::id(), + name + )); + std::fs::remove_dir_all(&path).ok(); + path + } + + fn test_env(name: &str) -> Arc { + EnvironmentBuilder::new(test_db_path(name), 1) + .build() + .expect("failed to open test environment") + } + + fn blocks_db(env: Arc) -> Database { + Database::new(env, BLOCKS_TABLE_NAME, Compression::Lz4) + .expect("failed to open blocks table") + } + + fn states_db(env: Arc) -> Database { + Database::new(env, GENESIS_STATE_TABLE_NAME, Compression::Zstd) + .expect("failed to open states table") + } + + fn h256(byte: u8) -> H256 { + H256::from_slice(&[byte; 32]) + } + + fn sample_block(slot: u64, root_byte: u8) -> Block { + Block { + slot: Slot(slot), + proposer_index: 0, + parent_root: H256::default(), + state_root: h256(root_byte), + body: BlockBody::default(), + } + } + + fn sample_state(genesis_time: u64) -> State { + State::generate_genesis_with_validators(genesis_time, vec![Validator::default(); 2]) + } + + fn block_key(slot: u64, root_byte: u8) -> BlockKey { + BlockKey { + slot: Slot(slot), + root: h256(root_byte), + } + } + + 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 compression_roundtrips() { + let data = vec![42u8; 1024]; + for compression in [Compression::None, Compression::Lz4, Compression::Zstd] { + let compressed = compression.compress(&data).expect("compress should succeed"); + let decompressed = compression + .decompress(&compressed) + .expect("decompress should succeed"); + assert_eq!(decompressed, data); + } + } + + #[test] + fn get_missing_key_returns_none() { + let db = blocks_db(test_env("get_missing")); + assert!( + db.get(&block_key(1, 1)) + .expect("get should succeed") + .is_none() + ); + } + + #[test] + fn put_then_get_roundtrips() { + let db = blocks_db(test_env("put_get")); + let block = sample_block(3, 7); + db.put(&block_key(3, 7), &block).expect("put should succeed"); + let read = db + .get(&block_key(3, 7)) + .expect("get should succeed") + .expect("value should exist"); + assert_ssz_eq(&read, &block); + } + + #[test] + fn put_overwrites_existing_value() { + let db = blocks_db(test_env("overwrite")); + let key = block_key(1, 1); + db.put(&key, &sample_block(1, 1)).expect("put should succeed"); + let updated = sample_block(1, 9); + db.put(&key, &updated).expect("second put should succeed"); + let read = db + .get(&key) + .expect("get should succeed") + .expect("value should exist"); + assert_ssz_eq(&read, &updated); + } + + #[test] + fn range_returns_blocks_in_slot_order() { + let db = blocks_db(test_env("range_order")); + for slot in [300u64, 0, 256, 255, 2] { + db.put(&block_key(slot, slot as u8), &sample_block(slot, slot as u8)) + .expect("put should succeed"); + } + let slots: Vec = db + .range::(..) + .expect("range should succeed") + .iter() + .map(|b| b.slot.0) + .collect(); + assert_eq!(slots, vec![0, 2, 255, 256, 300]); + } + + #[test] + fn range_respects_bounds() { + let db = blocks_db(test_env("range_bounds")); + for slot in 0..6u64 { + db.put(&block_key(slot, slot as u8), &sample_block(slot, slot as u8)) + .expect("put should succeed"); + } + + let slots = |blocks: Vec| blocks.iter().map(|b| b.slot.0).collect::>(); + + let from = db + .range(BySlot(Slot(2))..) + .expect("range from should succeed"); + assert_eq!(slots(from), vec![2, 3, 4, 5]); + + let below = db + .range(..BySlot(Slot(3))) + .expect("range below should succeed"); + assert_eq!(slots(below), vec![0, 1, 2]); + + let inclusive = db + .range(..=BySlot(Slot(3))) + .expect("inclusive range should succeed"); + assert_eq!(slots(inclusive), vec![0, 1, 2, 3]); + } + + #[test] + fn range_returns_all_forks_at_a_slot() { + let db = blocks_db(test_env("range_forks")); + db.put(&block_key(4, 1), &sample_block(4, 1)).expect("put should succeed"); + db.put(&block_key(4, 2), &sample_block(4, 2)).expect("put should succeed"); + db.put(&block_key(5, 3), &sample_block(5, 3)).expect("put should succeed"); + + let at_slot_4 = db + .range(BySlot(Slot(4))..=BySlot(Slot(4))) + .expect("range should succeed"); + assert_eq!(at_slot_4.len(), 2); + assert!(at_slot_4.iter().all(|b| b.slot.0 == 4)); + } + + #[test] + fn delete_range_removes_and_counts() { + let db = blocks_db(test_env("delete_range")); + for slot in 0..6u64 { + db.put(&block_key(slot, slot as u8), &sample_block(slot, slot as u8)) + .expect("put should succeed"); + } + + let deleted = db + .delete_range(..BySlot(Slot(3))) + .expect("delete_range should succeed"); + assert_eq!(deleted, 3); + + assert!( + db.get(&block_key(0, 0)) + .expect("get should succeed") + .is_none() + ); + let remaining = db.range::(..).expect("range should succeed"); + assert_eq!(remaining.len(), 3); + } + + #[test] + fn state_key_prefix_returns_newest_state() { + let db = states_db(test_env("state_newest")); + db.put(&StateKey(Slot(0)), &sample_state(1000)) + .expect("put should succeed"); + db.put(&StateKey(Slot(64)), &sample_state(2000)) + .expect("put should succeed"); + + let newest = db + .range(StateKeyPrefix..) + .expect("range should succeed") + .pop() + .expect("state should exist"); + assert_eq!(newest.config.genesis_time, 2000); + } + + #[test] + fn delete_range_prunes_old_states() { + let db = states_db(test_env("state_prune")); + db.put(&StateKey(Slot(0)), &sample_state(1000)) + .expect("put should succeed"); + db.put(&StateKey(Slot(64)), &sample_state(2000)) + .expect("put should succeed"); + + let deleted = db + .delete_range(..StateKey(Slot(64))) + .expect("delete_range should succeed"); + assert_eq!(deleted, 1); + + assert!( + db.get(&StateKey(Slot(0))) + .expect("get should succeed") + .is_none() + ); + assert!( + db.get(&StateKey(Slot(64))) + .expect("get should succeed") + .is_some() + ); + } + + #[test] + fn environment_builder_reuses_live_environment() { + let path = test_db_path("registry_reuse"); + let first = EnvironmentBuilder::new(path.clone(), 1) + .build() + .expect("first build should succeed"); + let second = EnvironmentBuilder::new(path, 1) + .build() + .expect("second build should succeed"); + assert!(Arc::ptr_eq(&first, &second)); + } + + #[test] + fn values_persist_across_environment_reopen() { + let path = test_db_path("reopen"); + let block = sample_block(7, 7); + { + let db = blocks_db( + EnvironmentBuilder::new(path.clone(), 1) + .build() + .expect("first build should succeed"), + ); + db.put(&block_key(7, 7), &block).expect("put should succeed"); + } + let db = blocks_db( + EnvironmentBuilder::new(path, 1) + .build() + .expect("reopen should succeed"), + ); + let read = db + .get(&block_key(7, 7)) + .expect("get should succeed") + .expect("value should survive reopen"); + assert_ssz_eq(&read, &block); + } +} diff --git a/lean_client/fork_choice/Cargo.toml b/lean_client/fork_choice/Cargo.toml index e80e2f4c..2b550ab1 100644 --- a/lean_client/fork_choice/Cargo.toml +++ b/lean_client/fork_choice/Cargo.toml @@ -6,6 +6,7 @@ edition = { workspace = true } anyhow = { workspace = true } bls = { workspace = true } containers = { workspace = true } +database = { workspace = true } env-config = { workspace = true } indexmap = { workspace = true } metrics = { workspace = true } diff --git a/lean_client/fork_choice/src/handlers.rs b/lean_client/fork_choice/src/handlers.rs index e6fcf7fd..17069c96 100644 --- a/lean_client/fork_choice/src/handlers.rs +++ b/lean_client/fork_choice/src/handlers.rs @@ -6,6 +6,7 @@ use containers::{ AttestationData, Checkpoint, SignatureKey, SignedAggregatedAttestation, SignedAttestation, SignedBlock, State, }; +use database::BlockKey; use metrics::METRICS; use parking_lot::RwLock; use ssz::{H256, SszHash}; @@ -727,6 +728,16 @@ pub fn apply_verified_block( store.blocks.insert(block_root, block.clone()); store.states.insert(block_root, Arc::clone(&new_state)); + if let Err(error) = store.blocks_db.put( + &BlockKey { + slot: block.slot, + root: block_root, + }, + &block, + ) { + warn!("failed to persist block to database: {error}"); + } + METRICS.get().map(|m| { m.grandine_store_blocks_size.set(store.blocks.len() as i64); m.grandine_store_states_size.set(store.states.len() as i64); diff --git a/lean_client/fork_choice/src/store.rs b/lean_client/fork_choice/src/store.rs index 6951b9c2..559ea312 100644 --- a/lean_client/fork_choice/src/store.rs +++ b/lean_client/fork_choice/src/store.rs @@ -1,11 +1,17 @@ use std::collections::{HashMap, HashSet}; use std::sync::Arc; +use crate::block_cache::BlockCache; +use crate::handlers::on_block; use anyhow::{Result, anyhow, ensure}; use containers::{ AggregatedSignatureProof, AttestationData, Block, BlockHeader, Checkpoint, Config, SignatureKey, SignedAggregatedAttestation, SignedAttestation, SignedBlock, Slot, State, }; +use database::{ + BLOCKS_TABLE_NAME, BlockKey, BySlot, Compression, Database, EnvironmentBuilder, + GENESIS_STATE_TABLE_NAME, StateKey, StateKeyPrefix, +}; use indexmap::IndexMap; use metrics::{METRICS, set_gauge_u64}; use ssz::{H256, SszHash}; @@ -26,7 +32,7 @@ pub const GOSSIP_DISPARITY_INTERVALS: u64 = 1; /// Forkchoice store tracking chain state and validator attestations -#[derive(Debug, Clone, Default)] +#[derive(Debug, Clone)] pub struct Store { pub time: Interval, @@ -108,6 +114,10 @@ pub struct Store { pub pending_fetch_roots: HashSet, pub log_inv_rate: usize, + + pub blocks_db: Database, + + pub genesis_db: Database, } const JUSTIFICATION_LOOKBACK_SLOTS: u64 = 3; @@ -224,7 +234,7 @@ pub fn get_forkchoice_store( config: Config, is_aggregator: bool, log_inv_rate: usize, -) -> Store { +) -> Result { // Extract the plain Block from the signed block let block = anchor_block.block.clone(); let block_slot = block.slot; @@ -261,10 +271,106 @@ pub fn get_forkchoice_store( let latest_justified = anchor_checkpoint.clone(); let latest_finalized = anchor_checkpoint; + let env_builder = EnvironmentBuilder::new("./data", 1); + let env = env_builder.build()?; + + let blocks_db: Database = + Database::new(env.clone(), BLOCKS_TABLE_NAME, Compression::Lz4)?; + let genesis_db: Database = + Database::new(env, GENESIS_STATE_TABLE_NAME, Compression::Zstd)?; + + if let Some(anchor_state_db) = genesis_db.range(StateKeyPrefix..)?.pop() { + let anchor_slot = anchor_state_db.latest_block_header.slot; + let anchor_header = BlockHeader { + slot: anchor_state_db.latest_block_header.slot, + proposer_index: anchor_state_db.latest_block_header.proposer_index, + parent_root: anchor_state_db.latest_block_header.parent_root, + state_root: anchor_state_db.hash_tree_root(), + body_root: anchor_state_db.latest_block_header.body_root, + }; + let anchor_root = anchor_header.hash_tree_root(); + if let Some(anchor_block_db) = blocks_db.get(&BlockKey { + slot: anchor_slot, + root: anchor_root, + })? { + let anchor_checkpoint = Checkpoint { + root: anchor_root, + slot: anchor_block_db.slot, + }; + + let mut store = Store { + time: anchor_block_db.slot.0 * INTERVALS_PER_SLOT, + config, + is_aggregator, + head: anchor_root, + safe_target: anchor_root, + latest_justified: anchor_checkpoint.clone(), + latest_finalized: anchor_checkpoint, + justified_ever_updated: anchor_block_db.slot.0 == 0, + finalized_ever_updated: false, + blocks: { + let mut m = HashMap::new(); + m.insert(anchor_root, anchor_block_db); + m + }, + states: { + let mut m = HashMap::new(); + m.insert(anchor_root, Arc::new(anchor_state_db)); + m + }, + latest_known_attestations: HashMap::new(), + latest_new_attestations: HashMap::new(), + gossip_signatures: HashMap::new(), + latest_known_aggregated_payloads: IndexMap::new(), + latest_new_aggregated_payloads: IndexMap::new(), + attestation_data_by_root: HashMap::new(), + pending_attestations: HashMap::new(), + pending_aggregated_attestations: HashMap::new(), + pending_fetch_roots: HashSet::new(), + log_inv_rate, + blocks_db, + genesis_db, + }; + + let mut cache = BlockCache::new(); + for block in store.blocks_db.range(BySlot(anchor_slot)..)? { + let signed_block = SignedBlock { + block, + proof: Default::default(), + }; + if let Err(error) = on_block(&mut store, &mut cache, signed_block, false) { + warn!("failed to replay block from database: {error}"); + } + } + + let prune_cutoff = store + .latest_finalized + .slot + .0 + .saturating_sub(STATE_PRUNE_BUFFER); + if prune_cutoff > 0 { + store.blocks_db.delete_range(..BySlot(Slot(prune_cutoff)))?; + } + + return Ok(store); + } + } + + if block_slot.0 == 0 { + genesis_db.put(&StateKey(block_slot), &anchor_state)?; + } + blocks_db.put( + &BlockKey { + slot: block_slot, + root: block_root, + }, + &block, + )?; + // Store the original anchor_state - do NOT modify it // Modifying checkpoints would change its hash_tree_root(), breaking the // consistency with block.state_root - Store { + Ok(Store { time: block_slot.0 * INTERVALS_PER_SLOT, config, is_aggregator, @@ -294,7 +400,9 @@ pub fn get_forkchoice_store( pending_aggregated_attestations: HashMap::new(), pending_fetch_roots: HashSet::new(), log_inv_rate, - } + blocks_db, + genesis_db, + }) } pub fn get_fork_choice_head( @@ -419,6 +527,7 @@ pub fn update_head(store: &mut Store) { } if let Some(block) = store.blocks.get(&finalized_root) { if block.slot == finalized_slot { + let finalized_advanced = finalized_slot > store.latest_finalized.slot; store.latest_finalized = Checkpoint { root: finalized_root, slot: finalized_slot, @@ -429,6 +538,29 @@ pub fn update_head(store: &mut Store) { m.lean_latest_finalized_slot.set(s); } }); + if finalized_advanced { + if let Some(finalized_state) = store.states.get(&finalized_root) { + if let Err(error) = store + .genesis_db + .put(&StateKey(finalized_slot), finalized_state.as_ref()) + { + warn!("failed to persist finalized state to database: {error}"); + } + if let Err(error) = + store.genesis_db.delete_range(..StateKey(finalized_slot)) + { + warn!("failed to prune old finalized states from database: {error}"); + } + } + let prune_cutoff = finalized_slot.0.saturating_sub(STATE_PRUNE_BUFFER); + if prune_cutoff > 0 { + if let Err(error) = + store.blocks_db.delete_range(..BySlot(Slot(prune_cutoff))) + { + warn!("failed to prune blocks from database: {error}"); + } + } + } } } } diff --git a/lean_client/fork_choice/tests/fork_choice_test_vectors.rs b/lean_client/fork_choice/tests/fork_choice_test_vectors.rs index 514e3e2b..d0c7820c 100644 --- a/lean_client/fork_choice/tests/fork_choice_test_vectors.rs +++ b/lean_client/fork_choice/tests/fork_choice_test_vectors.rs @@ -469,7 +469,8 @@ fn forkchoice(spec_file: &str) { body_root, }; - let mut store = get_forkchoice_store(anchor_state, anchor_block, config, false, 1); + let mut store = get_forkchoice_store(anchor_state, anchor_block, config, false, 1) + .expect("failed to create test store"); let mut cache = BlockCache::new(); let mut block_labels: HashMap = HashMap::new(); diff --git a/lean_client/fork_choice/tests/unit_tests/common.rs b/lean_client/fork_choice/tests/unit_tests/common.rs index c9c71da2..8ff3da3a 100644 --- a/lean_client/fork_choice/tests/unit_tests/common.rs +++ b/lean_client/fork_choice/tests/unit_tests/common.rs @@ -22,5 +22,5 @@ pub fn create_test_store() -> Store { proof: Default::default(), }; - get_forkchoice_store(state, signed_block, config, true, 1) + get_forkchoice_store(state, signed_block, config, true, 1).expect("failed to create test store") } diff --git a/lean_client/fork_choice/tests/unit_tests/validator.rs b/lean_client/fork_choice/tests/unit_tests/validator.rs index 97e5f37d..3dec435a 100644 --- a/lean_client/fork_choice/tests/unit_tests/validator.rs +++ b/lean_client/fork_choice/tests/unit_tests/validator.rs @@ -9,6 +9,9 @@ use containers::{ AggregatedSignatureProof, AggregationBits, Attestation, AttestationData, Block, BlockBody, Checkpoint, Config, MultiMessageAggregate, SignedBlock, Slot, State, Validator, }; +use database::{ + BLOCKS_TABLE_NAME, Compression, Database, EnvironmentBuilder, GENESIS_STATE_TABLE_NAME, +}; use fork_choice::block_cache::BlockCache; use fork_choice::handlers::on_block; use fork_choice::store::{Store, get_forkchoice_store, produce_block_with_signatures, update_head}; @@ -122,7 +125,8 @@ fn create_test_store_with_signers() -> (Store, HashMap) { }; ( - get_forkchoice_store(state, signed_block, config, true, 1), + get_forkchoice_store(state, signed_block, config, true, 1) + .expect("failed to create test store"), keys, ) } @@ -514,7 +518,8 @@ fn test_validator_operations_empty_store() { proof: Default::default(), }; - let mut store = get_forkchoice_store(state, signed_block, config, true, 1); + let mut store = get_forkchoice_store(state, signed_block, config, true, 1) + .expect("failed to create test store"); // Should be able to produce block and attestation let (_root, block, _sig) = @@ -554,16 +559,38 @@ fn test_produce_block_missing_parent_state() { }; // Create store with missing parent state + let env = EnvironmentBuilder::new(std::env::temp_dir().join("lean_test_missing_parent_db"), 1) + .build() + .expect("failed to open test database environment"); + let blocks_db = Database::new(env.clone(), BLOCKS_TABLE_NAME, Compression::Lz4) + .expect("failed to open blocks table"); + let genesis_db = Database::new(env, GENESIS_STATE_TABLE_NAME, Compression::Zstd) + .expect("failed to open genesis table"); + let store = Store { time: 100, config: Config { genesis_time: 1000 }, + is_aggregator: true, head: H256::from_slice(&[0xab; 32]), safe_target: H256::from_slice(&[0xab; 32]), latest_justified: checkpoint.clone(), latest_finalized: checkpoint, + justified_ever_updated: false, + finalized_ever_updated: false, blocks: Default::default(), states: Default::default(), - ..Default::default() + latest_known_attestations: Default::default(), + latest_new_attestations: Default::default(), + gossip_signatures: Default::default(), + latest_known_aggregated_payloads: Default::default(), + latest_new_aggregated_payloads: Default::default(), + attestation_data_by_root: Default::default(), + pending_attestations: Default::default(), + pending_aggregated_attestations: Default::default(), + pending_fetch_roots: Default::default(), + log_inv_rate: 1, + blocks_db, + genesis_db, }; let mut s = store; diff --git a/lean_client/http_api/Cargo.toml b/lean_client/http_api/Cargo.toml index 42627d89..97817181 100644 --- a/lean_client/http_api/Cargo.toml +++ b/lean_client/http_api/Cargo.toml @@ -24,6 +24,7 @@ tracing = { workspace = true } xmss = { workspace = true } [dev-dependencies] +database = { workspace = true } fork_choice = { workspace = true } http-body-util = { workspace = true } parking_lot = { workspace = true } diff --git a/lean_client/http_api/src/test_driver.rs b/lean_client/http_api/src/test_driver.rs index afa6a4ab..baeb71a9 100644 --- a/lean_client/http_api/src/test_driver.rs +++ b/lean_client/http_api/src/test_driver.rs @@ -179,7 +179,11 @@ async fn init_fork_choice( genesis_time: anchor_state.config.genesis_time, }; - let new_store = get_forkchoice_store(anchor_state, anchor_block, config, false, 1); + let std::result::Result::Ok(new_store) = + get_forkchoice_store(anchor_state, anchor_block, config, false, 1) + else { + return StatusCode::INTERNAL_SERVER_ERROR; + }; *state.store.write() = new_store; *state.cache.write() = BlockCache::new(); diff --git a/lean_client/http_api/tests/api_endpoint.rs b/lean_client/http_api/tests/api_endpoint.rs index ebc07dde..09865ad9 100644 --- a/lean_client/http_api/tests/api_endpoint.rs +++ b/lean_client/http_api/tests/api_endpoint.rs @@ -12,6 +12,9 @@ use axum::{ body::Body, http::{Request, header::CONTENT_TYPE}, }; +use database::{ + BLOCKS_TABLE_NAME, Compression, Database, EnvironmentBuilder, GENESIS_STATE_TABLE_NAME, +}; use fork_choice::store::Store; use http_api::{AggregatorController, HttpServerConfig, SharedStore, normal_routes}; use http_body_util::BodyExt; @@ -68,9 +71,43 @@ fn api_endpoint(spec_file: &str) { // Build a minimal store — aggregator handlers only read/write // `store.is_aggregator`; no chain state is needed. let initial_is_aggregator = case.initial_is_aggregator.unwrap_or(false); + let env = EnvironmentBuilder::new( + std::env::temp_dir().join(format!( + "lean_api_test_db_{}", + spec_file.replace(['/', '\\'], "_") + )), + 1, + ) + .build() + .expect("failed to open test database environment"); + let blocks_db = Database::new(env.clone(), BLOCKS_TABLE_NAME, Compression::Lz4) + .expect("failed to open blocks table"); + let genesis_db = Database::new(env, GENESIS_STATE_TABLE_NAME, Compression::Zstd) + .expect("failed to open genesis table"); let store: SharedStore = Arc::new(RwLock::new(Store { + time: Default::default(), + config: Default::default(), is_aggregator: initial_is_aggregator, - ..Default::default() + head: Default::default(), + safe_target: Default::default(), + latest_justified: Default::default(), + latest_finalized: Default::default(), + justified_ever_updated: false, + finalized_ever_updated: false, + blocks: Default::default(), + states: Default::default(), + latest_known_attestations: Default::default(), + latest_new_attestations: Default::default(), + gossip_signatures: Default::default(), + latest_known_aggregated_payloads: Default::default(), + latest_new_aggregated_payloads: Default::default(), + attestation_data_by_root: Default::default(), + pending_attestations: Default::default(), + pending_aggregated_attestations: Default::default(), + pending_fetch_roots: Default::default(), + log_inv_rate: Default::default(), + blocks_db, + genesis_db, })); let controller = Some(Arc::new(AggregatorController::new(store.clone(), None))); diff --git a/lean_client/src/main.rs b/lean_client/src/main.rs index de675c45..e25448ce 100644 --- a/lean_client/src/main.rs +++ b/lean_client/src/main.rs @@ -1069,7 +1069,7 @@ async fn main() -> Result<()> { config.clone(), args.is_aggregator, genesis_log_inv_rate as usize, - ))); + )?)); // Seed the block provider so we can serve the anchor block to peers via BlocksByRoot. {