diff --git a/Cargo.lock b/Cargo.lock index 78e83e1b..12638f65 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -682,12 +682,24 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "1fd0f2584146f6f2ef48085050886acf353beff7305ebd1ae69500e27c67f64b" +[[package]] +name = "byteorder-lite" +version = "0.1.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f1fe948ff07f4bd06c30984e69f5b4899c516a3ef74f34df92a2df2ab535495" + [[package]] name = "bytes" version = "1.12.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" +[[package]] +name = "byteview" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d74937895e761d5984206f82ca8ff7725d0fd8021921011921894b41ab1b9f7" + [[package]] name = "cast" version = "0.3.0" @@ -806,6 +818,12 @@ dependencies = [ "unreachable", ] +[[package]] +name = "compare" +version = "0.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ea0095f6103c2a8b44acd6fd15960c801dafebf02e21940360833e0673f48ba7" + [[package]] name = "const-oid" version = "0.9.6" @@ -899,6 +917,16 @@ dependencies = [ "crossbeam-utils", ] +[[package]] +name = "crossbeam-skiplist" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "df29de440c58ca2cc6e587ec3d22347551a32435fbde9d2bff64e78a9ffa151b" +dependencies = [ + "crossbeam-epoch", + "crossbeam-utils", +] + [[package]] name = "crossbeam-utils" version = "0.8.22" @@ -1037,6 +1065,20 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "dashmap" +version = "6.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e6361d5c062261c78a176addb82d4c821ae42bed6089de0e12603cd25de2059c" +dependencies = [ + "cfg-if", + "crossbeam-utils", + "hashbrown 0.14.5", + "lock_api", + "once_cell", + "parking_lot_core", +] + [[package]] name = "defmt" version = "1.1.1" @@ -1307,6 +1349,18 @@ dependencies = [ "syn 3.0.3", ] +[[package]] +name = "enum_dispatch" +version = "0.3.13" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aa18ce2bc66555b3218614519ac839ddb759a7d6720732f979ef8d13be147ecd" +dependencies = [ + "once_cell", + "proc-macro2", + "quote", + "syn 2.0.119", +] + [[package]] name = "env_filter" version = "2.0.0" @@ -1418,6 +1472,22 @@ version = "1.0.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "059c31d7d36c43fe39d89e55711858b4da8be7eb6dabac23c7289b1a19489406" +[[package]] +name = "fjall" +version = "3.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d5427b70d8592043024955a4a40b5cf44af323fdad7c1f62848a01ec7806709e" +dependencies = [ + "byteorder-lite", + "byteview", + "dashmap", + "flume", + "log", + "lsm-tree", + "tempfile", + "xxhash-rust", +] + [[package]] name = "flume" version = "0.12.0" @@ -1681,6 +1751,12 @@ dependencies = [ "ahash", ] +[[package]] +name = "hashbrown" +version = "0.14.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" + [[package]] name = "hashbrown" version = "0.15.5" @@ -1690,6 +1766,12 @@ dependencies = [ "allocator-api2", ] +[[package]] +name = "hashbrown" +version = "0.16.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100" + [[package]] name = "hashbrown" version = "0.17.1" @@ -1916,6 +1998,15 @@ dependencies = [ "hashbrown 0.17.1", ] +[[package]] +name = "interval-heap" +version = "0.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "11274e5e8e89b8607cfedc2910b6626e998779b48a019151c7604d0adcb86ac6" +dependencies = [ + "compare", +] + [[package]] name = "is_terminal_polyfill" version = "1.70.2" @@ -2148,6 +2239,27 @@ version = "0.4.33" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0ceec5bc11778974d1bcb055b18002eba7f4b3518b6a0081b3af5f21666da9ad" +[[package]] +name = "lsm-tree" +version = "3.1.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "03c4da84fa3a8638fb26d4432bfeb69a1b560905ee2980134d8993e2cfa0dbf3" +dependencies = [ + "byteorder-lite", + "byteview", + "crossbeam-skiplist", + "enum_dispatch", + "interval-heap", + "log", + "quick_cache", + "rustc-hash", + "self_cell", + "sfa", + "tempfile", + "varint-rs", + "xxhash-rust", +] + [[package]] name = "magic-root-interface" version = "0.1.2" @@ -2244,7 +2356,6 @@ dependencies = [ "agave-transaction-view", "derive_more", "futures", - "heed", "oneshot", "prometheus", "serde", @@ -2314,10 +2425,9 @@ version = "0.1.2" dependencies = [ "agave-transaction-view", "bitcode", - "bytemuck", "derive_more", + "fjall", "flume", - "heed", "magicblock-engine-nucleus", "magicblock-ledger", "memmap2", @@ -2802,6 +2912,16 @@ dependencies = [ "syn 2.0.119", ] +[[package]] +name = "quick_cache" +version = "0.6.24" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b9c6658afe513a3b484e3abfdaa0d03ef3c0bbf017542c178dd55f94eb3051f9" +dependencies = [ + "equivalent", + "hashbrown 0.16.1", +] + [[package]] name = "quote" version = "1.0.47" @@ -3033,6 +3153,12 @@ version = "0.1.28" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b74b56ffa8bb2830709a538c2cbcae9aa062db0d2a42563bfb09bdaae44020eb" +[[package]] +name = "rustc-hash" +version = "2.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6b1e7f9a428571be2dc5bc0505c13fb6bf936822b894ec87abf8a08a4e51742d" + [[package]] name = "rustc_version" version = "0.4.1" @@ -3122,6 +3248,12 @@ dependencies = [ "zeroize", ] +[[package]] +name = "self_cell" +version = "1.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ab42ca02749e120097e328d91d415325bdf43b1c72c4c8badf37375fe40a813" + [[package]] name = "semver" version = "1.0.28" @@ -3199,6 +3331,17 @@ dependencies = [ "serde_core", ] +[[package]] +name = "sfa" +version = "1.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a1296838937cab56cd6c4eeeb8718ec777383700c33f060e2869867bd01d1175" +dependencies = [ + "byteorder-lite", + "log", + "xxhash-rust", +] + [[package]] name = "sha2" version = "0.9.9" @@ -4986,6 +5129,12 @@ version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65" +[[package]] +name = "varint-rs" +version = "2.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "bfa6c38708f6257f1ec2ca7e5a11f9bbf58a27d7060078b6b333624968183d96" + [[package]] name = "vcpkg" version = "0.2.15" @@ -5305,6 +5454,12 @@ dependencies = [ "rustix", ] +[[package]] +name = "xxhash-rust" +version = "0.8.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6" + [[package]] name = "yoke" version = "0.8.3" diff --git a/Cargo.toml b/Cargo.toml index a47ed259..5be51f7c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -57,6 +57,7 @@ clonetree = "0.0.2" criterion = "0.8.2" derive_more = "2.1.1" env_logger = "0.11.8" +fjall = { version = "3.1.9", default-features = false } flume = { version = "0.12" } futures = { version = "0.3.32", default-features = false } heed = { version = "0.22.1", default-features = false } diff --git a/accountsdb/Cargo.toml b/accountsdb/Cargo.toml index d6cc7aeb..422dd07d 100644 --- a/accountsdb/Cargo.toml +++ b/accountsdb/Cargo.toml @@ -16,7 +16,7 @@ name = "accountsdb" testkit = [] [dependencies] -nucleus = { workspace = true, features = ["heed", "metrics"] } +nucleus = { workspace = true, features = ["metrics"] } ahash = { workspace = true } bincode = { workspace = true } diff --git a/accountsdb/src/lib.rs b/accountsdb/src/lib.rs index d1dc7ff0..61fca515 100644 --- a/accountsdb/src/lib.rs +++ b/accountsdb/src/lib.rs @@ -9,13 +9,12 @@ use std::{ use derive_more::From; use nucleus::Slot; -use nucleus::heed::RoTxnTls; use solana_account::{AccountSeqLock, AccountSharedData, CoWAccount}; use solana_pubkey::Pubkey; use tracing::{info, warn}; use crate::{ - store::{DatabaseVersion, PersistedProgramIter, PersistedStore}, + store::{DatabaseVersion, PersistedProgramIter, PersistedStore, index::RoTxnTls}, volatile::VolatileStore, }; diff --git a/accountsdb/src/store/index.rs b/accountsdb/src/store/index.rs index 7b3c95c2..100343d3 100644 --- a/accountsdb/src/store/index.rs +++ b/accountsdb/src/store/index.rs @@ -7,9 +7,8 @@ use std::{fs, mem, path::Path}; use heed::{ Database, DatabaseFlags, Env, EnvFlags, EnvOpenOptions, IntegerComparator, Result, RoIter, - RoTxn, RwTxn, iteration_method::MoveOnCurrentKeyDuplicates, + RoTxn, RwTxn, WithTls, iteration_method::MoveOnCurrentKeyDuplicates, }; -use nucleus::heed::{DatabaseIndex, RoTxnTls}; use solana_pubkey::Pubkey; use crate::store::kv::{KeyTail, Offset, OwnerAndOffset, PubkeyBytes, U32LE}; @@ -32,6 +31,29 @@ const FREELIST_INDEX: &str = "freelist"; type RoAccountIter<'a> = RoIter<'a, PubkeyBytes, OwnerAndOffset>; /// Duplicate iterator over program-owned persisted accounts. type RoProgramIter<'a> = RoIter<'a, KeyTail, Offset, MoveOnCurrentKeyDuplicates>; +/// Read-only transaction using heed thread-local storage. +pub(crate) type RoTxnTls<'e> = RoTxn<'e, WithTls>; +/// Optional write transaction used by batched updates. +pub(crate) type OptRwTxn<'t, 'e> = &'t mut Option>; +/// Optional read transaction used by batched reads. +pub(crate) type OptRoTxn<'t, 'e> = &'t mut Option>; + +/// Uses the supplied write transaction or opens one against `env` on demand. +pub(crate) fn write_txn<'t, 'e>(env: &'e Env, txn: OptRwTxn<'t, 'e>) -> Result<&'t mut RwTxn<'e>> { + if let Some(txn) = txn { + return Ok(txn); + } + Ok(txn.insert(env.write_txn()?)) +} + +/// Uses the supplied read transaction or opens one against `env` on demand. +pub(crate) fn read_txn<'t, 'e>(env: &'e Env, txn: OptRoTxn<'t, 'e>) -> Result<&'t RoTxnTls<'e>> { + if let Some(txn) = txn { + return Ok(txn); + } + Ok(txn.insert(env.read_txn()?)) +} + /// Iterator over persisted accounts. pub(crate) struct AccountIter<'a> { /// Iterator over `pubkey -> account` entries. @@ -100,6 +122,16 @@ impl Index { }) } + /// Returns the owning heed environment. + pub(crate) fn env(&self) -> &Env { + &self.env + } + + /// Flushes the index databases to durable storage. + pub(crate) fn flush(&self) -> Result<()> { + self.env.force_sync() + } + /// Returns the persisted offset for `pubkey`. pub(crate) fn offset(&self, key: &Pubkey, txn: &RoTxn<'_>) -> Result> { let entry = self.accounts.get(txn, key)?; @@ -199,9 +231,3 @@ impl Index { self.programs.put(txn, &owner, &offset) } } - -impl DatabaseIndex for Index { - fn env(&self) -> &Env { - &self.env - } -} diff --git a/accountsdb/src/store/mod.rs b/accountsdb/src/store/mod.rs index 9214362c..22bb0fdf 100644 --- a/accountsdb/src/store/mod.rs +++ b/accountsdb/src/store/mod.rs @@ -11,14 +11,13 @@ use solana_account::{ use solana_pubkey::Pubkey; use tracing::{error, warn}; -use nucleus::heed::{DatabaseIndex, OptRoTxn, OptRwTxn, read_txn, write_txn}; use twox_hash::XxHash3_64; use crate::{ AccountEntry, AccountsDBError, Result, StoreKind, metrics::{self, Operation}, store::{ - index::{Index, OwnerIter}, + index::{Index, OptRoTxn, OptRwTxn, OwnerIter, read_txn, write_txn}, kv::{Offset, OwnerAndOffset}, mmap::{DatabaseMeta, MappedStorage}, }, diff --git a/accountsdb/src/tests.rs b/accountsdb/src/tests.rs index ac2ff604..5fc42165 100644 --- a/accountsdb/src/tests.rs +++ b/accountsdb/src/tests.rs @@ -8,17 +8,14 @@ use std::sync::atomic::Ordering::{Relaxed, Release}; use assert_matches::assert_matches; -use nucleus::{ - heed::{DatabaseIndex, read_txn}, - testkit::{TempDir, init_tracing, tempdir}, -}; +use nucleus::testkit::{TempDir, init_tracing, tempdir}; use solana_account::{ AccountBuilder, AccountMode, AccountSharedData, ReadableAccount, WritableAccount, }; use solana_pubkey::Pubkey; use super::*; -use crate::{snapshot::VOLATILE_DB_FILE, store::MIN_REMAINDER}; +use crate::{snapshot::VOLATILE_DB_FILE, store::MIN_REMAINDER, store::index::read_txn}; /// Fresh database on a throwaway directory; the `TempDir` must outlive the db. fn db() -> (TempDir, AccountsDB) { diff --git a/ledger/Cargo.toml b/ledger/Cargo.toml index f1077cdd..b763d971 100644 --- a/ledger/Cargo.toml +++ b/ledger/Cargo.toml @@ -16,20 +16,19 @@ name = "ledger" testkit = [] [dependencies] -nucleus = { workspace = true, features = ["heed", "ledger", "service"] } +nucleus = { workspace = true, features = ["ledger", "service"] } bitcode = { workspace = true } -bytemuck = { workspace = true, features = ["derive", "extern_crate_std"] } derive_more = { workspace = true, features = ["deref", "from"] } +fjall = { workspace = true } flume = { workspace = true } -heed = { workspace = true } memmap2 = { workspace = true } num_cpus = { workspace = true } oneshot = { workspace = true, features = ["std"] } parking_lot = { workspace = true } rustix = { workspace = true, features = ["fs"] } thiserror = { workspace = true } -tokio = { workspace = true, features = ["sync"] } +tokio = { workspace = true, features = ["rt", "sync", "time"] } tracing = { workspace = true } wincode = { workspace = true } zstd = { workspace = true } diff --git a/ledger/README.md b/ledger/README.md index a1ac0c60..898695a4 100644 --- a/ledger/README.md +++ b/ledger/README.md @@ -32,8 +32,25 @@ snapshot's checksum and cumulative transaction count so it remains self-describing after retention removes the preceding blockstore. Reader requests run on a worker pool. Each worker owns its decode buffers and -reads only through published cursors. The optional `testkit` feature reduces -LMDB map sizes and uses one reader worker without changing the on-disk format. +reads only through published cursors. The active Fjall index uses two background +workers and an 8 MiB cache. Sealed indexes open with one worker on demand; the +two immediately preceding the active head are exempt from idle eviction, while +older indexes close after ten idle minutes. The optional `testkit` feature uses +one reader worker without changing the on-disk format. + +Each block atomically commits its transaction, block, and account index changes. +Ordinary boundaries flush data files and the Fjall journal to the operating +system with `Buffer` durability before publishing cursors. Explicit syncs, +seals, resets, shutdown, and retention boundaries additionally use file +`sync_data` and Fjall `SyncData` before synchronously flushing metadata. Thus a +process crash recovers complete published blocks, while an OS or power failure +may discard the active tail after the last strong boundary. + +The account index stores `pubkey_prefix || execution_span_be` as its key and an +empty value. Fixed-width big-endian slot and account-span key components make +reverse Fjall ranges start at the newest entry. Opaque span values remain +little-endian. LZ4 is disabled because the realistic index fixture reduced +closed-directory size by only 7.37%. During coordinated shutdown, one queue marker per reader closes the pool after earlier requests. A final appender sync flushes every preceding event, reports @@ -48,8 +65,9 @@ through the active head in full. ## Retention At a block boundary, the appender checks used bytes on the ledger filesystem. -When the configured limit is reached, `Ledger::truncate` removes the oldest -sealed superblock; the active head is never removed. +When the configured limit is reached, `Ledger::truncate` durably removes the +oldest sealed superblock from ledger metadata and synchronously purges its +directory; the active head is never removed. The size check assumes the ledger directory is on a dedicated filesystem. Unrelated files on that filesystem contribute to the used-byte total and can diff --git a/ledger/src/appender.rs b/ledger/src/appender.rs index aa64c13c..458be7f3 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -2,20 +2,13 @@ use std::{ collections::HashMap, - path::Path, sync::{Arc, atomic::Ordering::*}, }; use agave_transaction_view::transaction_view::TransactionView; use bitcode::Buffer; use flume::Receiver; -use heed::{Env, RwTxn}; -use nucleus::{ - Slot, - heed::{DatabaseIndex, OptRwTxn, write_txn}, - ledger::BlockstorePosition, - shutdown::{ShutdownHandle, ShutdownReason}, -}; +use nucleus::{Slot, ledger::BlockstorePosition}; use solana_signature::Signature; use tokio::sync::broadcast::Sender; use tracing::{info, warn}; @@ -25,13 +18,13 @@ use zstd::bulk::Compressor; use crate::{ Ledger, Superblock, error::{LedgerError, Result}, - index::{Index, Span, TxSpan}, + index::{IndexWriter, Span, TxSpan}, metrics::{self, Operation}, schema::{ Block, BlockstoreEntry, Event, Execution, ExecutionDetails, MAX_EXECUTION_DETAILS_SIZE, SuperblockSeal, TransactionEntry, blockstore, }, - storage::{AppendFile, MetaMap, SuperblockMeta}, + storage::{AppendFile, Durability}, }; /// Blockstore stream file name inside a superblock. @@ -40,6 +33,8 @@ pub(crate) const BLOCKSTORE_DB: &str = "blockstore.db"; pub(crate) const EXECUTIONS_DB: &str = "executions.db"; /// Superblock metadata file name. pub(crate) const SUPERBLOCK_META: &str = "superblock.meta"; +/// Frequency of ledger size checks in slots. +pub(crate) const SIZE_CHECK_FREQUENCY: u64 = 32; /// Background service that appends ledger events into the active superblock. pub(crate) struct LedgerAppender { @@ -49,35 +44,35 @@ pub(crate) struct LedgerAppender { writer: SuperblockWriter, /// Transactions waiting for their matching execution details. pending: HashMap, - /// Active superblock index. - index: Arc, + /// Active superblock index writer and pending atomic boundary. + index: IndexWriter, /// Event stream from the execution pipeline. rx: Receiver, - /// Transactions written since the last durable sync. + /// Transactions written since the last published boundary. transactions: u64, /// Broadcasts the blockstore write position after each committed block. position: Sender, } impl LedgerAppender { - /// Opens the active superblock and registers it on the ledger handle. - pub(crate) fn new( + /// Opens the active superblock and processes events until the stream closes. + pub(crate) fn run( ledger: Arc, rx: Receiver, position: Sender, - ) -> Result { + ) -> Result<()> { let head = ledger.meta.head(); - let directory = Superblock::init_dir(&ledger.directory, head)?; - let writer = SuperblockWriter::new(&directory)?; metrics::pending_transactions(0); - let index = ledger + let superblock = ledger .superblocks .read() .get(&head) - .map(|s| s.index.clone()) + .cloned() .ok_or(LedgerError::Corruption("active superblock missing"))?; + let writer = SuperblockWriter::new(superblock.clone())?; + let index = superblock.index.get()?.writer(); - Ok(Self { + let mut appender = Self { ledger, writer, index, @@ -85,99 +80,59 @@ impl LedgerAppender { pending: HashMap::new(), transactions: 0, position, - }) - } - - /// Runs until shutdown (when the event stream closes). - pub(crate) fn run(mut self, mut shutdown: ShutdownHandle) { - let reason = loop { - let env = self.index.env().clone(); - match self.run_epoch(&env) { - Ok(Epoch::Rotate) => (), - Ok(Epoch::Shutdown) => break ShutdownReason::Signalled, - Err(err) => break ShutdownReason::Error(Box::new(err)), - } }; - // Release ledger ownership before the manager can reopen it. - drop(self); - shutdown.terminate(reason); - } - /// Processes events against one stable superblock environment. - fn run_epoch<'e>(&mut self, env: &'e Env) -> Result { - let mut txn: Option> = None; - loop { - let Ok(event) = self.rx.recv() else { - return Ok(Epoch::Shutdown); - }; - if let Some(epoch) = self.process(event, env, &mut txn)? { - return Ok(epoch); + while let Ok(event) = appender.rx.recv() { + match event { + Event::Transaction(transaction) => appender.write_transaction(transaction)?, + Event::Execution(execution) => appender.write_execution(execution)?, + Event::Block(block) => appender.write_block(block)?, + Event::Superblock(seal) => appender.seal(seal, false)?, + Event::Bootstrap(seal) => appender.seal(seal, true)?, + Event::Reset(slot) => appender.write_reset(slot)?, + Event::Sync { response, is_final } => { + let _ = response.send(appender.sync(None)); + if is_final { + break; + } + } } } + Ok(()) } /// Rotates to the next superblock directory. fn rotate(&mut self, seal: SuperblockSeal) -> Result<()> { let _timer = metrics::time(Operation::Rotate); let head = seal.id + 1; - let superblock = Superblock::open(&self.ledger.directory, head)?; + let superblock = Superblock::open(&self.ledger.directory, head, true)?; // Seal N opens N+1, which stores N's snapshot archive and seal metadata. superblock.meta.checksum.store(seal.checksum, Release); superblock.meta.transactions.store(seal.transactions, Release); - self.writer = SuperblockWriter::new(&superblock.directory)?; + superblock.meta.flush()?; + let writer = SuperblockWriter::new(superblock.clone())?; + let index = superblock.index.get()?.writer(); + + let mut superblocks = self.ledger.superblocks.write(); + superblocks.insert(head, superblock); self.ledger.meta.head.store(head, Release); self.ledger.meta.superblocks.fetch_add(1, Release); self.ledger.meta.flush()?; - self.index = superblock.index.clone(); - superblock.meta.flush()?; - self.ledger.superblocks.write().insert(head, superblock); + if let Some(sealed) = superblocks.get(&seal.id) { + sealed.index.seal(); + } + drop(superblocks); + + self.writer = writer; + self.index = index; info!(head, "opened active superblock"); Ok(()) } - /// Processes one append event, rotating after a superblock seal. - /// - fn process<'e>( - &mut self, - event: Event, - env: &'e Env, - txn: OptRwTxn<'_, 'e>, - ) -> Result> { - match event { - Event::Transaction(transaction) => { - self.write_transaction(transaction)?; - } - Event::Execution(execution) => { - self.write_execution(execution, env, txn)?; - } - Event::Block(block) => { - self.write_block(block, env, txn)?; - } - Event::Superblock(seal) => { - self.seal(seal, false, txn)?; - return Ok(Some(Epoch::Rotate)); - } - Event::Bootstrap(seal) => { - self.seal(seal, true, txn)?; - return Ok(Some(Epoch::Rotate)); - } - Event::Reset(slot) => { - self.write_reset(slot, txn)?; - } - Event::Sync { response, is_final } => { - let _ = response.send(self.sync(None, txn)); - if is_final { - return Ok(Some(Epoch::Shutdown)); - } - } - } - Ok(None) - } - /// Seals the active superblock, optionally adopting a restored snapshot's /// cumulative transaction count before publishing the successor metadata. - fn seal(&mut self, seal: SuperblockSeal, bootstrap: bool, txn: OptRwTxn<'_, '_>) -> Result<()> { - self.write_superblock(seal, txn)?; + fn seal(&mut self, seal: SuperblockSeal, bootstrap: bool) -> Result<()> { + self.write_superblock(seal)?; if bootstrap { self.ledger.meta.transactions.store(seal.transactions, Release); } @@ -199,12 +154,7 @@ impl LedgerAppender { } /// Writes execution details and adds transaction/account indexes. - fn write_execution<'e>( - &mut self, - execution: Execution, - env: &'e Env, - txn: OptRwTxn<'_, 'e>, - ) -> Result<()> { + fn write_execution(&mut self, execution: Execution) -> Result<()> { let signature: Signature = execution.header.signature; let Some(pending) = self.pending.remove(&signature) else { warn!(%signature, "ledger execution arrived without a pending transaction; skipping"); @@ -216,20 +166,20 @@ impl LedgerAppender { blockstore: pending.span, execution, }; - let txn = write_txn(env, txn)?; - self.index.insert_transaction(txn, &signature, &span)?; + self.index.insert_transaction(&signature, span); let view = TransactionView::try_new_unsanitized(pending.transaction)?; let accounts = view.static_account_keys(); - self.index.insert_accounts(txn, accounts, &span.execution)?; + self.index.insert_accounts(accounts, span.execution); Ok(()) } - /// Writes a block boundary and publishes it after data and indexes are durable. - fn write_block<'e>(&mut self, block: Block, env: &'e Env, txn: OptRwTxn<'_, 'e>) -> Result<()> { + /// Writes a block boundary and publishes it after data and indexes reach the OS. + fn write_block(&mut self, block: Block) -> Result<()> { let span = self.writer.write_blockstore(&BlockstoreEntry::Block(block))?; - self.index.insert_block(write_txn(env, txn)?, &block.slot, &span)?; - self.sync(Some(block.slot), txn)?; - if self.ledger.size_exceeded()? { + self.index.insert_block(block.slot, span); + self.publish(Some(block.slot), Durability::Buffer)?; + if block.slot.is_multiple_of(SIZE_CHECK_FREQUENCY) && self.ledger.size_exceeded()? { + self.sync(None)?; self.ledger.truncate()?; } metrics::ledger_counts(&self.ledger); @@ -237,18 +187,18 @@ impl LedgerAppender { } /// Writes a superblock seal and prepares files for read-only access. - fn write_superblock(&mut self, seal: SuperblockSeal, txn: OptRwTxn<'_, '_>) -> Result<()> { + fn write_superblock(&mut self, seal: SuperblockSeal) -> Result<()> { self.writer.write_blockstore(&BlockstoreEntry::Superblock(seal))?; - self.sync(None, txn)?; + self.sync(None)?; self.writer.finalize()?; info!(superblock = seal.id, "sealed superblock"); Ok(()) } /// Writes and publishes a volatile-state reset marker. - fn write_reset(&mut self, slot: Slot, txn: OptRwTxn<'_, '_>) -> Result<()> { + fn write_reset(&mut self, slot: Slot) -> Result<()> { self.writer.write_blockstore(&BlockstoreEntry::Reset(slot))?; - self.sync(None, txn)?; + self.sync(None)?; info!(slot, "appended volatile state reset"); Ok(()) } @@ -256,35 +206,31 @@ impl LedgerAppender { /// Makes files and indexes durable, publishes their cursors and accumulated /// transaction count, and broadcasts the new blockstore position. When /// `slot` is supplied, the same boundary also publishes block metadata. - fn sync(&mut self, slot: Option, txn: OptRwTxn<'_, '_>) -> Result<()> { - let cursors = self.writer.sync()?; - if let Some(txn) = txn.take() { - txn.commit()?; - } - self.index.flush()?; - self.writer.publish(cursors, slot)?; + fn sync(&mut self, slot: Option) -> Result<()> { + self.publish(slot, Durability::SyncData) + } + + /// Publishes one complete boundary with buffered or data-synced durability. + fn publish(&mut self, slot: Option, durability: Durability) -> Result<()> { + let cursors = self.writer.persist(durability)?; + self.index.persist(durability)?; + self.writer.publish(cursors, slot, durability)?; self.ledger.meta.transactions.fetch_add(self.transactions, Release); if let Some(slot) = slot { self.ledger.meta.blocks.fetch_add(1, Release); self.ledger.meta.range.end.store(slot, Release); } - self.ledger.meta.flush()?; + self.ledger.meta.persist(durability)?; self.transactions = 0; let position = BlockstorePosition { superblock: self.ledger.meta.head(), - offset: cursors.0, + offset: cursors.blockstore, }; let _ = self.position.send(position); Ok(()) } } -/// Boundary reached while processing one superblock environment. -enum Epoch { - Rotate, - Shutdown, -} - /// Transaction bytes already written but not yet paired with execution details. struct PendingTx { /// Transaction bytes retained until execution details arrive for indexing. @@ -293,8 +239,17 @@ struct PendingTx { span: Span, } +/// Published byte cursors for the active superblock data files. +#[derive(Clone, Copy)] +struct Cursors { + blockstore: u64, + executions: u64, +} + /// Writable superblock files and reusable execution-detail encoder. struct SuperblockWriter { + /// Active superblock whose metadata publishes these files. + superblock: Arc, /// Compressor reused for execution metadata payloads. compressor: Compressor<'static>, /// Scratch buffer owned by bitcode while encoding metadata. @@ -303,28 +258,22 @@ struct SuperblockWriter { blockstore: AppendFile, /// Buffered execution details stream. executions: AppendFile, - /// Mmap-backed metadata for this superblock. - metadata: MetaMap, } impl SuperblockWriter { - /// Opens writable files and metadata under `directory`. - fn new(directory: &Path) -> Result { - // SAFETY: `SuperblockMeta` and its nested headers have stable C layouts, - // and all fields that can change while mapped are atomic. The ledger - // exclusively creates and updates this superblock metadata file. - let metadata = unsafe { MetaMap::::new(&directory.join(SUPERBLOCK_META)) }?; - + /// Opens writable files for the active `superblock`. + fn new(superblock: Arc) -> Result { + let directory = &superblock.directory; Ok(Self { blockstore: AppendFile::new( &directory.join(BLOCKSTORE_DB), - &metadata.cursors.blockstore, + &superblock.meta.cursors.blockstore, )?, executions: AppendFile::new( &directory.join(EXECUTIONS_DB), - &metadata.cursors.executions, + &superblock.meta.cursors.executions, )?, - metadata, + superblock, compressor: Compressor::new(0)?, buffer: Buffer::new(), }) @@ -353,25 +302,31 @@ impl SuperblockWriter { Ok(Span::new(offset, size)) } - /// Syncs data files and returns durable cursors. - fn sync(&mut self) -> Result<(u64, u64)> { - let _timer = metrics::time(Operation::FileSync); - Ok((self.blockstore.sync()?, self.executions.sync()?)) + /// Persists data files and returns their published cursors. + fn persist(&mut self, durability: Durability) -> Result { + let _timer = if durability.requires_sync() { + metrics::time(Operation::FileSync) + } else { + metrics::time(Operation::BufferSync) + }; + Ok(Cursors { + blockstore: self.blockstore.persist(durability)?, + executions: self.executions.persist(durability)?, + }) } - /// Publishes durable cursors into superblock metadata. - fn publish(&self, cursors: (u64, u64), slot: Option) -> Result<()> { - let (blockstore, executions) = cursors; - self.metadata.cursors.blockstore.store(blockstore, Release); - self.metadata.cursors.executions.store(executions, Release); + /// Publishes file cursors into superblock metadata at the selected durability. + fn publish(&self, cursors: Cursors, slot: Option, durability: Durability) -> Result<()> { + let metadata = &self.superblock.meta; + metadata.cursors.blockstore.store(cursors.blockstore, Release); + metadata.cursors.executions.store(cursors.executions, Release); if let Some(slot) = slot { - self.metadata.range.end.store(slot, Release); + metadata.range.end.store(slot, Release); // The first block of a segment fixes its start // slot; later blocks only extend the end. - let _ = self.metadata.range.start.compare_exchange(0, slot, Release, Relaxed); + let _ = metadata.range.start.compare_exchange(0, slot, Release, Relaxed); } - self.metadata.flush()?; - Ok(()) + metadata.persist(durability) } /// Trims preallocated file space after the superblock cursors are durable. diff --git a/ledger/src/error.rs b/ledger/src/error.rs index 4c899384..19839b4f 100644 --- a/ledger/src/error.rs +++ b/ledger/src/error.rs @@ -3,7 +3,6 @@ use std::io; use agave_transaction_view::result::TransactionViewError; -use heed::BoxedError; use oneshot::RecvError; use tokio::time::error::Elapsed; @@ -22,12 +21,9 @@ pub enum LedgerError { /// Bitcode failed while serializing execution details. #[error("ledger execution-details codec error: {0}")] Bitcode(#[source] bitcode::Error), - /// LMDB key/value codec failed while encoding or decoding an index value. - #[error("ledger index key/value codec error: {0}")] - IndexCodec(#[source] BoxedError), - /// LMDB failed while opening or accessing a ledger index. + /// Fjall failed while opening or accessing a ledger index. #[error("ledger index error: {0}")] - Index(#[source] heed::Error), + Index(#[source] fjall::Error), /// Sanitized transaction view could not be decoded. #[error("transaction view error: {0:?}")] TransactionView(TransactionViewError), diff --git a/ledger/src/index.rs b/ledger/src/index.rs index 247d9a5d..2ccf47c4 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -1,31 +1,32 @@ -//! LMDB index schema and codecs for ledger blockstore entries. - -use std::{array, borrow::Cow, fs, path::Path}; - -use bytemuck::{Pod, Zeroable}; -use heed::{ - BoxedError, BytesDecode, BytesEncode, Database, DatabaseFlags, Env, EnvFlags, EnvOpenOptions, - IntegerComparator, RoIter, RwTxn, byteorder::LittleEndian, - iteration_method::MoveOnCurrentKeyDuplicates, types::U64, +//! Fjall index schema and codecs for ledger blockstore entries. + +use std::{ + iter::Rev, + ops::RangeInclusive, + path::{Path, PathBuf}, + sync::Arc, + time::Instant, }; -use nucleus::{ - Slot, - heed::{DatabaseIndex, OptRoTxn, read_txn}, + +use fjall::{ + CompressionType, Database, Keyspace, KeyspaceCreateOptions, OwnedWriteBatch, PersistMode, + Readable, Snapshot, + config::{CompressionPolicy, FilterPolicy}, + util::prefixed_range, }; +use nucleus::{MB, Slot}; +use parking_lot::Mutex; use solana_pubkey::Pubkey; use solana_signature::Signature; -use crate::schema::Offset; +use crate::{ + error::{LedgerError, Result}, + schema::Offset, + storage::Durability, +}; /// Index directory below each superblock directory. const INDEX_SUBDIR: &str = "index"; -/// Maximum LMDB map size for ledger indexes. -#[cfg(feature = "testkit")] -const INDEX_MAP_SIZE: usize = 32 * nucleus::MB; -#[cfg(not(feature = "testkit"))] -const INDEX_MAP_SIZE: usize = 16 * nucleus::GB; -/// Number of LMDB named databases in the ledger index. -const INDEX_DBS: u32 = 3; /// Transaction signature index name. const TRANSACTIONS_INDEX: &str = "transactions"; /// Slot-to-offset index name. @@ -33,38 +34,32 @@ const SLOTS_INDEX: &str = "slots"; /// Account-to-offset index name. const ACCOUNTS_INDEX: &str = "accounts"; /// Bytes kept from wide keys in compact index keys. -const KEY_BYTES: usize = 16; - -/// Result type returned by heed codec hooks. -type CodecResult = Result; -/// Little-endian u64 codec used by LMDB keys and values. -type U64Le = U64; -/// Little-endian slot key with integer ordering. -type SlotKey = U64; -/// Truncated transaction signature key. -/// -/// The 16-byte prefix is used as an index tag, not as a collision-proof -/// identity. Collision probability is negligible for the ledger's expected -/// scale, so the index accepts that risk to keep keys compact. -#[derive(Pod, Zeroable, Clone, Copy)] -#[repr(C)] -pub(crate) struct SignatureKey([u8; KEY_BYTES]); - -/// Truncated account pubkey key. +const PREFIX_BYTES: usize = 16; +/// Bytes occupied by one encoded span. +const SPAN_BYTES: usize = size_of::(); +/// Bytes occupied by a transaction's two spans. +const TX_SPAN_BYTES: usize = 2 * SPAN_BYTES; +/// Bytes occupied by an account prefix and ordered execution span. +const ACCOUNT_KEY_BYTES: usize = PREFIX_BYTES + SPAN_BYTES; +/// Block cache capacity for one opened superblock index. +const CACHE_SIZE: u64 = 8 * MB as u64; +/// Fjall maintenance workers assigned to the active writable index. +const ACTIVE_WORKERS: usize = 2; +/// Fjall maintenance workers assigned to an on-demand sealed index. +const SEALED_WORKERS: usize = 1; + +/// Truncated signature or account key. /// -/// See `SignatureKey`; account history uses the same compact prefix tag. -#[derive(Pod, Zeroable, Clone, Copy)] -#[repr(C)] -pub(crate) struct AccountKey([u8; KEY_BYTES]); +/// The 16-byte prefix is an index tag, not a collision-proof identity. The +/// index accepts the negligible collision risk to keep keys compact. +type Prefix = [u8; PREFIX_BYTES]; /// Packed span inside a ledger data file. /// /// The high 39 bits store the byte offset. The low 25 bits store the encoded -/// entry size. Transaction entries are bounded by Solana's transaction-size -/// limit, and execution details are expected to stay at a few dozen KiB before -/// compression, so spans stay well below the 32 MiB encoded-size cap. -#[derive(Zeroable, Pod, Clone, Copy, Default, Debug, PartialEq, Eq, PartialOrd, Ord)] -#[repr(C)] +/// entry size. Values use little-endian encoding; spans embedded in ordered +/// keys use big-endian so Fjall's byte ordering matches numeric ordering. +#[derive(Clone, Copy, Default, Debug, PartialEq, Eq, PartialOrd, Ord)] pub(crate) struct Span(u64); impl Span { @@ -91,11 +86,32 @@ impl Span { pub(crate) fn size(&self) -> u64 { self.0 & Self::SIZE_MASK } + + /// Encodes an opaque index value using the ledger's little-endian format. + fn value_bytes(self) -> [u8; SPAN_BYTES] { + self.0.to_le_bytes() + } + + /// Encodes a numeric key component preserving its natural ordering. + fn key_bytes(self) -> [u8; SPAN_BYTES] { + self.0.to_be_bytes() + } + + /// Decodes a little-endian span stored as an opaque value. + fn from_value(bytes: &[u8]) -> Result { + let value = fixed(bytes, "invalid span value")?; + Ok(Self(u64::from_le_bytes(value))) + } + + /// Decodes a big-endian span embedded in an ordered key. + fn from_key(bytes: &[u8]) -> Result { + let key = fixed(bytes, "invalid span key")?; + Ok(Self(u64::from_be_bytes(key))) + } } /// Pair of blockstore-file and execution-file spans for a transaction. -#[derive(Zeroable, Pod, Clone, Copy)] -#[repr(C)] +#[derive(Clone, Copy)] pub(crate) struct TxSpan { /// Span of the raw transaction entry in the blockstore. pub(crate) blockstore: Span, @@ -103,219 +119,294 @@ pub(crate) struct TxSpan { pub(crate) execution: Span, } -/// Account-index span codec. -pub(crate) enum AccountSpan {} +/// Reverse account-index iterator over execution spans. +pub(crate) struct AccountIter { + inner: Rev, +} + +/// Reverse block-index iterator over `(slot, span)` pairs. +pub(crate) struct BlockIter { + inner: Rev, +} + +impl Iterator for BlockIter { + type Item = Result<(Slot, Span)>; + + fn next(&mut self) -> Option { + self.inner.next().map(|entry| { + let (key, value) = entry.into_inner()?; + let key = fixed(&key, "invalid slot key")?; + Ok((u64::from_be_bytes(key), Span::from_value(&value)?)) + }) + } +} + +impl Iterator for AccountIter { + type Item = Result; -/// Duplicate account-index iterator over execution spans. -pub(crate) type AccountIter<'a> = RoIter<'a, AccountKey, AccountSpan, MoveOnCurrentKeyDuplicates>; + fn next(&mut self) -> Option { + self.inner.next().map(|entry| { + let key = entry.key()?; + key.get(PREFIX_BYTES..) + .ok_or(LedgerError::Corruption("invalid account index key")) + .and_then(Span::from_key) + }) + } +} -/// LMDB databases used to locate ledger data. +/// Databases used to locate ledger data inside one superblock. pub(crate) struct Index { - /// Owning LMDB environment. - env: Env, - /// Transaction signature to transaction/execution spans. - transactions: Database, - /// Slot to blockstore-file span. - blocks: Database, - /// Account key to execution-details span. - accounts: Database, + /// Signature prefix to transaction and execution spans. + transactions: Keyspace, + /// Slot to blockstore span. + blocks: Keyspace, + /// Account prefix and execution span keys. + accounts: Keyspace, + /// Owning Fjall database. + db: Database, +} + +/// Atomic index mutations accumulated for one published ledger boundary. +pub(crate) struct IndexWriter { + /// Index receiving the pending mutations. + index: Arc, + /// Mutations pending publication at the next boundary. + batch: OwnedWriteBatch, +} + +/// Consistent read snapshot that keeps its underlying index lease alive. +pub(crate) struct IndexReader { + /// Index lease retained for the snapshot lifetime. + index: Arc, + /// Consistent view across every index keyspace. + snapshot: Snapshot, +} + +/// Lazy per-superblock index slot with lease-safe eviction. +pub(crate) struct IndexSlot { + /// Superblock directory containing the index. + directory: PathBuf, + /// Synchronized active and cached state. + state: Mutex, +} + +/// Mutable state of a lazy index slot. +struct IndexState { + /// Whether this index belongs to the writable head. + active: bool, + /// Open index, absent until a sealed index is first read. + cached: Option, +} + +/// Open index and its most recent lease time. +struct CachedIndex { + /// Shared index lease. + index: Arc, + /// Most recent call to [`IndexSlot::get`]. + used: Instant, } impl Index { - /// Opens or creates the index directory and databases. - pub(crate) fn new(path: &Path) -> heed::Result { - let path = path.join(INDEX_SUBDIR); - fs::create_dir_all(&path)?; - // SAFETY: this process owns the index directory for the lifetime of - // the database, so the backing files are not mutated behind LMDB's back. - let env = unsafe { - EnvOpenOptions::new() - .max_dbs(INDEX_DBS) - .map_size(INDEX_MAP_SIZE) - .flags(EnvFlags::WRITE_MAP) - .flags(EnvFlags::NO_SYNC) - .open(path)? + /// Opens or creates an index using `workers` Fjall maintenance threads. + pub(crate) fn new(path: &Path, workers: usize) -> Result { + let db = Database::builder(path.join(INDEX_SUBDIR)) + .cache_size(CACHE_SIZE) + .worker_threads(workers) + .manual_journal_persist(true) + .journal_compression(CompressionType::None) + .open()?; + let options = || { + KeyspaceCreateOptions::default() + .data_block_compression_policy(CompressionPolicy::disabled()) + .index_block_compression_policy(CompressionPolicy::disabled()) + .filter_policy(FilterPolicy::disabled()) }; - - let mut txn = env.write_txn()?; - let transactions = - env.database_options().name(TRANSACTIONS_INDEX).types().create(&mut txn)?; - let blocks = env - .database_options() - .name(SLOTS_INDEX) - .key_comparator() - .types() - .create(&mut txn)?; - let accounts = env - .database_options() - .name(ACCOUNTS_INDEX) - .flags(DatabaseFlags::DUP_SORT | DatabaseFlags::DUP_FIXED | DatabaseFlags::REVERSE_DUP) - .types() - .create(&mut txn)?; - txn.commit()?; + let transactions = db.keyspace(TRANSACTIONS_INDEX, options)?; + let blocks = db.keyspace(SLOTS_INDEX, options)?; + let accounts = db.keyspace(ACCOUNTS_INDEX, options)?; Ok(Self { - env, + db, transactions, blocks, accounts, }) } - /// Indexes a block boundary by slot. - pub(crate) fn insert_block( - &self, - txn: &mut RwTxn<'_>, - slot: &Slot, - span: &Span, - ) -> heed::Result<()> { - self.blocks.put(txn, slot, span) + /// Creates the single writer for this superblock index. + pub(crate) fn writer(self: Arc) -> IndexWriter { + let batch = self.db.batch(); + IndexWriter { index: self, batch } } - /// Indexes a transaction and its execution details. - pub(crate) fn insert_transaction( - &self, - txn: &mut RwTxn<'_>, - signature: &Signature, - span: &TxSpan, - ) -> heed::Result<()> { - self.transactions.put(txn, signature, span) + /// Opens a consistent read view across all logical keyspaces. + pub(crate) fn reader(self: Arc) -> IndexReader { + let snapshot = self.db.snapshot(); + IndexReader { index: self, snapshot } } +} - /// Adds account-to-execution entries for all static transaction accounts. - pub(crate) fn insert_accounts( - &self, - txn: &mut RwTxn<'_>, - accounts: &[Pubkey], - span: &Span, - ) -> heed::Result<()> { - for account in accounts { - self.accounts.put(txn, account, span)?; +impl IndexSlot { + /// Creates a lazy index slot with its superblock lifecycle state. + pub(crate) fn new(directory: &Path, active: bool) -> Self { + Self { + directory: directory.to_owned(), + state: Mutex::new(IndexState { active, cached: None }), } - Ok(()) } - /// Locates a transaction by its first signature. - pub(crate) fn transaction<'t, 'e>( - &'e self, - signature: &Signature, - txn: OptRoTxn<'t, 'e>, - ) -> heed::Result> { - let txn = read_txn(&self.env, txn)?; - self.transactions.get(txn, signature) + /// Opens or leases this superblock's index. + pub(crate) fn get(&self) -> Result> { + let mut state = self.state.lock(); + if let Some(cached) = &mut state.cached { + cached.used = Instant::now(); + return Ok(cached.index.clone()); + } + let workers = if state.active { ACTIVE_WORKERS } else { SEALED_WORKERS }; + let index = Arc::new(Index::new(&self.directory, workers)?); + state.cached = Some(CachedIndex::new(index.clone())); + Ok(index) } - /// Locates a block boundary by slot. - pub(crate) fn block<'t, 'e>( - &'e self, - slot: &Slot, - txn: OptRoTxn<'t, 'e>, - ) -> heed::Result> { - let txn = read_txn(&self.env, txn)?; - self.blocks.get(txn, slot) + /// Marks the former writable index as sealed and cache-eligible. + pub(crate) fn seal(&self) { + let mut state = self.state.lock(); + state.active = false; + if let Some(cached) = &mut state.cached { + cached.used = Instant::now(); + } } - /// Returns execution spans that mention `pubkey`. - pub(crate) fn accounts<'t, 'e>( - &'e self, - pubkey: &Pubkey, - txn: OptRoTxn<'t, 'e>, - ) -> heed::Result>> { - let txn = read_txn(&self.env, txn)?; - self.accounts.get_duplicates(txn, pubkey) + /// Returns the last use of an opened sealed index. + pub(crate) fn last_used(&self) -> Option { + let state = self.state.lock(); + if state.active { + return None; + } + state.cached.as_ref().map(|cached| cached.used) } -} -impl DatabaseIndex for Index { - fn env(&self) -> &Env { - &self.env + /// Closes the index when no external reader or writer lease remains. + pub(crate) fn evict(&self) -> bool { + let mut state = self.state.lock(); + if state.active { + return false; + } + let Some(cached) = &state.cached else { + return true; + }; + if Arc::strong_count(&cached.index) != 1 { + return false; + } + state.cached.take(); + true } } -impl> From for SignatureKey { - fn from(signature: S) -> Self { - Self(array::from_fn(|i| signature.as_ref().as_array()[i])) +impl CachedIndex { + fn new(index: Arc) -> Self { + Self { index, used: Instant::now() } } } -impl<'a> BytesEncode<'a> for SignatureKey { - type EItem = Signature; - - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - Ok(item.as_array()[..KEY_BYTES].into()) +impl IndexReader { + /// Locates a transaction by its first signature. + pub(crate) fn transaction(&self, signature: &Signature) -> Result> { + let Some(value) = + self.snapshot.get(&self.index.transactions, prefix(signature.as_array()))? + else { + return Ok(None); + }; + let value = fixed::(&value, "invalid transaction span value")?; + Ok(Some(TxSpan { + blockstore: Span::from_value(&value[..SPAN_BYTES])?, + execution: Span::from_value(&value[SPAN_BYTES..])?, + })) } -} - -impl<'a> BytesEncode<'a> for AccountKey { - type EItem = Pubkey; - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - Ok(item.as_array()[..KEY_BYTES].into()) + /// Locates a block boundary by slot. + pub(crate) fn block(&self, slot: Slot) -> Result> { + self.snapshot + .get(&self.index.blocks, slot.to_be_bytes())? + .map(|value| Span::from_value(&value)) + .transpose() } -} - -impl<'a> BytesDecode<'a> for SignatureKey { - type DItem = &'a Self; - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - bytemuck::try_from_bytes(bytes).map_err(Into::into) + /// Scans a bounded slot range newest first. + pub(crate) fn blocks(&self, slots: RangeInclusive) -> BlockIter { + let start = slots.start().to_be_bytes(); + let end = slots.end().to_be_bytes(); + BlockIter { + inner: self.snapshot.range(&self.index.blocks, start..=end).rev(), + } } -} -impl<'a> BytesDecode<'a> for AccountKey { - type DItem = &'a Self; - - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - bytemuck::try_from_bytes(bytes).map_err(Into::into) + /// Returns execution spans that mention `pubkey`, newest first. + pub(crate) fn accounts(&self, pubkey: &Pubkey, before: Option) -> AccountIter { + let prefix = prefix(pubkey.as_array()); + let inner = match before { + Some(span) => self.snapshot.range( + &self.index.accounts, + prefixed_range(prefix, ..span.key_bytes()), + ), + None => self.snapshot.prefix(&self.index.accounts, prefix), + }; + AccountIter { inner: inner.rev() } } } -impl<'a> BytesEncode<'a> for Span { - type EItem = Self; - - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - U64Le::bytes_encode(&item.0) +impl IndexWriter { + /// Indexes a block boundary by slot. + pub(crate) fn insert_block(&mut self, slot: Slot, span: Span) { + self.batch.insert(&self.index.blocks, slot.to_be_bytes(), span.value_bytes()); } -} -impl<'a> BytesDecode<'a> for Span { - type DItem = Self; - - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - U64Le::bytes_decode(bytes).map(Self) + /// Indexes a transaction and its execution details. + pub(crate) fn insert_transaction(&mut self, signature: &Signature, span: TxSpan) { + let mut value = [0; TX_SPAN_BYTES]; + value[..SPAN_BYTES].copy_from_slice(&span.blockstore.value_bytes()); + value[SPAN_BYTES..].copy_from_slice(&span.execution.value_bytes()); + self.batch.insert( + &self.index.transactions, + prefix(signature.as_array()), + value, + ); } -} -impl<'a> BytesEncode<'a> for AccountSpan { - type EItem = Span; - - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - // `REVERSE_DUP` compares fixed-size values from the end, which makes - // little-endian u64 values sort numerically ascending. Store the - // inverted span so higher execution offsets are returned first. - Ok((!item.0).to_le_bytes().to_vec().into()) + /// Adds account-to-execution entries for all static transaction accounts. + pub(crate) fn insert_accounts(&mut self, accounts: &[Pubkey], span: Span) { + for account in accounts { + let mut key = [0; ACCOUNT_KEY_BYTES]; + key[..PREFIX_BYTES].copy_from_slice(&prefix(account.as_array())); + key[PREFIX_BYTES..].copy_from_slice(&span.key_bytes()); + self.batch.insert(&self.index.accounts, key, []); + } } -} -impl<'a> BytesDecode<'a> for AccountSpan { - type DItem = Span; - - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - U64Le::bytes_decode(bytes).map(|span| Span(!span)) + /// Publishes the pending atomic batch at the requested durability. + pub(crate) fn persist(&mut self, durability: Durability) -> Result<()> { + let batch = std::mem::replace(&mut self.batch, self.index.db.batch()); + if batch.is_empty() { + if durability.requires_sync() { + self.index.db.persist(PersistMode::SyncData)?; + } + return Ok(()); + } + let mode = match durability { + Durability::Buffer => PersistMode::Buffer, + Durability::SyncData => PersistMode::SyncData, + }; + batch.durability(Some(mode)).commit().map_err(Into::into) } } -impl<'a> BytesEncode<'a> for TxSpan { - type EItem = Self; - - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - Ok(bytemuck::bytes_of(item).into()) - } +/// Returns the compact prefix used by signature and account indexes. +fn prefix(bytes: &[u8; N]) -> Prefix { + let mut prefix = [0; PREFIX_BYTES]; + prefix.copy_from_slice(&bytes[..PREFIX_BYTES]); + prefix } -impl<'a> BytesDecode<'a> for TxSpan { - type DItem = Self; - - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - bytemuck::try_pod_read_unaligned(bytes).map_err(Into::into) - } +/// Converts persisted bytes to a fixed-width array or reports corruption. +fn fixed(bytes: &[u8], error: &'static str) -> Result<[u8; N]> { + bytes.try_into().map_err(|_| LedgerError::Corruption(error)) } diff --git a/ledger/src/lib.rs b/ledger/src/lib.rs index 8af03947..0b20173b 100644 --- a/ledger/src/lib.rs +++ b/ledger/src/lib.rs @@ -6,6 +6,7 @@ use std::{ path::{Path, PathBuf}, sync::{Arc, atomic::Ordering::*}, thread, + time::{Duration, Instant}, }; pub use crate::error::{LedgerError, LedgerRequestError}; @@ -13,10 +14,10 @@ use derive_more::Deref; use nucleus::{ Slot, ledger::BlockstorePosition, - shutdown::{Service, ShutdownManager}, + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, }; use parking_lot::RwLock; -use tokio::sync::broadcast; +use tokio::{sync::broadcast, time}; use tracing::info; mod appender; @@ -34,7 +35,7 @@ mod tests; use crate::{ appender::{BLOCKSTORE_DB, EXECUTIONS_DB, LedgerAppender, SUPERBLOCK_META}, error::Result, - index::Index, + index::IndexSlot, reader::LedgerReader, request::ReaderSender, schema::Event, @@ -43,6 +44,10 @@ use crate::{ const LEDGER_META: &str = "ledger.meta"; const SERVICE_QUEUE_CAPACITY: usize = 128; +/// Sealed indexes immediately preceding the active head kept open when cached. +const HOT_SEALED_INDEXES: u64 = 2; +/// Idle time after which a sealed index is closed. +const SEALED_INDEX_TTL: Duration = Duration::from_secs(10 * 60); /// Top-level ledger handle. /// @@ -60,7 +65,8 @@ pub struct Ledger { } impl Ledger { - /// Opens the ledger and starts one appender plus the reader worker pool. + /// Opens ledger files and starts one appender plus the reader worker pool. + /// The appender opens the active index and reports recovery failures through shutdown. pub fn init( directory: impl AsRef, size_limit: u64, @@ -69,13 +75,21 @@ impl Ledger { let directory = directory.as_ref().to_owned(); let ledger = Arc::new(Self::new(directory, size_limit)?); metrics::init(&ledger); + let sh = shutdown.handle(Service::LedgerMaintenance); + let maintenance = ledger.clone(); + tokio::spawn(async move { maintenance.maintain(sh).await }); let (appender_tx, rx) = flume::bounded(SERVICE_QUEUE_CAPACITY); let (position, _) = broadcast::channel(256); - let appender = LedgerAppender::new(ledger.clone(), rx, position.clone())?; - let sh = shutdown.handle(Service::LedgerAppender); - thread::Builder::new() - .name("ledger-appender".into()) - .spawn(|| appender.run(sh))?; + let mut sh = shutdown.handle(Service::LedgerAppender); + let appender_ledger = ledger.clone(); + let appender_position = position.clone(); + thread::Builder::new().name("ledger-appender".into()).spawn(move || { + let reason = match LedgerAppender::run(appender_ledger, rx, appender_position) { + Ok(()) => ShutdownReason::Signalled, + Err(error) => ShutdownReason::Error(Box::new(error)), + }; + sh.terminate(reason); + })?; let (reader_tx, rx) = flume::bounded(SERVICE_QUEUE_CAPACITY); #[cfg(not(feature = "testkit"))] @@ -100,7 +114,7 @@ impl Ledger { } /// Iterates retained superblocks from newest to oldest. - pub fn iter(&self) -> impl Iterator> { + pub fn iter(&self) -> impl Iterator> + '_ { let range = self.meta.superblocks(); range.rev().filter_map(|id| self.superblocks.read().get(&id).cloned()) } @@ -115,16 +129,24 @@ impl Ledger { /// Blockstore write offset of a retained superblock, `None` when it is not retained. pub fn cursor(&self, superblock: u64) -> Option { - self.iter().find_map(|sb| { - (sb.id == superblock).then_some(sb.meta.cursors.blockstore.load(Acquire)) - }) + if !self.meta.superblocks().contains(&superblock) { + return None; + } + self.superblocks + .read() + .get(&superblock) + .map(|superblock| superblock.meta.cursors.blockstore.load(Acquire)) } /// Iterates retained superblocks after `superblock` through the active head, /// so replay excludes the sealed snapshot state but includes the unsealed head. - fn iter_after(&self, superblock: u64) -> impl Iterator> { - let range = superblock + 1..=self.meta.head(); - range.filter_map(|id| self.superblocks.read().get(&id).cloned()) + fn iter_after(&self, superblock: u64) -> impl Iterator> + '_ { + let head = self.meta.head(); + superblock + .checked_add(1) + .into_iter() + .flat_map(move |start| start..=head) + .filter_map(|id| self.superblocks.read().get(&id).cloned()) } /// Opens ledger metadata and retained superblocks without starting services. @@ -135,9 +157,11 @@ impl Ledger { // and all fields that can change while mapped are atomic. This process // exclusively creates and updates the metadata file at `meta`. let meta = unsafe { MetaMap::::new(&meta) }?; + let retained = meta.superblocks(); let mut superblocks = BTreeMap::new(); - for id in meta.superblocks() { - let superblock = Superblock::open(&directory, id)?; + let head = meta.head(); + for id in retained { + let superblock = Superblock::open(&directory, id, id == head)?; superblocks.insert(id, superblock); } @@ -170,28 +194,62 @@ impl Ledger { /// over its size limit; calling it directly forces a single retention pass /// regardless of that limit. pub fn truncate(&self) -> Result<()> { - let _timer = metrics::time(metrics::Operation::Truncate); - let Some((id, superblock)) = self - .superblocks - .read() - .first_key_value() - .map(|(id, superblock)| (*id, superblock.clone())) - else { + let timer = metrics::time(metrics::Operation::Truncate); + let mut superblocks = self.superblocks.write(); + let Some((&id, _)) = superblocks.first_key_value() else { return Ok(()); }; if id >= self.meta.head() { return Ok(()); } + let Some((_, superblock)) = superblocks.pop_first() else { + return Ok(()); + }; let end = superblock.meta.range.end.load(Acquire); - superblock.purge()?; - self.superblocks.write().remove(&id); self.meta.superblocks.fetch_sub(1, Release); self.meta.range.start.store(end + 1, Release); self.meta.flush()?; - info!(id, end, "purged oldest superblock for retention"); + drop(superblocks); + // index drop and directory cleanup can take a few seconds, + // we don't want to block ledger appender during that time + thread::spawn(move || { + { + superblock.index.evict(); + superblock.purge()?; + info!(id, end, "purged oldest superblock"); + drop(timer); + Ok::<(), LedgerError>(()) + } + .inspect_err(|error| tracing::error!(%error, "failed to purge superblock")) + }); Ok(()) } + + /// Maintains the sealed-index cache. + async fn maintain(&self, mut shutdown: ShutdownHandle) { + loop { + self.expire_indexes(); + if time::timeout(Duration::from_secs(1), shutdown.signalled()).await.is_ok() { + break; + } + } + shutdown.terminate(ShutdownReason::Signalled); + } + + /// Closes expired indexes except the two sealed predecessors of the head. + fn expire_indexes(&self) { + let expiry = Instant::now() - SEALED_INDEX_TTL; + let cold = ..self.meta.head().saturating_sub(HOT_SEALED_INDEXES); + self.superblocks + .read() + .range(cold) + .map(|(_, superblock)| superblock) + .filter(|superblock| superblock.index.last_used().is_some_and(|used| used <= expiry)) + .for_each(|superblock| { + superblock.index.evict(); + }); + } } /// Cloneable senders over shared ledger state: append events, read requests, and @@ -211,7 +269,7 @@ pub struct LedgerHandle { } impl LedgerHandle { - /// Returns the number of transactions published at durable sync boundaries. + /// Returns the number of transactions published at complete block boundaries. pub fn transactions(&self) -> u64 { self.ledger.meta.transactions.load(Acquire) } @@ -221,7 +279,7 @@ impl LedgerHandle { self.ledger.meta.head() } - /// Position of the next byte to append: active superblock plus its durable cursor. + /// Position of the next byte to append: active superblock plus its published cursor. pub fn position(&self) -> BlockstorePosition { let superblock = self.head(); let offset = self.cursor(superblock).unwrap_or_default(); @@ -239,8 +297,8 @@ pub struct Superblock { pub blockstore: File, /// Transaction execution metadata file. executions: File, - /// LMDB index for this superblock. - index: Arc, + /// Lazily opened Fjall index for this superblock. + index: IndexSlot, /// Superblock directory path. pub directory: PathBuf, } @@ -263,9 +321,9 @@ impl Superblock { } /// Opens a superblock directory, creating its data files when needed. - fn open(root: &Path, id: u64) -> Result> { + fn open(root: &Path, id: u64, active: bool) -> Result> { let directory = Self::init_dir(root, id)?; - let index = Arc::new(Index::new(&directory)?); + let index = IndexSlot::new(&directory, active); let meta = unsafe { MetaMap::::new(&directory.join(SUPERBLOCK_META)) }?; let blockstore = Self::file(&directory.join(BLOCKSTORE_DB))?; let executions = Self::file(&directory.join(EXECUTIONS_DB))?; diff --git a/ledger/src/metrics.rs b/ledger/src/metrics.rs index 30387b5c..3c4e6a95 100644 --- a/ledger/src/metrics.rs +++ b/ledger/src/metrics.rs @@ -57,6 +57,8 @@ pub(crate) enum Operation { Truncate, /// Data-file sync path. FileSync, + /// Data sync to OS buffers + BufferSync, /// Data-file finalization path. FileFinalize, } @@ -75,6 +77,7 @@ impl MetricOperation for Operation { Operation::Truncate => "truncate", Operation::FileSync => "file_sync", Operation::FileFinalize => "file_finalize", + Operation::BufferSync => "buffer_sync", } } } diff --git a/ledger/src/reader.rs b/ledger/src/reader.rs index cb7750e0..c177ca08 100644 --- a/ledger/src/reader.rs +++ b/ledger/src/reader.rs @@ -14,7 +14,6 @@ use bitcode::Buffer; use flume::Receiver; use nucleus::{ Slot, - heed::OptRoTxn, shutdown::{ShutdownHandle, ShutdownReason}, }; use solana_signature::Signature; @@ -24,7 +23,7 @@ use zstd::bulk::Decompressor; use crate::{ Ledger, LedgerError, Result, Superblock, - index::{AccountIter, Span}, + index::{IndexReader, Span}, metrics::{self, Operation}, request::{ AccountSignature, AccountSignaturesParams, AccountSignaturesPayload, @@ -113,7 +112,8 @@ impl LedgerReader { if request.cancelled() { return Ok(None); } - let Some(spans) = superblock.index.transaction(&request.params, &mut None)? else { + let index = superblock.index.get()?.reader(); + let Some(spans) = index.transaction(&request.params)? else { continue; }; return Ok(Some(TransactionResponse { @@ -133,7 +133,8 @@ impl LedgerReader { if request.cancelled() { return Ok(None); } - let Some(spans) = superblock.index.transaction(&request.params, &mut None)? else { + let index = superblock.index.get()?.reader(); + let Some(spans) = index.transaction(&request.params)? else { continue; }; let header = self.header(&superblock, spans.execution)?; @@ -160,33 +161,19 @@ impl LedgerReader { if request.cancelled() { return Ok(signatures); } - let mut txn = None; - let Some(iter) = superblock.index.accounts(&pubkey, &mut txn)? else { - continue; - }; - // SAFETY: `iter` borrows the read transaction stored in `txn`. - // `txn` was populated by `accounts`, remains in this stack frame, - // is not replaced or dropped while `iter` is used, and every later - // index read only reuses that already-open read transaction. - let iter = unsafe { mem::transmute::, AccountIter<'_>>(iter) }; - + let index = superblock.index.get()?.reader(); let mut upper = None; if let Some(signature) = &before { // Skip newest-first segments until `before`, then include every older segment. - let Some(cutoff) = superblock.index.transaction(signature, &mut txn)? else { + let Some(cutoff) = index.transaction(signature)? else { continue; }; upper.replace(cutoff.execution); before.take(); } - for result in iter { - let span = result?.1; - if let Some(s) = upper - && span >= s - { - continue; - }; + for result in index.accounts(&pubkey, upper) { + let span = result?; let header = self.header(&superblock, span)?; let sig: Signature = header.signature; if let Some(signature) = until @@ -198,7 +185,7 @@ impl LedgerReader { let blocktime = match blocktimes.get(&header.slot) { Some(time) => *time, None => { - let time = self.blocktime(&superblock, header.slot, &mut txn)?; + let time = self.blocktime(&superblock, &index, header.slot)?; blocktimes.insert(header.slot, time); time } @@ -232,7 +219,8 @@ impl LedgerReader { if !superblock.meta.range.contains(&slot) { continue; } - let Some(span) = superblock.index.block(&slot, &mut None)? else { + let index = superblock.index.get()?.reader(); + let Some(span) = index.block(slot)? else { return Ok(None); }; position.replace((superblock, span)); @@ -245,22 +233,19 @@ impl LedgerReader { /// Reads block boundaries for a slot range in ascending slot order. fn blocks(&mut self, range: Range) -> Result> { let mut blocks = Vec::with_capacity(range.clone().count()); - let mut range = range.into_iter().rev().peekable(); + let Some(last) = range.end.checked_sub(1) else { return Ok(blocks) }; for superblock in self.ledger.clone().iter() { - let mut txn = None; let start = superblock.meta.range.start.load(Acquire); - while let Some(&slot) = range.peek() { - // Slots descend and superblocks go newest to oldest, so a slot - // below this segment's start belongs to an older superblock; - // leave it in the iterator rather than consuming it here. - if slot < start { - break; - } - range.next(); - let Some(span) = superblock.index.block(&slot, &mut txn)? else { - continue; - }; + let end = superblock.meta.range.end.load(Acquire); + let start = start.max(range.start); + let end = end.min(last); + if start > end { + continue; + } + let index = superblock.index.get()?.reader(); + for entry in index.blocks(start..=end) { + let (_, span) = entry?; if let BlockstoreEntry::Block(b) = self.blockstore_entry(&superblock, span)? { blocks.push(b); } @@ -353,9 +338,11 @@ impl LedgerReader { if matches!(details, BlockDetails::None) { return Ok(BlockResponse::Bare(block)); } - let mut txn = None; - // Slots are contiguous, so the previous block boundary is always `slot - 1`. - let start = match superblock.index.block(&(slot - 1), &mut txn)? { + let index = superblock.index.get()?.reader(); + // Slots are contiguous, so the previous block boundary is `slot - 1` + // when this is not the first possible slot. + let previous = slot.checked_sub(1).map(|slot| index.block(slot)).transpose()?.flatten(); + let start = match previous { Some(previous) => previous.offset() + previous.size(), None => 0, }; @@ -391,7 +378,7 @@ impl LedgerReader { signatures.push(signature); continue; } - let Some(spans) = superblock.index.transaction(&signature, &mut txn)? else { + let Some(spans) = index.transaction(&signature)? else { continue; }; let execution = self.execution(superblock, spans.execution)?; @@ -416,13 +403,13 @@ impl LedgerReader { } /// Reads the timestamp from a block boundary entry. - fn blocktime<'t, 'e>( + fn blocktime( &mut self, - superblock: &'e Superblock, + superblock: &Superblock, + index: &IndexReader, slot: Slot, - txn: OptRoTxn<'t, 'e>, ) -> Result { - let Some(span) = superblock.index.block(&slot, txn)? else { + let Some(span) = index.block(slot)? else { return Ok(0); }; let entry = self.blockstore_entry(superblock, span)?; diff --git a/ledger/src/storage.rs b/ledger/src/storage.rs index e4fc99d4..d5ed9504 100644 --- a/ledger/src/storage.rs +++ b/ledger/src/storage.rs @@ -36,6 +36,22 @@ const PREALLOCATION_SIZE: u64 = 4 * nucleus::GB as u64; /// Remaining allocation that triggers another reservation. const PREALLOCATION_THRESHOLD: u64 = PREALLOCATION_SIZE / 16; +/// Persistence strength for a complete ledger boundary. +#[derive(Clone, Copy, Debug)] +pub(crate) enum Durability { + /// Flushes through userspace into operating-system buffers. + Buffer, + /// Additionally synchronizes file data to the storage device. + SyncData, +} + +impl Durability { + /// Returns whether persistence must synchronize data to the storage device. + pub(crate) fn requires_sync(self) -> bool { + matches!(self, Self::SyncData) + } +} + /// Buffered append writer that tracks logical file position. pub(crate) struct AppendFile { /// Backing file. @@ -92,13 +108,15 @@ impl AppendFile { Ok(()) } - /// Flushes buffered bytes, syncs file data, and returns the durable cursor. - pub(crate) fn sync(&mut self) -> Result { + /// Persists buffered bytes at `durability` and returns the published cursor. + pub(crate) fn persist(&mut self, durability: Durability) -> Result { if self.len.saturating_sub(self.cursor) < PREALLOCATION_THRESHOLD { self.preallocate()?; } self.flush()?; - self.file.sync_data()?; + if durability.requires_sync() { + self.file.sync_data()?; + } Ok(self.cursor) } @@ -114,7 +132,6 @@ impl AppendFile { offset, PREALLOCATION_SIZE, )?; - self.file.sync_all()?; self.len = offset + PREALLOCATION_SIZE; debug!(len = self.len, "preallocated ledger file space"); Ok(()) @@ -200,6 +217,21 @@ impl MetaMap { pub(crate) fn flush(&self) -> Result<()> { self.mmap.flush().map_err(Into::into) } + + /// Schedules dirty metadata pages for asynchronous writeback. + #[inline] + pub(crate) fn flush_async(&self) -> Result<()> { + self.mmap.flush_async().map_err(Into::into) + } + + /// Publishes metadata using the boundary's persistence strength. + #[inline] + pub(crate) fn persist(&self, durability: Durability) -> Result<()> { + match durability { + Durability::Buffer => self.flush_async(), + Durability::SyncData => self.flush(), + } + } } impl Deref for MetaMap { @@ -264,7 +296,7 @@ impl LedgerMeta { #[derive(Default)] #[repr(C)] pub(crate) struct SuperblockMeta { - /// Durable append cursors for files in this superblock. + /// Published append cursors for files in this superblock. pub(crate) cursors: FileCursors, /// Slot range stored in this segment. pub(crate) range: BlockRange, @@ -274,13 +306,13 @@ pub(crate) struct SuperblockMeta { pub(crate) transactions: AtomicU64, } -/// Durable append cursors for superblock data files. +/// Published append cursors for superblock data files. #[derive(Default)] #[repr(C)] pub(crate) struct FileCursors { - /// Durable byte cursor in `blockstore.db`. + /// Published byte cursor in `blockstore.db`. pub(crate) blockstore: AtomicU64, - /// Durable byte cursor in `executions.db`. + /// Published byte cursor in `executions.db`. pub(crate) executions: AtomicU64, } diff --git a/ledger/src/tests/index.rs b/ledger/src/tests/index.rs index 4307eb36..518306a9 100644 --- a/ledger/src/tests/index.rs +++ b/ledger/src/tests/index.rs @@ -1,30 +1,35 @@ //! Index unit tests. +use std::sync::Arc; + use nucleus::{ Slot, - heed::DatabaseIndex, testkit::{TempDir, init_tracing, tempdir}, }; use solana_pubkey::Pubkey; use solana_signature::Signature; -use crate::index::{Index, Span, TxSpan}; +use crate::{ + index::{Index, Span, TxSpan}, + storage::Durability, +}; /// Opens a fresh index on a throwaway directory kept alive by the returned guard. -fn index() -> (TempDir, Index) { +fn index() -> (TempDir, Arc) { init_tracing(); let dir = tempdir(); - let index = Index::new(dir.path()).unwrap(); + let index = Arc::new(Index::new(dir.path(), 2).unwrap()); (dir, index) } /// Drains every execution span the account index holds for `pubkey`. -fn account_spans(index: &Index, pubkey: &Pubkey) -> Vec { - let mut txn = None; - let Some(iter) = index.accounts(pubkey, &mut txn).unwrap() else { - return Vec::new(); - }; - iter.map(|entry| entry.unwrap().1).collect() +fn account_spans(index: &Arc, pubkey: &Pubkey) -> Vec { + index + .clone() + .reader() + .accounts(pubkey, None) + .map(|entry| entry.unwrap()) + .collect() } #[test] @@ -58,22 +63,27 @@ fn transaction_and_block_roundtrip() { let block_a = Span::new(0, 8); let block_b = Span::new(8, 16); - let mut txn = index.env().write_txn().unwrap(); - index.insert_transaction(&mut txn, &signature, &txspan).unwrap(); - index.insert_block(&mut txn, &slot_a, &block_a).unwrap(); - index.insert_block(&mut txn, &slot_b, &block_b).unwrap(); - txn.commit().unwrap(); + let mut writer = index.clone().writer(); + writer.insert_transaction(&signature, txspan); + writer.insert_block(slot_a, block_a); + writer.insert_block(slot_b, block_b); + writer.persist(Durability::Buffer).unwrap(); - let mut txn = None; - let got = index.transaction(&signature, &mut txn).unwrap().expect("transaction present"); + let reader = index.clone().reader(); + let got = reader.transaction(&signature).unwrap().expect("transaction present"); assert_eq!(got.blockstore, txspan.blockstore); assert_eq!(got.execution, txspan.execution); - assert_eq!(index.block(&slot_a, &mut txn).unwrap(), Some(block_a)); - assert_eq!(index.block(&slot_b, &mut txn).unwrap(), Some(block_b)); + assert_eq!(reader.block(slot_a).unwrap(), Some(block_a)); + assert_eq!(reader.block(slot_b).unwrap(), Some(block_b)); + assert_eq!( + reader.blocks(slot_a..=slot_b).map(|entry| entry.unwrap().0).collect::>(), + vec![slot_b, slot_a], + "big-endian slot keys scan newest first" + ); // Absent keys resolve to nothing rather than a stale or default hit. - assert!(index.transaction(&Signature::from([9; 64]), &mut txn).unwrap().is_none()); - assert_eq!(index.block(&99, &mut txn).unwrap(), None); + assert!(reader.transaction(&Signature::from([9; 64])).unwrap().is_none()); + assert_eq!(reader.block(99).unwrap(), None); } #[test] @@ -84,12 +94,12 @@ fn account_signature_duplicates() { let spans = [Span::new(100, 10), Span::new(200, 20), Span::new(300, 30)]; let other_span = Span::new(400, 40); - let mut txn = index.env().write_txn().unwrap(); + let mut writer = index.clone().writer(); for span in &spans { - index.insert_accounts(&mut txn, &[account], span).unwrap(); + writer.insert_accounts(&[account], *span); } - index.insert_accounts(&mut txn, &[other], &other_span).unwrap(); - txn.commit().unwrap(); + writer.insert_accounts(&[other], other_span); + writer.persist(Durability::Buffer).unwrap(); // Account duplicate spans are newest-first, so later execution offsets are returned first. assert_eq!( @@ -99,4 +109,36 @@ fn account_signature_duplicates() { // Duplicates stay partitioned per account key. assert_eq!(account_spans(&index, &other), vec![other_span]); + + let reader = index.clone().reader(); + assert_eq!( + reader + .accounts(&account, Some(spans[2])) + .map(|entry| entry.unwrap()) + .collect::>(), + vec![spans[1], spans[0]], + "exclusive big-endian cutoff starts pagination below the requested span" + ); +} + +/// Proves sealed indexes open on demand and cannot close while a read lease exists. +#[test] +fn sealed_index_is_lazy_and_lease_safe() { + let dir = tempdir(); + let superblock = crate::Superblock::open(dir.path(), 1, false).unwrap(); + assert!(superblock.index.last_used().is_none()); + + let lease = superblock.index.get().unwrap(); + assert!(superblock.index.last_used().is_some()); + assert!( + !superblock.index.evict(), + "an outstanding lease prevents close" + ); + drop(lease); + assert!(superblock.index.evict()); + assert!(superblock.index.last_used().is_none()); + + // Reopening after a full close proves the exclusive Fjall lock was released. + drop(superblock.index.get().unwrap()); + assert!(superblock.index.evict()); } diff --git a/ledger/src/tests/integration.rs b/ledger/src/tests/integration.rs index bceb391b..84e228ac 100644 --- a/ledger/src/tests/integration.rs +++ b/ledger/src/tests/integration.rs @@ -8,13 +8,16 @@ //! appender's account/signature extraction and the reader's block reconstruction //! exercise the real codecs. //! -//! The appender only makes data durable at a block boundary (sync + index -//! commit + cursor publish), so every append batch here ends with a `Block`; +//! The appender publishes complete data at a block boundary (buffer flush + +//! index commit + cursor publish), so every append batch here ends with a `Block`; //! that also mirrors how a caller must frame writes. use std::{ + env, ops::Range, + process::Command, sync::{Arc, atomic::Ordering::Acquire}, + time::Duration, }; use nucleus::{ @@ -26,11 +29,14 @@ use nucleus::{ use solana_pubkey::Pubkey; use solana_signature::Signature; use solana_transaction_error::TransactionResult; -use tokio::sync::{broadcast, mpsc}; +use tokio::{ + sync::{broadcast, mpsc}, + time, +}; use crate::{ Ledger, - appender::LedgerAppender, + appender::{LedgerAppender, SIZE_CHECK_FREQUENCY}, reader::LedgerReader, request::{ AccountSignature, AccountSignaturesParams, BlockDetails, BlockParams, BlockResponse, @@ -65,10 +71,37 @@ fn append(ledger: &Arc, events: Vec) { tx.send(event).unwrap(); } drop(tx); - let mut shutdown = ShutdownManager::default(); - LedgerAppender::new(ledger.clone(), rx, position) - .unwrap() - .run(shutdown.handle(Service::LedgerAppender)); + LedgerAppender::run(ledger.clone(), rx, position).unwrap(); +} + +/// Proves a process crash recovers both the last strong boundary and a complete +/// block published with Fjall `Buffer` durability. +#[tokio::test] +async fn test_process_crash_recovers_buffered_block() { + const CHILD_LEDGER: &str = "LEDGER_CRASH_TEST_DIR"; + if let Some(directory) = env::var_os(CHILD_LEDGER) { + let ledger = Arc::new(Ledger::new(directory.into(), u64::MAX).unwrap()); + block_of(&ledger, 1, 1); + append(&ledger, vec![seal(1)]); + block_of(&ledger, 2, 1); + std::process::exit(77); + } + + let dir = tempdir(); + let status = Command::new(env::current_exe().unwrap()) + .args([ + "--exact", + "tests::integration::test_process_crash_recovers_buffered_block", + "--nocapture", + ]) + .env(CHILD_LEDGER, dir.path()) + .status() + .unwrap(); + assert_eq!(status.code(), Some(77)); + + let ledger = Arc::new(Ledger::new(dir.path().to_owned(), u64::MAX).unwrap()); + assert!(read_block(&ledger, 1, BlockDetails::None).await.is_some()); + assert!(read_block(&ledger, 2, BlockDetails::None).await.is_some()); } /// Execution metadata carrying a recognizable `fee`/`logs` for read assertions. @@ -342,12 +375,12 @@ async fn test_block_detail_levels_partition_transactions() { } // A sealed superblock stays readable after the writer rotates to a new segment, -// and retention purges the oldest sealed superblock — dropping its transactions -// while preserving the active head and advancing the retained slot range. +// and retention evicts the oldest sealed superblock before removing its files +// in the background while preserving the active head and range. #[tokio::test] async fn test_superblock_rotation_and_retention() { // size_limit 0 makes every block boundary trigger a retention pass. - let (_dir, ledger) = ledger(0); + let (dir, ledger) = ledger(0); let old = block_of(&ledger, 1, 1); // Seal superblock 1 and rotate to superblock 2. let events = vec![seal(1)]; @@ -358,7 +391,7 @@ async fn test_superblock_rotation_and_retention() { assert!(read_transaction(&ledger, old[0]).await.is_some()); // Writing a block into the new head triggers truncation of superblock 1. - let new = block_of(&ledger, 2, 1); + let new = block_of(&ledger, SIZE_CHECK_FREQUENCY, 1); assert_eq!(ledger.meta.head(), 2, "active head is never purged"); assert!( ledger.superblocks.read().get(&1).is_none(), @@ -369,6 +402,15 @@ async fn test_superblock_rotation_and_retention() { assert!(read_transaction(&ledger, new[0]).await.is_some()); // Retention advances the retained range past the purged superblock's end. assert_eq!(ledger.meta.range.start.load(Acquire), 2); + + let purged = dir.path().join("superblock-000000001"); + time::timeout(Duration::from_secs(5), async { + while purged.exists() { + time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("background truncation purges the directory"); } // A single-block read resolves a slot living in an older sealed superblock, not diff --git a/nucleus/Cargo.toml b/nucleus/Cargo.toml index aeae2e35..59083623 100644 --- a/nucleus/Cargo.toml +++ b/nucleus/Cargo.toml @@ -21,7 +21,6 @@ config = [ "dep:solana-signer" ] default = [] -heed = ["dep:heed"] ledger = ["solana-hash/copy", "solana-hash/wincode", "wincode/derive"] metrics = ["dep:prometheus", "dep:tracing"] notifier = ["tokio/sync"] @@ -56,7 +55,6 @@ tls = ["dep:solana-instruction-error", "dep:solana-pubkey", "dep:wincode"] [dependencies] derive_more = { workspace = true, optional = true, features = ["deref", "from"] } futures = { workspace = true, features = ["alloc"], optional = true } -heed = { workspace = true, optional = true } oneshot = { workspace = true, features = ["async"], optional = true } prometheus = { workspace = true, optional = true } serde = { workspace = true, features = ["derive"], optional = true } diff --git a/nucleus/README.md b/nucleus/README.md index a463b5b6..94bc691a 100644 --- a/nucleus/README.md +++ b/nucleus/README.md @@ -9,8 +9,6 @@ zero when the system clock predates the epoch. Its default feature set is empty. - `config`: serializable authority, accountsdb, blockstore, and ledger configuration types. Authority serialization includes the complete local keypair; consumers must redact it before exposing serialized output. -- `heed`: LMDB transaction aliases, safe environment-bound transaction reuse - helpers, and the shared `DatabaseIndex` trait. - `shutdown`: ordered cancellation, service handles, and termination reporting. The pacemaker quiesces execution and terminally syncs the ledger before the sequencer and appender tier; remaining backing services stop afterward. diff --git a/nucleus/src/heed.rs b/nucleus/src/heed.rs deleted file mode 100644 index fe95adcf..00000000 --- a/nucleus/src/heed.rs +++ /dev/null @@ -1,37 +0,0 @@ -//! Shared heed index plumbing. - -use ::heed::{Env, Result, RoTxn, RwTxn, WithTls}; - -/// Read-only transaction using heed thread-local storage. -pub type RoTxnTls<'e> = RoTxn<'e, WithTls>; -/// Optional write transaction used by batched updates. -pub type OptRwTxn<'t, 'e> = &'t mut Option>; -/// Optional read transaction used by batched reads. -pub type OptRoTxn<'t, 'e> = &'t mut Option>; - -/// Common access for heed-backed indexes. -pub trait DatabaseIndex { - /// Returns the owning heed environment. - fn env(&self) -> &Env; - - /// Flushes the index databases to durable storage. - fn flush(&self) -> Result<()> { - self.env().force_sync() - } -} - -/// Uses the supplied write transaction or opens one against `env` on demand. -pub fn write_txn<'t, 'e>(env: &'e Env, txn: OptRwTxn<'t, 'e>) -> Result<&'t mut RwTxn<'e>> { - if let Some(txn) = txn { - return Ok(txn); - } - Ok(txn.insert(env.write_txn()?)) -} - -/// Uses the supplied read transaction or opens one against `env` on demand. -pub fn read_txn<'t, 'e>(env: &'e Env, txn: OptRoTxn<'t, 'e>) -> Result<&'t RoTxnTls<'e>> { - if let Some(txn) = txn { - return Ok(txn); - } - Ok(txn.insert(env.read_txn()?)) -} diff --git a/nucleus/src/lib.rs b/nucleus/src/lib.rs index 61b41626..a1f6f545 100644 --- a/nucleus/src/lib.rs +++ b/nucleus/src/lib.rs @@ -5,9 +5,6 @@ use std::time::{Duration, UNIX_EPOCH}; #[cfg(feature = "config")] pub mod config; -#[cfg(feature = "heed")] -pub mod heed; - #[cfg(feature = "shutdown")] pub mod shutdown; diff --git a/nucleus/src/shutdown.rs b/nucleus/src/shutdown.rs index 3dd2396d..ec4a7795 100644 --- a/nucleus/src/shutdown.rs +++ b/nucleus/src/shutdown.rs @@ -31,6 +31,8 @@ pub enum Service { LedgerAppender, /// Ledger read worker. LedgerReader, + /// Ledger sealed-index cache maintenance task. + LedgerMaintenance, /// Ledger replay worker. LedgerReplayer, /// Transaction scheduler service.