From 4e6f5a943c320867b29c9bf8a814b9c2cf16f710 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 18 Aug 2026 21:44:10 +0400 Subject: [PATCH 1/6] feat: replace ledger LMDB index with Fjall --- Cargo.lock | 161 +++++++++- Cargo.toml | 1 + accountsdb/Cargo.toml | 2 +- accountsdb/src/lib.rs | 3 +- accountsdb/src/store/index.rs | 42 ++- accountsdb/src/store/mod.rs | 3 +- accountsdb/src/tests.rs | 7 +- ledger/Cargo.toml | 7 +- ledger/README.md | 26 +- ledger/src/appender.rs | 186 ++++++------ ledger/src/error.rs | 8 +- ledger/src/index.rs | 519 +++++++++++++++++++------------- ledger/src/lib.rs | 105 +++++-- ledger/src/metrics.rs | 3 + ledger/src/reader.rs | 77 ++--- ledger/src/storage.rs | 48 ++- ledger/src/tests/index.rs | 92 ++++-- ledger/src/tests/integration.rs | 52 +++- nucleus/Cargo.toml | 2 - nucleus/README.md | 2 - nucleus/src/heed.rs | 37 --- nucleus/src/lib.rs | 3 - nucleus/src/shutdown.rs | 2 + 23 files changed, 893 insertions(+), 495 deletions(-) delete mode 100644 nucleus/src/heed.rs 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 f9fede63..0b3d12c9 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -2,17 +2,14 @@ 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}, }; @@ -25,13 +22,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 +37,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,11 +48,11 @@ 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, @@ -67,15 +66,15 @@ impl LedgerAppender { position: Sender, ) -> 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 { ledger, @@ -91,8 +90,7 @@ impl LedgerAppender { /// 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) { + match self.run_epoch() { Ok(Epoch::Rotate) => (), Ok(Epoch::Shutdown) => break ShutdownReason::Signalled, Err(err) => break ShutdownReason::Error(Box::new(err)), @@ -104,13 +102,12 @@ impl LedgerAppender { } /// Processes events against one stable superblock environment. - fn run_epoch<'e>(&mut self, env: &'e Env) -> Result { - let mut txn: Option> = None; + fn run_epoch(&mut self) -> Result { loop { let Ok(event) = self.rx.recv() else { return Ok(Epoch::Shutdown); }; - if let Some(epoch) = self.process(event, env, &mut txn)? { + if let Some(epoch) = self.process(event)? { return Ok(epoch); } } @@ -120,52 +117,55 @@ impl LedgerAppender { 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> { + fn process(&mut self, event: Event) -> Result> { match event { Event::Transaction(transaction) => { self.write_transaction(transaction)?; } Event::Execution(execution) => { - self.write_execution(execution, env, txn)?; + self.write_execution(execution)?; } Event::Block(block) => { - self.write_block(block, env, txn)?; + self.write_block(block)?; } Event::Superblock(seal) => { - self.seal(seal, false, txn)?; + self.seal(seal, false)?; return Ok(Some(Epoch::Rotate)); } Event::Bootstrap(seal) => { - self.seal(seal, true, txn)?; + self.seal(seal, true)?; return Ok(Some(Epoch::Rotate)); } Event::Reset(slot) => { - self.write_reset(slot, txn)?; + self.write_reset(slot)?; } Event::Sync { response, is_final } => { - let _ = response.send(self.sync(None, txn)); + let _ = response.send(self.sync(None)); if is_final { return Ok(Some(Epoch::Shutdown)); } @@ -176,8 +176,8 @@ impl LedgerAppender { /// 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 +199,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 +211,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 +232,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,25 +251,25 @@ 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(()) @@ -295,8 +290,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 +309,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 +353,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..49946810 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,298 @@ 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, + /// Owning Fjall database. + db: Database, + /// Signature prefix to transaction and execution spans. + transactions: Keyspace, + /// Slot to blockstore span. + blocks: Keyspace, + /// Account prefix and execution span keys. + accounts: Keyspace, +} + +/// 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)?; - } - Ok(()) +impl IndexSlot { + /// Creates an active slot eagerly or a sealed slot for on-demand opening. + pub(crate) fn new(directory: &Path, active: bool) -> Result { + let cached = active + .then(|| Index::new(directory, ACTIVE_WORKERS).map(Arc::new)) + .transpose()? + .map(CachedIndex::new); + Ok(Self { + directory: directory.to_owned(), + state: Mutex::new(IndexState { active, cached }), + }) } - /// 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..256b299f 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. /// @@ -69,6 +74,9 @@ 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())?; @@ -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,16 +123,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 +151,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); } @@ -171,27 +189,58 @@ impl Ledger { /// 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 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); + let start = self.meta.range.start.load(Acquire); 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"); + if let Err(error) = self.meta.flush() { + self.meta.superblocks.fetch_add(1, Release); + self.meta.range.start.store(start, Release); + superblocks.insert(id, superblock); + return Err(error); + } + drop(superblocks); + superblock.index.evict(); + superblock.purge()?; + info!(id, end, "purged oldest 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 +260,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 +270,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 +288,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 +312,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..3433ad2c 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, @@ -71,6 +73,36 @@ fn append(ledger: &Arc, events: Vec) { .run(shutdown.handle(Service::LedgerAppender)); } +/// 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. fn execution(signature: Signature, slot: Slot, result: TransactionResult<()>) -> Execution { Execution { @@ -342,12 +374,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 synchronously purges the oldest sealed superblock 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 +390,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 +401,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(), + "truncation purges the directory before returning" + ); } // 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. From 4bb99d57814d7c0d4cb0d489b889dac1bb986d27 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 18 Aug 2026 23:42:00 +0400 Subject: [PATCH 2/6] fix(ledger): move index structural ops into background --- ledger/src/appender.rs | 91 ++++++++------------------------- ledger/src/index.rs | 18 +++---- ledger/src/lib.rs | 45 +++++++++------- ledger/src/tests/integration.rs | 26 ++++++---- 4 files changed, 69 insertions(+), 111 deletions(-) diff --git a/ledger/src/appender.rs b/ledger/src/appender.rs index 0b3d12c9..458be7f3 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -8,11 +8,7 @@ use std::{ use agave_transaction_view::transaction_view::TransactionView; use bitcode::Buffer; use flume::Receiver; -use nucleus::{ - Slot, - ledger::BlockstorePosition, - shutdown::{ShutdownHandle, ShutdownReason}, -}; +use nucleus::{Slot, ledger::BlockstorePosition}; use solana_signature::Signature; use tokio::sync::broadcast::Sender; use tracing::{info, warn}; @@ -59,12 +55,12 @@ pub(crate) struct LedgerAppender { } 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(); metrics::pending_transactions(0); let superblock = ledger @@ -76,7 +72,7 @@ impl LedgerAppender { let writer = SuperblockWriter::new(superblock.clone())?; let index = superblock.index.get()?.writer(); - Ok(Self { + let mut appender = Self { ledger, writer, index, @@ -84,33 +80,25 @@ 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 { - match self.run_epoch() { - 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(&mut self) -> Result { - loop { - let Ok(event) = self.rx.recv() else { - return Ok(Epoch::Shutdown); - }; - if let Some(epoch) = self.process(event)? { - 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. @@ -141,39 +129,6 @@ impl LedgerAppender { Ok(()) } - /// Processes one append event, rotating after a superblock seal. - fn process(&mut self, event: Event) -> Result> { - 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)?; - return Ok(Some(Epoch::Rotate)); - } - Event::Bootstrap(seal) => { - self.seal(seal, true)?; - return Ok(Some(Epoch::Rotate)); - } - Event::Reset(slot) => { - self.write_reset(slot)?; - } - Event::Sync { response, is_final } => { - let _ = response.send(self.sync(None)); - 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) -> Result<()> { @@ -276,12 +231,6 @@ impl LedgerAppender { } } -/// 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. diff --git a/ledger/src/index.rs b/ledger/src/index.rs index 49946810..2ccf47c4 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -156,14 +156,14 @@ impl Iterator for AccountIter { /// Databases used to locate ledger data inside one superblock. pub(crate) struct Index { - /// Owning Fjall database. - db: 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. @@ -246,16 +246,12 @@ impl Index { } impl IndexSlot { - /// Creates an active slot eagerly or a sealed slot for on-demand opening. - pub(crate) fn new(directory: &Path, active: bool) -> Result { - let cached = active - .then(|| Index::new(directory, ACTIVE_WORKERS).map(Arc::new)) - .transpose()? - .map(CachedIndex::new); - Ok(Self { + /// 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 }), - }) + state: Mutex::new(IndexState { active, cached: None }), + } } /// Opens or leases this superblock's index. diff --git a/ledger/src/lib.rs b/ledger/src/lib.rs index 256b299f..0b20173b 100644 --- a/ledger/src/lib.rs +++ b/ledger/src/lib.rs @@ -65,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, @@ -79,11 +80,16 @@ impl Ledger { 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"))] @@ -188,7 +194,7 @@ 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 timer = metrics::time(metrics::Operation::Truncate); let mut superblocks = self.superblocks.write(); let Some((&id, _)) = superblocks.first_key_value() else { return Ok(()); @@ -201,19 +207,22 @@ impl Ledger { }; let end = superblock.meta.range.end.load(Acquire); - let start = self.meta.range.start.load(Acquire); self.meta.superblocks.fetch_sub(1, Release); self.meta.range.start.store(end + 1, Release); - if let Err(error) = self.meta.flush() { - self.meta.superblocks.fetch_add(1, Release); - self.meta.range.start.store(start, Release); - superblocks.insert(id, superblock); - return Err(error); - } + self.meta.flush()?; drop(superblocks); - superblock.index.evict(); - superblock.purge()?; - info!(id, end, "purged oldest superblock"); + // 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(()) } @@ -314,7 +323,7 @@ impl Superblock { /// Opens a superblock directory, creating its data files when needed. fn open(root: &Path, id: u64, active: bool) -> Result> { let directory = Self::init_dir(root, id)?; - let index = IndexSlot::new(&directory, active)?; + 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/tests/integration.rs b/ledger/src/tests/integration.rs index 3433ad2c..84e228ac 100644 --- a/ledger/src/tests/integration.rs +++ b/ledger/src/tests/integration.rs @@ -17,6 +17,7 @@ use std::{ ops::Range, process::Command, sync::{Arc, atomic::Ordering::Acquire}, + time::Duration, }; use nucleus::{ @@ -28,7 +29,10 @@ 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, @@ -67,10 +71,7 @@ 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 @@ -374,8 +375,8 @@ async fn test_block_detail_levels_partition_transactions() { } // A sealed superblock stays readable after the writer rotates to a new segment, -// and retention synchronously purges the oldest sealed superblock while -// preserving the active head and 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. @@ -403,10 +404,13 @@ async fn test_superblock_rotation_and_retention() { assert_eq!(ledger.meta.range.start.load(Acquire), 2); let purged = dir.path().join("superblock-000000001"); - assert!( - !purged.exists(), - "truncation purges the directory before returning" - ); + 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 From 20c2f7e4f44cfe96a228e5951b706ec148487c5b Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Thu, 20 Aug 2026 11:23:57 +0400 Subject: [PATCH 3/6] refactor(ledger): consolidate Fjall index --- ledger/README.md | 38 +++-- ledger/src/appender.rs | 53 ++++-- ledger/src/error.rs | 3 + ledger/src/index.rs | 287 ++++++++++++-------------------- ledger/src/lib.rs | 107 +++++------- ledger/src/reader.rs | 14 +- ledger/src/tests/index.rs | 65 ++++---- ledger/src/tests/integration.rs | 21 +-- ledger/src/tests/mod.rs | 2 +- nucleus/src/shutdown.rs | 2 - replicator/tests/integration.rs | 13 +- 11 files changed, 261 insertions(+), 344 deletions(-) diff --git a/ledger/README.md b/ledger/README.md index 898695a4..296d08eb 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,11 +32,12 @@ 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 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. +reads only through published cursors. The ledger-wide Fjall index uses two +background workers and one 8 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 @@ -46,11 +47,13 @@ seals, resets, shutdown, and retention boundaries additionally use file 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%. +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 @@ -66,8 +69,11 @@ through the active head in full. At a block boundary, the appender checks used bytes on the ledger filesystem. 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. +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 458be7f3..f4a2fb76 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -3,6 +3,7 @@ use std::{ collections::HashMap, sync::{Arc, atomic::Ordering::*}, + thread::JoinHandle, }; use agave_transaction_view::transaction_view::TransactionView; @@ -34,7 +35,7 @@ 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; +pub(crate) const SIZE_CHECK_FREQUENCY: u64 = 128; /// Background service that appends ledger events into the active superblock. pub(crate) struct LedgerAppender { @@ -46,6 +47,8 @@ pub(crate) struct LedgerAppender { pending: HashMap, /// 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 published boundary. @@ -70,28 +73,36 @@ impl LedgerAppender { .cloned() .ok_or(LedgerError::Corruption("active superblock missing"))?; let writer = SuperblockWriter::new(superblock.clone())?; - let index = superblock.index.get()?.writer(); + let index = ledger.index.writer(&superblock.index); let mut appender = Self { ledger, writer, index, + truncation: None, rx, pending: HashMap::new(), transactions: 0, position, }; - while let Ok(event) = appender.rx.recv() { + let result = appender.serve(); + let truncation = appender.join_truncation(); + result.and(truncation) + } + + /// 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) => 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::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 } => { - let _ = response.send(appender.sync(None)); + let _ = response.send(self.sync(None)); if is_final { break; } @@ -105,22 +116,19 @@ impl LedgerAppender { 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, true)?; + 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); superblock.meta.flush()?; let writer = SuperblockWriter::new(superblock.clone())?; - let index = superblock.index.get()?.writer(); + 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()?; - if let Some(sealed) = superblocks.get(&seal.id) { - sealed.index.seal(); - } drop(superblocks); self.writer = writer; @@ -180,12 +188,25 @@ impl LedgerAppender { 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()?; + 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) -> Result<()> { self.writer.write_blockstore(&BlockstoreEntry::Superblock(seal))?; diff --git a/ledger/src/error.rs b/ledger/src/error.rs index 19839b4f..28ac3beb 100644 --- a/ledger/src/error.rs +++ b/ledger/src/error.rs @@ -31,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 2ccf47c4..3bd0520c 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -1,56 +1,53 @@ //! Fjall index schema and codecs for ledger blockstore entries. -use std::{ - iter::Rev, - ops::RangeInclusive, - path::{Path, PathBuf}, - sync::Arc, - time::Instant, -}; +use std::{iter::Rev, ops::RangeInclusive, path::Path}; 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::{ + Superblock, error::{LedgerError, Result}, schema::Offset, storage::Durability, }; -/// Index directory below each superblock directory. +/// Ledger-wide index directory. const INDEX_SUBDIR: &str = "index"; -/// 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 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 = 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; +const ACCOUNT_KEY_BYTES: usize = ACCOUNT_PREFIX_BYTES + SPAN_BYTES; +/// Ledger-wide block cache capacity. +const CACHE_SIZE: u64 = 64 * MB as u64; +/// Fjall maintenance workers assigned to the ledger index. +const WORKERS: usize = 2; /// Truncated signature or account key. /// -/// The 16-byte prefix is an index tag, not a collision-proof identity. The +/// 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]; @@ -135,7 +132,8 @@ impl Iterator for BlockIter { fn next(&mut self) -> Option { self.inner.next().map(|entry| { let (key, value) = entry.into_inner()?; - let key = fixed(&key, "invalid slot key")?; + 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)?)) }) } @@ -147,173 +145,83 @@ impl Iterator for AccountIter { fn next(&mut self) -> Option { self.inner.next().map(|entry| { let key = entry.key()?; - key.get(PREFIX_BYTES..) + key.get(ACCOUNT_PREFIX_BYTES..) .ok_or(LedgerError::Corruption("invalid account index key")) .and_then(Span::from_key) }) } } -/// Databases used to locate ledger data inside one superblock. +/// Ledger-wide Fjall database partitioned by one keyspace per superblock. +#[derive(Clone)] pub(crate) struct Index { - /// 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, + /// Active superblock keyspace. + keyspace: Keyspace, + /// Ledger-wide database receiving the pending mutations. + db: Database, } -/// 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, -} +/// Read access to one retained superblock's append-only index. +/// +/// Point lookups use Fjall's latest visible sequence number. Range and prefix +/// calls already return iterators carrying their own snapshot-tracker nonce, so +/// retaining a database-wide snapshot here would add tracking work without +/// strengthening these reads. +pub(crate) struct IndexReader<'a>(&'a Keyspace); impl Index { - /// Opens or creates an index using `workers` Fjall maintenance threads. - pub(crate) fn new(path: &Path, workers: usize) -> Result { + /// 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) - .worker_threads(workers) + .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 transactions = db.keyspace(TRANSACTIONS_INDEX, options)?; - let blocks = db.keyspace(SLOTS_INDEX, options)?; - let accounts = db.keyspace(ACCOUNTS_INDEX, options)?; - Ok(Self { - db, - transactions, - blocks, - accounts, - }) - } - - /// Creates the single writer for this superblock index. - pub(crate) fn writer(self: Arc) -> IndexWriter { - let batch = self.db.batch(); - IndexWriter { index: self, batch } - } - - /// 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 } - } -} - -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 }), - } - } - - /// 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) + Ok(Self { db }) } - /// 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(); - } + /// 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) } - /// 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) + /// Destroys a truncated superblock keyspace. + pub(crate) fn delete(&self, keyspace: Keyspace) -> Result<()> { + self.db.delete_keyspace(keyspace).map_err(Into::into) } - /// 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; + /// 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, } - state.cached.take(); - true } } -impl CachedIndex { - fn new(index: Arc) -> Self { - Self { index, used: Instant::now() } +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 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 { + let Some(value) = self.0.get(transaction_key(signature))? else { return Ok(None); }; let value = fixed::(&value, "invalid transaction span value")?; @@ -325,30 +233,24 @@ impl IndexReader { /// 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() + self.0.get(block_key(slot))?.map(|value| Span::from_value(&value)).transpose() } /// 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(); + let start = block_key(*slots.start()); + let end = block_key(*slots.end()); BlockIter { - inner: self.snapshot.range(&self.index.blocks, start..=end).rev(), + inner: self.0.range(start..=end).rev(), } } /// 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 prefix = account_prefix(pubkey); 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), + Some(span) => self.0.range(prefixed_range(prefix, ..span.key_bytes())), + None => self.0.prefix(prefix), }; AccountIter { inner: inner.rev() } } @@ -357,7 +259,7 @@ impl IndexReader { 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()); + self.batch.insert(&self.keyspace, block_key(slot), span.value_bytes()); } /// Indexes a transaction and its execution details. @@ -365,29 +267,22 @@ impl IndexWriter { 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, - ); + self.batch.insert(&self.keyspace, transaction_key(signature), value); } /// 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, []); + self.batch.insert(&self.keyspace, account_key(account, 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()); + let batch = std::mem::replace(&mut self.batch, self.db.batch()); if batch.is_empty() { if durability.requires_sync() { - self.index.db.persist(PersistMode::SyncData)?; + self.db.persist(PersistMode::SyncData)?; } return Ok(()); } @@ -399,6 +294,38 @@ impl IndexWriter { } } +/// 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 +} + +/// 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 +} + +/// 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 +} + +/// 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 +} + /// Returns the compact prefix used by signature and account indexes. fn prefix(bytes: &[u8; N]) -> Prefix { let mut prefix = [0; PREFIX_BYTES]; diff --git a/ledger/src/lib.rs b/ledger/src/lib.rs index 0b20173b..80e098f2 100644 --- a/ledger/src/lib.rs +++ b/ledger/src/lib.rs @@ -5,19 +5,19 @@ use std::{ fs::{self, File}, path::{Path, PathBuf}, sync::{Arc, atomic::Ordering::*}, - thread, - time::{Duration, Instant}, + thread::{self, JoinHandle}, }; pub use crate::error::{LedgerError, LedgerRequestError}; use derive_more::Deref; +use fjall::Keyspace; use nucleus::{ Slot, ledger::BlockstorePosition, - shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + shutdown::{Service, ShutdownManager, ShutdownReason}, }; use parking_lot::RwLock; -use tokio::{sync::broadcast, time}; +use tokio::sync::broadcast; use tracing::info; mod appender; @@ -35,7 +35,7 @@ mod tests; use crate::{ appender::{BLOCKSTORE_DB, EXECUTIONS_DB, LedgerAppender, SUPERBLOCK_META}, error::Result, - index::IndexSlot, + index::Index, reader::LedgerReader, request::ReaderSender, schema::Event, @@ -44,10 +44,6 @@ 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. /// @@ -58,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. @@ -65,8 +63,7 @@ pub struct Ledger { } impl Ledger { - /// Opens ledger files and starts one appender plus the reader worker pool. - /// The appender opens the active index and reports recovery failures through shutdown. + /// Opens ledger files and the global index, then starts the appender and reader pool. pub fn init( directory: impl AsRef, size_limit: u64, @@ -75,9 +72,6 @@ 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 mut sh = shutdown.handle(Service::LedgerAppender); @@ -152,6 +146,7 @@ impl Ledger { /// 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 @@ -159,9 +154,8 @@ impl Ledger { let meta = unsafe { MetaMap::::new(&meta) }?; let retained = meta.superblocks(); let mut superblocks = BTreeMap::new(); - let head = meta.head(); for id in retained { - let superblock = Superblock::open(&directory, id, id == head)?; + let superblock = Superblock::open(&directory, id, &index)?; superblocks.insert(id, superblock); } @@ -169,6 +163,7 @@ impl Ledger { Ok(Self { meta, superblocks: superblocks.into(), + index, directory, size_limit, }) @@ -188,67 +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); + /// 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(()); + return Ok(None); }; if id >= self.meta.head() { - return Ok(()); + return Ok(None); } let Some((_, superblock)) = superblocks.pop_first() else { - return Ok(()); + return Ok(None); }; + drop(superblocks); + let timer = metrics::time(metrics::Operation::Truncate); let end = superblock.meta.range.end.load(Acquire); self.meta.superblocks.fetch_sub(1, Release); self.meta.range.start.store(end + 1, Release); self.meta.flush()?; - 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(); - }); + 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(()) + }))) } } @@ -297,16 +261,21 @@ pub struct Superblock { pub blockstore: File, /// Transaction execution metadata file. executions: File, - /// Lazily opened Fjall index for this superblock. - index: IndexSlot, + /// 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) } @@ -321,9 +290,9 @@ impl Superblock { } /// Opens a superblock directory, creating its data files when needed. - fn open(root: &Path, id: u64, active: bool) -> Result> { + fn open(root: &Path, id: u64, index: &Index) -> Result> { let directory = Self::init_dir(root, id)?; - let index = IndexSlot::new(&directory, active); + 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/reader.rs b/ledger/src/reader.rs index c177ca08..d3245358 100644 --- a/ledger/src/reader.rs +++ b/ledger/src/reader.rs @@ -112,7 +112,7 @@ impl LedgerReader { if request.cancelled() { return Ok(None); } - let index = superblock.index.get()?.reader(); + let index = IndexReader::new(&superblock.index); let Some(spans) = index.transaction(&request.params)? else { continue; }; @@ -133,7 +133,7 @@ impl LedgerReader { if request.cancelled() { return Ok(None); } - let index = superblock.index.get()?.reader(); + let index = IndexReader::new(&superblock.index); let Some(spans) = index.transaction(&request.params)? else { continue; }; @@ -161,7 +161,7 @@ impl LedgerReader { if request.cancelled() { return Ok(signatures); } - let index = superblock.index.get()?.reader(); + 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. @@ -219,7 +219,7 @@ impl LedgerReader { if !superblock.meta.range.contains(&slot) { continue; } - let index = superblock.index.get()?.reader(); + let index = IndexReader::new(&superblock.index); let Some(span) = index.block(slot)? else { return Ok(None); }; @@ -243,7 +243,7 @@ impl LedgerReader { if start > end { continue; } - let index = superblock.index.get()?.reader(); + 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)? { @@ -338,7 +338,7 @@ impl LedgerReader { if matches!(details, BlockDetails::None) { return Ok(BlockResponse::Bare(block)); } - let index = superblock.index.get()?.reader(); + 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(); @@ -406,7 +406,7 @@ impl LedgerReader { fn blocktime( &mut self, superblock: &Superblock, - index: &IndexReader, + index: &IndexReader<'_>, slot: Slot, ) -> Result { let Some(span) = index.block(slot)? else { diff --git a/ledger/src/tests/index.rs b/ledger/src/tests/index.rs index 518306a9..8cf800bf 100644 --- a/ledger/src/tests/index.rs +++ b/ledger/src/tests/index.rs @@ -1,7 +1,6 @@ //! Index unit tests. -use std::sync::Arc; - +use fjall::Keyspace; use nucleus::{ Slot, testkit::{TempDir, init_tracing, tempdir}, @@ -10,23 +9,22 @@ use solana_pubkey::Pubkey; use solana_signature::Signature; use crate::{ - index::{Index, Span, TxSpan}, + index::{Index, IndexReader, Span, TxSpan}, storage::Durability, }; /// Opens a fresh index on a throwaway directory kept alive by the returned guard. -fn index() -> (TempDir, Arc) { +fn index() -> (TempDir, Index, Keyspace) { init_tracing(); let dir = tempdir(); - let index = Arc::new(Index::new(dir.path(), 2).unwrap()); - (dir, index) + let index = Index::new(dir.path()).unwrap(); + let keyspace = index.keyspace(1).unwrap(); + (dir, index, keyspace) } /// Drains every execution span the account index holds for `pubkey`. -fn account_spans(index: &Arc, pubkey: &Pubkey) -> Vec { - index - .clone() - .reader() +fn account_spans(keyspace: &Keyspace, pubkey: &Pubkey) -> Vec { + IndexReader::new(keyspace) .accounts(pubkey, None) .map(|entry| entry.unwrap()) .collect() @@ -53,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), @@ -63,13 +61,13 @@ fn transaction_and_block_roundtrip() { let block_a = Span::new(0, 8); let block_b = Span::new(8, 16); - let mut writer = index.clone().writer(); + 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 reader = index.clone().reader(); + 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); @@ -88,13 +86,13 @@ fn transaction_and_block_roundtrip() { #[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 writer = index.clone().writer(); + let mut writer = index.writer(&keyspace); for span in &spans { writer.insert_accounts(&[account], *span); } @@ -103,14 +101,14 @@ fn account_signature_duplicates() { // 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 = index.clone().reader(); + let reader = IndexReader::new(&keyspace); assert_eq!( reader .accounts(&account, Some(spans[2])) @@ -121,24 +119,19 @@ fn account_signature_duplicates() { ); } -/// Proves sealed indexes open on demand and cannot close while a read lease exists. +/// Proves deleting one superblock keyspace preserves in-flight reads and isolates reuse. #[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()); +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)); - // Reopening after a full close proves the exclusive Fjall lock was released. - drop(superblock.index.get().unwrap()); - assert!(superblock.index.evict()); + 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 84e228ac..e304992f 100644 --- a/ledger/src/tests/integration.rs +++ b/ledger/src/tests/integration.rs @@ -17,7 +17,6 @@ use std::{ ops::Range, process::Command, sync::{Arc, atomic::Ordering::Acquire}, - time::Duration, }; use nucleus::{ @@ -29,10 +28,7 @@ use nucleus::{ use solana_pubkey::Pubkey; use solana_signature::Signature; use solana_transaction_error::TransactionResult; -use tokio::{ - sync::{broadcast, mpsc}, - time, -}; +use tokio::sync::{broadcast, mpsc}; use crate::{ Ledger, @@ -375,8 +371,8 @@ async fn test_block_detail_levels_partition_transactions() { } // A sealed superblock stays readable after the writer rotates to a new segment, -// and retention evicts the oldest sealed superblock before removing its files -// in the background while preserving the active head and 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. @@ -404,13 +400,10 @@ async fn test_superblock_rotation_and_retention() { 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"); + 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/src/shutdown.rs b/nucleus/src/shutdown.rs index ec4a7795..3dd2396d 100644 --- a/nucleus/src/shutdown.rs +++ b/nucleus/src/shutdown.rs @@ -31,8 +31,6 @@ pub enum Service { LedgerAppender, /// Ledger read worker. LedgerReader, - /// Ledger sealed-index cache maintenance task. - LedgerMaintenance, /// Ledger replay worker. LedgerReplayer, /// Transaction scheduler service. 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" From 4869aad23eb650f02c9dac71078d0aaffd00904c Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Thu, 20 Aug 2026 15:08:57 +0400 Subject: [PATCH 4/6] feat: rotate memtables on restart --- ledger/README.md | 6 ++++-- ledger/src/appender.rs | 1 + ledger/src/index.rs | 9 +++++++++ 3 files changed, 14 insertions(+), 2 deletions(-) diff --git a/ledger/README.md b/ledger/README.md index 296d08eb..04b65658 100644 --- a/ledger/README.md +++ b/ledger/README.md @@ -33,7 +33,7 @@ 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 ledger-wide Fjall index uses two -background workers and one 8 MiB cache across all superblock keyspaces. The +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 @@ -45,7 +45,9 @@ 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. +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 diff --git a/ledger/src/appender.rs b/ledger/src/appender.rs index f4a2fb76..90cd2eba 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -212,6 +212,7 @@ impl LedgerAppender { self.writer.write_blockstore(&BlockstoreEntry::Superblock(seal))?; self.sync(None)?; self.writer.finalize()?; + self.index.rotate_memtable()?; info!(superblock = seal.id, "sealed superblock"); Ok(()) } diff --git a/ledger/src/index.rs b/ledger/src/index.rs index 3bd0520c..77f86cca 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -42,6 +42,8 @@ const ACCOUNT_PREFIX_BYTES: usize = 1 + PREFIX_BYTES; const ACCOUNT_KEY_BYTES: usize = ACCOUNT_PREFIX_BYTES + SPAN_BYTES; /// Ledger-wide block cache capacity. const CACHE_SIZE: u64 = 64 * MB as u64; +/// Retained journal bytes before Fjall flushes keyspaces blocking reclamation. +const JOURNAL_SIZE: u64 = 256 * MB as u64; /// Fjall maintenance workers assigned to the ledger index. const WORKERS: usize = 2; @@ -181,6 +183,7 @@ impl 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) @@ -292,6 +295,12 @@ impl IndexWriter { }; batch.durability(Some(mode)).commit().map_err(Into::into) } + + /// Queues the immutable keyspace's active memtable for background flushing. + pub(crate) fn rotate_memtable(&self) -> Result<()> { + self.keyspace.rotate_memtable()?; + Ok(()) + } } /// Returns the namespaced transaction key. From ae2c76459ec0185bb57d5ebb93b6746997066291 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Thu, 20 Aug 2026 17:42:36 +0400 Subject: [PATCH 5/6] docs(ledger): clarify Fjall index comments --- ledger/src/index.rs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/ledger/src/index.rs b/ledger/src/index.rs index 77f86cca..0b5058f3 100644 --- a/ledger/src/index.rs +++ b/ledger/src/index.rs @@ -42,7 +42,7 @@ const ACCOUNT_PREFIX_BYTES: usize = 1 + PREFIX_BYTES; const ACCOUNT_KEY_BYTES: usize = ACCOUNT_PREFIX_BYTES + SPAN_BYTES; /// Ledger-wide block cache capacity. const CACHE_SIZE: u64 = 64 * MB as u64; -/// Retained journal bytes before Fjall flushes keyspaces blocking reclamation. +/// 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; @@ -171,11 +171,6 @@ pub(crate) struct IndexWriter { } /// Read access to one retained superblock's append-only index. -/// -/// Point lookups use Fjall's latest visible sequence number. Range and prefix -/// calls already return iterators carrying their own snapshot-tracker nonce, so -/// retaining a database-wide snapshot here would add tracking work without -/// strengthening these reads. pub(crate) struct IndexReader<'a>(&'a Keyspace); impl Index { From 63f155b186971b3b0ca251f1ca15ce5c0af4b30a Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Thu, 20 Aug 2026 17:59:24 +0400 Subject: [PATCH 6/6] fix(ledger): synchronize appender shutdown --- keeper/src/accessor.rs | 2 +- ledger/src/appender.rs | 7 ++++--- ledger/src/schema.rs | 6 +++--- 3 files changed, 8 insertions(+), 7 deletions(-) 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/src/appender.rs b/ledger/src/appender.rs index 90cd2eba..84e9ff6a 100644 --- a/ledger/src/appender.rs +++ b/ledger/src/appender.rs @@ -102,14 +102,15 @@ impl LedgerAppender { Event::Bootstrap(seal) => self.seal(seal, true)?, Event::Reset(slot) => self.write_reset(slot)?, Event::Sync { response, is_final } => { - let _ = response.send(self.sync(None)); + self.sync(None)?; + let _ = response.send(()); if is_final { - break; + return Ok(()); } } } } - Ok(()) + self.sync(None) } /// Rotates to the next superblock directory. 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, },