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/keeper/src/accessor.rs b/keeper/src/accessor.rs index 5d729b96..076eaf88 100644 --- a/keeper/src/accessor.rs +++ b/keeper/src/accessor.rs @@ -367,6 +367,6 @@ impl SuperblockAccessor<'_> { let (response, ack) = oneshot::channel(); let event = Event::Sync { response, is_final }; self.keeper.ledger.appender.send(event)?; - ack.recv().map_err(LedgerRequestError::from)?.map_err(Into::into) + ack.recv().map_err(LedgerRequestError::from).map_err(Into::into) } } 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..04b65658 100644 --- a/ledger/README.md +++ b/ledger/README.md @@ -1,17 +1,17 @@ # `magicblock-ledger` The ledger stores transaction bytes, execution metadata, block boundaries, -superblock seals, and volatile-state reset markers. History is partitioned into -self-contained superblock directories so retention removes a complete sealed -segment without compacting the active store. +superblock seals, and volatile-state reset markers. History files are partitioned +into superblock directories, while one global Fjall database partitions index +entries into a matching keyspace per superblock. ```text ledger.meta +index/ superblock-000000001/ superblock.meta blockstore.db executions.db - index/ ``` `blockstore.db` is a wincode stream. Blockstore decoding permits allocations up @@ -32,8 +32,30 @@ 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 ledger-wide Fjall index uses two +background workers and one 64 MiB cache across all superblock keyspaces. The +optional `testkit` feature uses one reader worker without changing the on-disk +format. Point lookups read the latest visible value, while range and prefix +iterators carry their own Fjall snapshot guard; the append-only index does not +need a request-wide snapshot. + +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. Sealing also queues +the immutable keyspace's active memtable for background SST flushing so its +journal history can be reclaimed. + +Within each superblock keyspace, a leading byte namespaces transaction, block, +and account entries. The account index stores +`account_tag || 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 +70,12 @@ 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, then returns a worker that +destroys its index keyspace and directory. The appender joins that worker before +starting another truncation and during shutdown, propagating cleanup failures; +the active head is never removed. Fjall may defer physical keyspace reclamation +until in-flight readers release their handles. 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 f9fede63..84e9ff6a 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -2,20 +2,14 @@ use std::{ collections::HashMap, - path::Path, sync::{Arc, atomic::Ordering::*}, + thread::JoinHandle, }; 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 +19,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 +34,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 = 128; /// Background service that appends ledger events into the active superblock. pub(crate) struct LedgerAppender { @@ -49,135 +45,103 @@ 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, + /// Outstanding physical cleanup for the last truncated superblock. + truncation: Option>>, /// 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 = ledger.index.writer(&superblock.index); - Ok(Self { + let mut appender = Self { ledger, writer, index, + truncation: None, rx, 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); + + let result = appender.serve(); + let truncation = appender.join_truncation(); + result.and(truncation) } - /// 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); + /// Processes append events until all senders close or a final sync arrives. + fn serve(&mut self) -> Result<()> { + while let Ok(event) = self.rx.recv() { + match event { + Event::Transaction(transaction) => self.write_transaction(transaction)?, + Event::Execution(execution) => self.write_execution(execution)?, + Event::Block(block) => self.write_block(block)?, + Event::Superblock(seal) => self.seal(seal, false)?, + Event::Bootstrap(seal) => self.seal(seal, true)?, + Event::Reset(slot) => self.write_reset(slot)?, + Event::Sync { response, is_final } => { + self.sync(None)?; + let _ = response.send(()); + if is_final { + return Ok(()); + } + } } } + self.sync(None) } /// 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, &self.ledger.index)?; // 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 = self.ledger.index.writer(&superblock.index); + + 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); + 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 +163,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,39 +175,53 @@ 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.ledger.truncate()?; + 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.truncate()?; } metrics::ledger_counts(&self.ledger); Ok(()) } + /// Joins the previous cleanup before starting another truncation worker. + fn truncate(&mut self) -> Result<()> { + self.join_truncation()?; + self.truncation = self.ledger.truncate()?; + Ok(()) + } + + /// Joins and clears the outstanding truncation worker, if any. + fn join_truncation(&mut self) -> Result<()> { + let Some(worker) = self.truncation.take() else { return Ok(()) }; + worker.join().map_err(|_| LedgerError::TruncationPanic)? + } + /// 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()?; + self.index.rotate_memtable()?; 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,37 +229,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()?; - } - if slot.is_none() { - 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. @@ -295,8 +262,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. @@ -305,28 +281,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(), }) @@ -355,25 +325,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..28ac3beb 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), @@ -35,6 +31,9 @@ pub enum LedgerError { #[error("ledger corruption: {0}")] #[from(skip)] Corruption(&'static str), + /// The background superblock cleanup worker panicked. + #[error("ledger truncation worker panicked")] + TruncationPanic, } /// Errors returned while waiting for a ledger reader response. diff --git a/ledger/src/index.rs b/ledger/src/index.rs index 247d9a5d..0b5058f3 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -1,70 +1,64 @@ -//! LMDB index schema and codecs for ledger blockstore entries. +//! Fjall index schema and codecs for ledger blockstore entries. -use std::{array, borrow::Cow, fs, path::Path}; +use std::{iter::Rev, ops::RangeInclusive, 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, -}; -use nucleus::{ - Slot, - heed::{DatabaseIndex, OptRoTxn, read_txn}, +use fjall::{ + CompressionType, Database, Keyspace, KeyspaceCreateOptions, OwnedWriteBatch, PersistMode, + config::{CompressionPolicy, FilterPolicy}, + util::prefixed_range, }; +use nucleus::{MB, Slot}; use solana_pubkey::Pubkey; use solana_signature::Signature; -use crate::schema::Offset; +use crate::{ + Superblock, + error::{LedgerError, Result}, + schema::Offset, + storage::Durability, +}; -/// Index directory below each superblock directory. +/// Ledger-wide index 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. -const SLOTS_INDEX: &str = "slots"; -/// Account-to-offset index name. -const ACCOUNTS_INDEX: &str = "accounts"; +/// Transaction signature key namespace. +const TRANSACTION: u8 = b't'; +/// Slot-to-offset key namespace. +const BLOCK: u8 = b'b'; +/// Account-to-offset key namespace. +const ACCOUNT: u8 = b'a'; /// 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 a namespaced truncated signature. +const TX_KEY_BYTES: usize = 1 + PREFIX_BYTES; +/// Bytes occupied by a namespaced slot. +const BLOCK_KEY_BYTES: usize = 1 + size_of::(); +/// Bytes occupied by a namespaced account prefix. +const ACCOUNT_PREFIX_BYTES: usize = 1 + PREFIX_BYTES; +/// Bytes occupied by an account prefix and ordered execution span. +const ACCOUNT_KEY_BYTES: usize = ACCOUNT_PREFIX_BYTES + SPAN_BYTES; +/// Ledger-wide block cache capacity. +const CACHE_SIZE: u64 = 64 * MB as u64; +/// Maximum total size of all Fjall journals, as configured by `max_journaling_size`. +const JOURNAL_SIZE: u64 = 256 * MB as u64; +/// Fjall maintenance workers assigned to the ledger index. +const WORKERS: usize = 2; + +/// 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 a compact identity, not a collision-proof one. 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 +85,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 +118,226 @@ pub(crate) struct TxSpan { pub(crate) execution: Span, } -/// Account-index span codec. -pub(crate) enum AccountSpan {} - -/// Duplicate account-index iterator over execution spans. -pub(crate) type AccountIter<'a> = RoIter<'a, AccountKey, AccountSpan, MoveOnCurrentKeyDuplicates>; +/// Reverse account-index iterator over execution spans. +pub(crate) struct AccountIter { + inner: Rev, +} -/// LMDB databases used to locate ledger data. -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, +/// Reverse block-index iterator over `(slot, span)` pairs. +pub(crate) struct BlockIter { + inner: Rev, } -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)? - }; +impl Iterator for BlockIter { + type Item = Result<(Slot, Span)>; - 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()?; - Ok(Self { - env, - transactions, - blocks, - accounts, + fn next(&mut self) -> Option { + self.inner.next().map(|entry| { + let (key, value) = entry.into_inner()?; + let key = key.get(1..).ok_or(LedgerError::Corruption("invalid slot key"))?; + let key = fixed(key, "invalid slot key")?; + Ok((u64::from_be_bytes(key), Span::from_value(&value)?)) }) } +} - /// 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) - } +impl Iterator for AccountIter { + type Item = Result; - /// 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) + fn next(&mut self) -> Option { + self.inner.next().map(|entry| { + let key = entry.key()?; + key.get(ACCOUNT_PREFIX_BYTES..) + .ok_or(LedgerError::Corruption("invalid account index key")) + .and_then(Span::from_key) + }) } +} - /// 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)?; - } - Ok(()) - } +/// Ledger-wide Fjall database partitioned by one keyspace per superblock. +#[derive(Clone)] +pub(crate) struct Index { + db: Database, +} - /// 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) - } +/// Atomic index mutations accumulated for one published ledger boundary. +pub(crate) struct IndexWriter { + /// Mutations pending publication at the next boundary. + batch: OwnedWriteBatch, + /// Active superblock keyspace. + keyspace: Keyspace, + /// Ledger-wide database receiving the pending mutations. + db: Database, +} - /// 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) - } +/// Read access to one retained superblock's append-only index. +pub(crate) struct IndexReader<'a>(&'a Keyspace); - /// 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) +impl Index { + /// Opens or creates the ledger-wide index. + pub(crate) fn new(path: &Path) -> Result { + let db = Database::builder(path.join(INDEX_SUBDIR)) + .cache_size(CACHE_SIZE) + .max_journaling_size(JOURNAL_SIZE) + .worker_threads(WORKERS) + .manual_journal_persist(true) + .journal_compression(CompressionType::None) + .open()?; + Ok(Self { db }) } -} -impl DatabaseIndex for Index { - fn env(&self) -> &Env { - &self.env + /// Opens or creates one superblock keyspace. + pub(crate) fn keyspace(&self, id: u64) -> Result { + let options = KeyspaceCreateOptions::default() + .data_block_compression_policy(CompressionPolicy::disabled()) + .index_block_compression_policy(CompressionPolicy::disabled()) + .filter_policy(FilterPolicy::disabled()); + self.db.keyspace(&Superblock::name(id), || options).map_err(Into::into) } -} -impl> From for SignatureKey { - fn from(signature: S) -> Self { - Self(array::from_fn(|i| signature.as_ref().as_array()[i])) + /// Destroys a truncated superblock keyspace. + pub(crate) fn delete(&self, keyspace: Keyspace) -> Result<()> { + self.db.delete_keyspace(keyspace).map_err(Into::into) } -} -impl<'a> BytesEncode<'a> for SignatureKey { - type EItem = Signature; - - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - Ok(item.as_array()[..KEY_BYTES].into()) + /// Creates the single writer for a superblock keyspace. + pub(crate) fn writer(&self, keyspace: &Keyspace) -> IndexWriter { + let batch = self.db.batch(); + IndexWriter { + db: self.db.clone(), + keyspace: keyspace.clone(), + batch, + } } } -impl<'a> BytesEncode<'a> for AccountKey { - type EItem = Pubkey; - - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - Ok(item.as_array()[..KEY_BYTES].into()) +impl<'a> IndexReader<'a> { + /// Borrows a retained superblock keyspace for point and iterator reads. + pub(crate) fn new(keyspace: &'a Keyspace) -> Self { + Self(keyspace) } -} -impl<'a> BytesDecode<'a> for SignatureKey { - type DItem = &'a Self; + /// Locates a transaction by its first signature. + pub(crate) fn transaction(&self, signature: &Signature) -> Result> { + let Some(value) = self.0.get(transaction_key(signature))? 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..])?, + })) + } - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - bytemuck::try_from_bytes(bytes).map_err(Into::into) + /// Locates a block boundary by slot. + pub(crate) fn block(&self, slot: Slot) -> Result> { + self.0.get(block_key(slot))?.map(|value| Span::from_value(&value)).transpose() } -} -impl<'a> BytesDecode<'a> for AccountKey { - type DItem = &'a Self; + /// Scans a bounded slot range newest first. + pub(crate) fn blocks(&self, slots: RangeInclusive) -> BlockIter { + let start = block_key(*slots.start()); + let end = block_key(*slots.end()); + BlockIter { + inner: self.0.range(start..=end).rev(), + } + } - 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 = account_prefix(pubkey); + let inner = match before { + Some(span) => self.0.range(prefixed_range(prefix, ..span.key_bytes())), + None => self.0.prefix(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.keyspace, block_key(slot), span.value_bytes()); } -} -impl<'a> BytesDecode<'a> for Span { - type DItem = 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.keyspace, transaction_key(signature), value); + } - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - U64Le::bytes_decode(bytes).map(Self) + /// 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 { + self.batch.insert(&self.keyspace, account_key(account, span), []); + } } -} -impl<'a> BytesEncode<'a> for AccountSpan { - type EItem = 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.db.batch()); + if batch.is_empty() { + if durability.requires_sync() { + self.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) + } - 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()) + /// Queues the immutable keyspace's active memtable for background flushing. + pub(crate) fn rotate_memtable(&self) -> Result<()> { + self.keyspace.rotate_memtable()?; + Ok(()) } } -impl<'a> BytesDecode<'a> for AccountSpan { - type DItem = Span; +/// Returns the namespaced transaction key. +fn transaction_key(signature: &Signature) -> [u8; TX_KEY_BYTES] { + let mut key = [0; TX_KEY_BYTES]; + key[0] = TRANSACTION; + key[1..].copy_from_slice(&prefix(signature.as_array())); + key +} - fn bytes_decode(bytes: &'a [u8]) -> CodecResult { - U64Le::bytes_decode(bytes).map(|span| Span(!span)) - } +/// Returns the namespaced, numerically ordered block key. +fn block_key(slot: Slot) -> [u8; BLOCK_KEY_BYTES] { + let mut key = [0; BLOCK_KEY_BYTES]; + key[0] = BLOCK; + key[1..].copy_from_slice(&slot.to_be_bytes()); + key } -impl<'a> BytesEncode<'a> for TxSpan { - type EItem = Self; +/// Returns the namespaced prefix shared by one account's entries. +fn account_prefix(pubkey: &Pubkey) -> [u8; ACCOUNT_PREFIX_BYTES] { + let mut key = [0; ACCOUNT_PREFIX_BYTES]; + key[0] = ACCOUNT; + key[1..].copy_from_slice(&prefix(pubkey.as_array())); + key +} - fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { - Ok(bytemuck::bytes_of(item).into()) - } +/// Returns the namespaced account entry key ordered by execution span. +fn account_key(pubkey: &Pubkey, span: Span) -> [u8; ACCOUNT_KEY_BYTES] { + let mut key = [0; ACCOUNT_KEY_BYTES]; + key[..ACCOUNT_PREFIX_BYTES].copy_from_slice(&account_prefix(pubkey)); + key[ACCOUNT_PREFIX_BYTES..].copy_from_slice(&span.key_bytes()); + key } -impl<'a> BytesDecode<'a> for TxSpan { - type DItem = Self; +/// 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 +} - 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..80e098f2 100644 --- a/ledger/src/lib.rs +++ b/ledger/src/lib.rs @@ -5,15 +5,16 @@ use std::{ fs::{self, File}, path::{Path, PathBuf}, sync::{Arc, atomic::Ordering::*}, - thread, + thread::{self, JoinHandle}, }; pub use crate::error::{LedgerError, LedgerRequestError}; use derive_more::Deref; +use fjall::Keyspace; use nucleus::{ Slot, ledger::BlockstorePosition, - shutdown::{Service, ShutdownManager}, + shutdown::{Service, ShutdownManager, ShutdownReason}, }; use parking_lot::RwLock; use tokio::sync::broadcast; @@ -53,6 +54,8 @@ pub struct Ledger { meta: MetaMap, /// Retained superblocks keyed by id. superblocks: RwLock>>, + /// Ledger-wide index partitioned by superblock keyspace. + index: Index, /// Root ledger directory. pub directory: PathBuf, /// Maximum used bytes allowed on the ledger filesystem before retention runs. @@ -60,7 +63,7 @@ pub struct Ledger { } impl Ledger { - /// Opens the ledger and starts one appender plus the reader worker pool. + /// Opens ledger files and the global index, then starts the appender and reader pool. pub fn init( directory: impl AsRef, size_limit: u64, @@ -71,11 +74,16 @@ impl Ledger { metrics::init(&ledger); 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 +108,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,29 +123,39 @@ 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. fn new(directory: PathBuf, size_limit: u64) -> Result { fs::create_dir_all(&directory)?; + let index = Index::new(&directory)?; let meta = directory.join(LEDGER_META); // SAFETY: `LedgerMeta` and its nested headers have stable C layouts, // 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)?; + for id in retained { + let superblock = Superblock::open(&directory, id, &index)?; superblocks.insert(id, superblock); } @@ -145,6 +163,7 @@ impl Ledger { Ok(Self { meta, superblocks: superblocks.into(), + index, directory, size_limit, }) @@ -164,33 +183,36 @@ impl Ledger { /// Removes the oldest sealed superblock while keeping the active head. /// /// Superblock slot ranges are sequential and non-overlapping, so the next - /// retained start slot is the removed superblock end plus one. - /// - /// The appender runs this at a block boundary once the ledger filesystem is - /// 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 { - return Ok(()); + /// retained start slot is the removed superblock end plus one. Metadata is + /// flushed before returning the cleanup worker, which removes the keyspace + /// and directory. The caller must join the worker to observe cleanup errors. + pub fn truncate(&self) -> Result>>> { + let mut superblocks = self.superblocks.write(); + let Some((&id, _)) = superblocks.first_key_value() else { + return Ok(None); }; if id >= self.meta.head() { - return Ok(()); + return Ok(None); } + let Some((_, superblock)) = superblocks.pop_first() else { + return Ok(None); + }; + drop(superblocks); + let timer = metrics::time(metrics::Operation::Truncate); 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"); - Ok(()) + + let index = self.index.clone(); + Ok(Some(thread::spawn(move || { + index.delete(superblock.index.clone())?; + superblock.purge()?; + info!(id, end, "purged oldest superblock"); + drop(timer); + Ok(()) + }))) } } @@ -211,7 +233,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 +243,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,16 +261,21 @@ pub struct Superblock { pub blockstore: File, /// Transaction execution metadata file. executions: File, - /// LMDB index for this superblock. - index: Arc, + /// Keyspace containing this superblock's namespaced index entries. + index: Keyspace, /// Superblock directory path. pub directory: PathBuf, } impl Superblock { + /// Canonical directory and keyspace name for one superblock. + fn name(id: u64) -> String { + format!("superblock-{id:0>9}") + } + /// Returns the directory path for `id` under `root`. pub fn init_dir(root: &Path, id: u64) -> Result { - let dir = root.join(format!("superblock-{id:0>9}")); + let dir = root.join(Self::name(id)); fs::create_dir_all(&dir).map_err(Into::into).map(|()| dir) } @@ -263,9 +290,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, index: &Index) -> Result> { let directory = Self::init_dir(root, id)?; - let index = Arc::new(Index::new(&directory)?); + let index = index.keyspace(id)?; 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..d3245358 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 = IndexReader::new(&superblock.index); + 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 = IndexReader::new(&superblock.index); + 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 = IndexReader::new(&superblock.index); 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 = IndexReader::new(&superblock.index); + 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 = IndexReader::new(&superblock.index); + 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 = IndexReader::new(&superblock.index); + // 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/schema.rs b/ledger/src/schema.rs index cbbd84d7..fe242b31 100644 --- a/ledger/src/schema.rs +++ b/ledger/src/schema.rs @@ -16,7 +16,7 @@ use solana_signature::Signature; use solana_transaction_error::TransactionResult; use wincode::{SchemaRead, SchemaWrite}; -use crate::{error::Result, index::Span}; +use crate::index::Span; /// Byte offset into a ledger data file. pub(crate) type Offset = u64; @@ -77,8 +77,8 @@ pub enum Event { Reset(Slot), /// Flush pending appends and optionally stop the appender after acknowledging. Sync { - /// Receives the durability result. - response: oneshot::Sender>, + /// Receives an acknowledgement after the append state is durable. + response: oneshot::Sender<()>, /// Whether this is the terminal engine sync. is_final: bool, }, 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..8cf800bf 100644 --- a/ledger/src/tests/index.rs +++ b/ledger/src/tests/index.rs @@ -1,30 +1,33 @@ //! Index unit tests. +use fjall::Keyspace; 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, IndexReader, 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, Index, Keyspace) { init_tracing(); let dir = tempdir(); let index = Index::new(dir.path()).unwrap(); - (dir, index) + let keyspace = index.keyspace(1).unwrap(); + (dir, index, keyspace) } /// 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(keyspace: &Keyspace, pubkey: &Pubkey) -> Vec { + IndexReader::new(keyspace) + .accounts(pubkey, None) + .map(|entry| entry.unwrap()) + .collect() } #[test] @@ -48,7 +51,7 @@ fn span_pack_and_order() { #[test] fn transaction_and_block_roundtrip() { - let (_dir, index) = index(); + let (_dir, index, keyspace) = index(); let signature = Signature::from([7; 64]); let txspan = TxSpan { blockstore: Span::new(10, 20), @@ -58,45 +61,77 @@ 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.writer(&keyspace); + 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 = IndexReader::new(&keyspace); + 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] fn account_signature_duplicates() { - let (_dir, index) = index(); + let (_dir, index, keyspace) = index(); let account = Pubkey::new_unique(); let other = Pubkey::new_unique(); 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.writer(&keyspace); 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!( - account_spans(&index, &account), + account_spans(&keyspace, &account), vec![spans[2], spans[1], spans[0]] ); // Duplicates stay partitioned per account key. - assert_eq!(account_spans(&index, &other), vec![other_span]); + assert_eq!(account_spans(&keyspace, &other), vec![other_span]); + + let reader = IndexReader::new(&keyspace); + 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 deleting one superblock keyspace preserves in-flight reads and isolates reuse. +#[test] +fn superblock_keyspace_deletion_is_lease_safe() { + let (_dir, index, keyspace) = index(); + let span = Span::new(10, 20); + let mut writer = index.writer(&keyspace); + writer.insert_block(1, span); + writer.persist(Durability::Buffer).unwrap(); + + let reader = IndexReader::new(&keyspace); + index.delete(keyspace.clone()).unwrap(); + assert_eq!(reader.block(1).unwrap(), Some(span)); + + let replacement = index.keyspace(1).unwrap(); + assert_eq!(IndexReader::new(&replacement).block(1).unwrap(), None); } diff --git a/ledger/src/tests/integration.rs b/ledger/src/tests/integration.rs index bceb391b..e304992f 100644 --- a/ledger/src/tests/integration.rs +++ b/ledger/src/tests/integration.rs @@ -8,12 +8,14 @@ //! 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}, }; @@ -30,7 +32,7 @@ use tokio::sync::{broadcast, mpsc}; use crate::{ Ledger, - appender::LedgerAppender, + appender::{LedgerAppender, SIZE_CHECK_FREQUENCY}, reader::LedgerReader, request::{ AccountSignature, AccountSignaturesParams, BlockDetails, BlockParams, BlockResponse, @@ -65,10 +67,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 +371,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 index +// and 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 +387,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 +398,12 @@ 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"); + assert!( + !purged.exists(), + "appender exit joins background truncation" + ); } // A single-block read resolves a slot living in an older sealed superblock, not diff --git a/ledger/src/tests/mod.rs b/ledger/src/tests/mod.rs index 794fc572..95c60337 100644 --- a/ledger/src/tests/mod.rs +++ b/ledger/src/tests/mod.rs @@ -1,6 +1,6 @@ //! Ledger test modules. //! -//! `index` covers the LMDB codec/index in isolation; `integration` drives the +//! `index` covers the Fjall codec/index in isolation; `integration` drives the //! append→seal→read pipeline end to end through the appender and reader. mod index; 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/replicator/tests/integration.rs b/replicator/tests/integration.rs index b545f489..0a1aa689 100644 --- a/replicator/tests/integration.rs +++ b/replicator/tests/integration.rs @@ -37,6 +37,13 @@ fn loopback_addr() -> SocketAddr { listener.local_addr().unwrap() } +/// Truncates one sealed superblock and waits for its storage cleanup. +fn truncate(ledger: &ledger::LedgerHandle) { + if let Some(worker) = ledger.truncate().unwrap() { + worker.join().expect("truncation worker panicked").unwrap(); + } +} + /// Starts an engine over throwaway directories seeded with `accounts`. async fn engine(authority: Authority, accounts: &[AccountSeed], pacing: Pacing) -> TestEngine { let dirs = Dirs::default(); @@ -472,8 +479,8 @@ async fn restores_the_newest_snapshot_then_streams_its_tail() { leader.advance(1).await; let expected = leader.sync().await; // Make the newest snapshot the only possible handshake response. - leader.ledger().truncate().unwrap(); // superblock 0 - leader.ledger().truncate().unwrap(); // superblock 1 + truncate(leader.ledger()); // superblock 0 + truncate(leader.ledger()); // superblock 1 assert!( leader.ledger().cursor(0).is_none(), "follower cursor was retained away" @@ -554,7 +561,7 @@ async fn cascades_replication_through_a_follower() { await_replication(&mut middle_positions, &middle, expected, state, 2).await; // The successor archive must become the only answer to the tail's cursor. - middle.ledger().truncate().unwrap(); // superblock 0 + truncate(middle.ledger()); // superblock 0 assert!( middle.ledger().cursor(0).is_none(), "tail cursor was retained away"