diff --git a/Cargo.toml b/Cargo.toml index b793825..2a3b251 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "accountsdb", + "ledger", "nucleus", "programs/magic-root-interface", "programs/magic-root-program", @@ -23,7 +24,8 @@ rust-version = "1.94.1" version = "0.1.0" [workspace.dependencies] -accountsdb = { path = "accountsdb" } +accountsdb = { path = "accountsdb", package = "magicblock-accountsdb" } +ledger = { path = "ledger", package = "magicblock-ledger" } magic-root-interface = { path = "programs/magic-root-interface" } magic-root-program = { path = "programs/magic-root-program" } nucleus = { path = "nucleus", package = "magicblock-engine-nucleus" } @@ -46,6 +48,7 @@ clonetree = "0.0.2" criterion = "0.7.0" derive_more = "2.1.1" env_logger = "0.11.8" +flume = { version = "0.12" } futures = { version = "0.3.32", default-features = false } heed = { version = "0.22.1", default-features = false } itertools = "0.14.0" diff --git a/ledger/Cargo.toml b/ledger/Cargo.toml new file mode 100644 index 0000000..2246e52 --- /dev/null +++ b/ledger/Cargo.toml @@ -0,0 +1,45 @@ +[package] +name = "magicblock-ledger" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "ledger" + +[dependencies] +nucleus = { workspace = true, features = ["heed", "ledger-schema", "metrics", "shutdown"] } + +bitcode = { workspace = true } +bytemuck = { workspace = true } +derive_more = { workspace = true, features = ["deref", "from"] } +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"] } +tracing = { workspace = true } +wincode = { workspace = true } +zstd = { workspace = true } + +agave-transaction-view = { workspace = true, features = ["agave-unstable-api"] } +solana-hash = { workspace = true } +solana-pubkey = { workspace = true } +solana-signature = { workspace = true, features = ["wincode"] } +solana-transaction-error = { workspace = true, features = ["wincode"] } + +[dev-dependencies] +nucleus = { workspace = true, features = ["testkit"] } +solana-hash = { workspace = true } + +[lints] +workspace = true diff --git a/ledger/README.md b/ledger/README.md new file mode 100644 index 0000000..d451ca8 --- /dev/null +++ b/ledger/README.md @@ -0,0 +1,44 @@ +# `magicblock-ledger` + +The ledger is the engine's durable record of what executed: transactions, block +boundaries, and the execution metadata that goes with them. The design problem it +solves is retention. A ledger grows without bound, and at some point old history +has to be dropped or archived — cheaply, and without disturbing the part still +being written. + +The answer is the **superblock**. Each superblock is a self-contained directory: +its own data files, its own LMDB index, everything it needs to be read in +isolation. Because nothing is shared across superblocks, retention and archival +become a filesystem operation — drop or move a whole directory — instead of a +delicate compaction over a single monolithic store. + +```text +ledger.meta +superblock-000000001/ + superblock.meta + blockstore.db # wincode stream of txns, block boundaries, superblock seals + executions.db # fixed execution header + compressed bitcode payload + index/ +``` + +## Reads and writes are split + +A **single writer** owns the ordered append path, and the ordering rules matter: +a transaction stays *pending* until its execution metadata arrives (so a record is +never half-written); a block boundary syncs the data files, commits the indexes, +publishes the durable cursors, and only then updates `ledger.meta`; a superblock +seal rotates to the next segment. That strict sequence is what lets a reader trust +any published cursor. + +Reads are served by a **pool of reader workers** through `LedgerHandle::reader`. +Each worker keeps its own decode buffers, so no mutable decoder state is ever +shared across threads — reads scale out without contending on the writer or each +other. + +## Retention caveat + +`Ledger::truncate` drops the oldest *sealed* superblock when the ledger +filesystem hits its size limit; the active head is never removed. One thing to +know: the size check assumes the ledger owns its filesystem outright. Any +unrelated files sharing that filesystem count toward "used space" and will make +truncation trigger earlier than you'd expect. diff --git a/ledger/src/appender.rs b/ledger/src/appender.rs new file mode 100644 index 0000000..cf783a6 --- /dev/null +++ b/ledger/src/appender.rs @@ -0,0 +1,301 @@ +//! Ledger append service and writable superblock storage. + +use std::{ + collections::HashMap, + fs, + path::Path, + sync::{Arc, atomic::Ordering::*}, +}; + +use agave_transaction_view::transaction_view::TransactionView; +use bitcode::Buffer; +use flume::Receiver; +use nucleus::{ + heed::{DatabaseIndex, OptRwTxn}, + shutdown::{ShutdownHandle, ShutdownReason}, +}; +use solana_signature::Signature; +use tracing::{debug, info, warn}; +use wincode::Error; +use wincode::io::std_write::WriteAdapter; +use zstd::bulk::Compressor; + +use crate::{ + Ledger, Superblock, + error::{LedgerError, Result}, + index::{Index, Span, TxSpan}, + metrics::{self, Operation}, + schema::{ + Block, BlockstoreEntry, Event, Execution, ExecutionDetails, SuperblockSeal, + TransactionEntry, + }, + storage::{AppendFile, MetaMap, SuperblockMeta}, +}; + +/// Blockstore stream file name inside a superblock. +pub(crate) const BLOCKSTORE_DB: &str = "blockstore.db"; +/// Execution details file name inside a superblock. +pub(crate) const EXECUTIONS_DB: &str = "executions.db"; +/// Superblock metadata file name. +pub(crate) const SUPERBLOCK_META: &str = "superblock.meta"; + +/// Background service that appends ledger events into the active superblock. +pub(crate) struct LedgerAppender { + /// Shared top-level ledger state. + ledger: Arc, + /// Writable files for the active superblock. + writer: SuperblockWriter, + /// Transactions waiting for their matching execution details. + pending: HashMap, + /// Active superblock index. + index: Arc, + /// Event stream from the execution pipeline. + rx: Receiver, + /// Transactions written since the last committed block boundary. + transactions: u64, +} + +impl LedgerAppender { + /// Opens the active superblock and registers it on the ledger handle. + pub(crate) fn new(ledger: Arc, rx: Receiver) -> Result { + let head = ledger.meta.head(); + let directory = Superblock::path(&ledger.directory, head); + let writer = SuperblockWriter::new(&directory)?; + metrics::pending_transactions(0); + let index = ledger + .superblocks + .read() + .get(&head) + .map(|s| s.index.clone()) + .ok_or(LedgerError::Corruption("active superblock missing"))?; + + Ok(Self { + ledger, + writer, + index, + rx, + pending: HashMap::new(), + transactions: 0, + }) + } + + /// Runs until shutdown (when the event stream closes). + pub(crate) fn run(mut self, mut shutdown: ShutdownHandle) { + let mut txn = None; + while let Ok(event) = self.rx.recv() { + let _timer = metrics::time(Operation::Append); + match self.process(event, &mut txn) { + Ok(true) => match self.rotate() { + Ok(appender) => self = appender, + Err(err) => { + shutdown.terminate(ShutdownReason::Error(Box::new(err))); + return; + } + }, + Ok(false) => (), + Err(err) => { + shutdown.terminate(ShutdownReason::Error(Box::new(err))); + return; + } + } + } + shutdown.terminate(ShutdownReason::Signalled); + } + + /// Rotates to the next superblock directory. + fn rotate(self) -> Result { + let _timer = metrics::time(Operation::Rotate); + let head = self.ledger.meta.advance(); + info!(superblock = head, "opened active superblock"); + let superblock = Superblock::open(&self.ledger.directory, head)?; + self.ledger.superblocks.write().insert(head, superblock); + self.ledger.meta.superblocks.fetch_add(1, Release); + self.ledger.meta.flush()?; + Self::new(self.ledger, self.rx) + } + + /// Processes one append event and returns true when the superblocks should rotate. + fn process(&mut self, event: Event, txn: OptRwTxn<'_, '_>) -> Result { + match event { + Event::Transaction(transaction) => self.write_transaction(transaction)?, + Event::Execution(execution) => self.write_execution(execution, txn)?, + Event::Block(block) => self.write_block(block, txn)?, + Event::Superblock(seal) => { + self.write_superblock(seal)?; + return Ok(true); + } + } + Ok(false) + } + + /// Writes a raw transaction and keeps it pending until execution arrives. + fn write_transaction(&mut self, transaction: TransactionEntry) -> Result<()> { + let entry = BlockstoreEntry::Transaction(transaction.payload.as_slice()); + let span = self.writer.write_blockstore(&entry)?; + let entry = PendingTx { + transaction: transaction.payload, + span, + }; + self.pending.insert(transaction.signature, entry); + metrics::pending_transactions(self.pending.len()); + self.transactions += 1; + Ok(()) + } + + /// Writes execution details and adds transaction/account indexes. + fn write_execution(&mut self, execution: Execution, txn: OptRwTxn<'_, '_>) -> 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"); + return Ok(()); + }; + metrics::pending_transactions(self.pending.len()); + let execution = self.writer.write_execution(&execution)?; + let span = TxSpan { + blockstore: pending.span, + execution, + }; + self.index.insert_transaction(txn, &signature, &span)?; + let view = TransactionView::try_new_unsanitized(pending.transaction)?; + let accounts = view.static_account_keys(); + self.index.insert_accounts(txn, accounts, &span.execution)?; + Ok(()) + } + + /// Writes a block boundary and publishes it after data and indexes are durable. + fn write_block(&mut self, block: Block, txn: OptRwTxn<'_, '_>) -> Result<()> { + let span = self.writer.write_blockstore(&BlockstoreEntry::Block(block))?; + self.index.insert_block(txn, &block.slot, &span)?; + let cursors = self.writer.sync()?; + if let Some(txn) = txn.take() { + txn.commit()?; + } + self.index.flush()?; + self.writer.publish(cursors, Some(block.slot))?; + self.ledger.meta.transactions.fetch_add(self.transactions, Release); + self.ledger.meta.blocks.fetch_add(1, Release); + self.ledger.meta.range.end.store(block.slot, Release); + self.ledger.meta.flush()?; + if self.ledger.size_limit_reached()? { + self.ledger.truncate()?; + } + metrics::ledger_counts(&self.ledger); + self.transactions = 0; + Ok(()) + } + + /// 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))?; + let cursors = self.writer.sync()?; + self.writer.publish(cursors, None)?; + self.writer.finalize()?; + info!(superblock = self.ledger.meta.head(), "sealed superblock"); + Ok(()) + } +} + +/// Transaction bytes already written but not yet paired with execution details. +struct PendingTx { + /// Transaction bytes retained until execution details arrive for indexing. + transaction: Arc>, + /// Blockstore-file span of the transaction bytes. + span: Span, +} + +/// Writable superblock files and reusable execution-detail encoder. +struct SuperblockWriter { + /// Compressor reused for execution metadata payloads. + compressor: Compressor<'static>, + /// Scratch buffer owned by bitcode while encoding metadata. + buffer: Buffer, + /// Buffered transaction/block/seal blockstore stream. + 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 { + fs::create_dir_all(directory)?; + let metadata = unsafe { MetaMap::::new(&directory.join(SUPERBLOCK_META)) }?; + + Ok(Self { + blockstore: AppendFile::new( + &directory.join(BLOCKSTORE_DB), + &metadata.cursors.blockstore, + )?, + executions: AppendFile::new( + &directory.join(EXECUTIONS_DB), + &metadata.cursors.executions, + )?, + metadata, + compressor: Compressor::new(0)?, + buffer: Buffer::new(), + }) + } + + /// Appends an entry to the blockstore file. + fn write_blockstore(&mut self, entry: &BlockstoreEntry<&[u8]>) -> Result { + let offset = self.blockstore.cursor; + wincode::serialize_into(WriteAdapter::new(&mut self.blockstore), entry) + .map_err(Into::::into)?; + let size = self.blockstore.cursor - offset; + Ok(Span::new(offset, size)) + } + + /// Appends execution details and returns their file span. + fn write_execution(&mut self, execution: &Execution) -> Result { + let offset = self.executions.cursor; + let writer = WriteAdapter::new(&mut self.executions); + wincode::serialize_into(writer, &execution.header).map_err(Into::::into)?; + let mut details = self.buffer.encode(&execution.details); + if details.len() >= crate::MAX_ENTRY_SIZE { + details = self.buffer.encode(&None::); + } + self.executions.compress(details, &mut self.compressor)?; + let size = self.executions.cursor - offset; + 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()?)) + } + + /// 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); + if let Some(slot) = slot { + self.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); + } + self.metadata.flush()?; + Ok(()) + } + + /// Trims preallocated file space after the superblock cursors are durable. + fn finalize(&mut self) -> Result<()> { + let _timer = metrics::time(Operation::FileFinalize); + self.blockstore.finalize()?; + self.executions.finalize()?; + debug!( + blockstore = self.blockstore.cursor, + executions = self.executions.cursor, + "trimmed sealed superblock files" + ); + Ok(()) + } +} + +// SAFETY: `LedgerAppender` is moved into one background thread. Its LMDB +// write transaction is created and consumed on that same thread. +unsafe impl Send for LedgerAppender {} diff --git a/ledger/src/error.rs b/ledger/src/error.rs new file mode 100644 index 0000000..4c89938 --- /dev/null +++ b/ledger/src/error.rs @@ -0,0 +1,54 @@ +//! Error types shared by the ledger crate. + +use std::io; + +use agave_transaction_view::result::TransactionViewError; +use heed::BoxedError; +use oneshot::RecvError; +use tokio::time::error::Elapsed; + +/// Errors returned by ledger storage, codecs, and indexes. +#[derive(Debug, derive_more::From, thiserror::Error)] +pub enum LedgerError { + /// Filesystem I/O failed while opening, writing, or flushing ledger files. + #[error("ledger storage I/O error: {0}")] + IO(#[source] io::Error), + /// Platform filesystem operation failed. + #[error("ledger filesystem operation failed: {0}")] + FS(#[source] rustix::io::Errno), + /// Wincode failed while serializing the blockstore stream. + #[error("ledger blockstore codec error: {0}")] + Wincode(#[source] wincode::Error), + /// 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. + #[error("ledger index error: {0}")] + Index(#[source] heed::Error), + /// Sanitized transaction view could not be decoded. + #[error("transaction view error: {0:?}")] + TransactionView(TransactionViewError), + /// Ledger index pointed to an entry of the wrong type or invalid shape. + #[error("ledger corruption: {0}")] + #[from(skip)] + Corruption(&'static str), +} + +/// Errors returned while waiting for a ledger reader response. +#[derive(Debug, thiserror::Error)] +pub enum LedgerRequestError { + /// Reader dropped the response channel before sending a result. + #[error("ledger reader response channel closed: {0}")] + ResponseFailed(#[from] RecvError), + /// Reader did not answer within the request timeout. + #[error("ledger reader request timed out: {0}")] + Timeout(#[from] Elapsed), +} + +/// Result type used by the ledger crate. +pub(crate) type Result = std::result::Result; +/// Result type used while waiting for asynchronous ledger reader responses. +pub(crate) type RequestResult = std::result::Result; diff --git a/ledger/src/index.rs b/ledger/src/index.rs new file mode 100644 index 0000000..93bf36e --- /dev/null +++ b/ledger/src/index.rs @@ -0,0 +1,317 @@ +//! 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, byteorder::LittleEndian, + iteration_method::MoveOnCurrentKeyDuplicates, types::U64, +}; +use nucleus::{ + GB, Slot, + heed::{DatabaseIndex, OptRoTxn, OptRwTxn}, +}; +use solana_pubkey::Pubkey; +use solana_signature::Signature; + +use crate::schema::Offset; + +/// Index directory below each superblock directory. +const INDEX_SUBDIR: &str = "index"; +/// Maximum LMDB map size for ledger indexes. +const INDEX_MAP_SIZE: usize = 16 * GB; +/// Number of LMDB named databases in the ledger index. +const INDEX_DBS: u32 = 4; +/// 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"; +/// 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. +/// +/// See `SignatureKey`; account history uses the same compact prefix tag. +#[derive(Pod, Zeroable, Clone, Copy)] +#[repr(C)] +pub(crate) struct AccountKey([u8; KEY_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)] +pub(crate) struct Span(u64); + +impl Span { + /// Number of low bits reserved for entry size. + const SIZE_BITS: u64 = 25; + /// Mask for the encoded entry size. + const SIZE_MASK: u64 = (1 << Self::SIZE_BITS) - 1; + /// Largest blockstore entry size that can be packed into an index value. + pub(crate) const MAX_SIZE: u64 = Self::SIZE_MASK; + + /// Packs an `offset` and `size` into one integer. + pub(crate) fn new(offset: Offset, size: u64) -> Self { + debug_assert!(offset <= u64::MAX >> Self::SIZE_BITS); + debug_assert!(size <= Self::MAX_SIZE); + Self(offset << Self::SIZE_BITS | size) + } + + /// Returns the byte offset in the backing file. + pub(crate) fn offset(&self) -> u64 { + self.0 >> Self::SIZE_BITS + } + + /// Returns the encoded entry size in bytes. + pub(crate) fn size(&self) -> u64 { + self.0 & Self::SIZE_MASK + } +} + +/// Pair of blockstore-file and execution-file spans for a transaction. +#[derive(Zeroable, Pod, Clone, Copy)] +#[repr(C)] +pub(crate) struct TxSpan { + /// Span of the raw transaction entry in the blockstore. + pub(crate) blockstore: Span, + /// Span of the execution details. + pub(crate) execution: Span, +} + +/// Account-index span codec. +pub(crate) enum AccountSpan {} + +/// Duplicate account-index iterator over execution spans. +pub(crate) type AccountIter<'a> = RoIter<'a, AccountKey, AccountSpan, MoveOnCurrentKeyDuplicates>; + +/// LMDB databases used to locate ledger data. +pub(crate) struct Index { + /// Owning LMDB environment. + env: Env, + /// Transaction signature to transaction/execution spans. + transactions: Database, + /// Slot to blockstore-file span. + blocks: Database, + /// Account key to execution-details span. + accounts: Database, +} + +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)? + }; + + let mut txn = env.write_txn()?; + let transactions = + env.database_options().name(TRANSACTIONS_INDEX).types().create(&mut txn)?; + let blocks = env + .database_options() + .name(SLOTS_INDEX) + .key_comparator() + .types() + .create(&mut txn)?; + let accounts = env + .database_options() + .name(ACCOUNTS_INDEX) + .flags(DatabaseFlags::DUP_SORT | DatabaseFlags::DUP_FIXED | DatabaseFlags::REVERSE_DUP) + .types() + .create(&mut txn)?; + txn.commit()?; + Ok(Self { + env, + transactions, + blocks, + accounts, + }) + } + + /// Indexes a block boundary by slot. + pub(crate) fn insert_block( + &self, + txn: OptRwTxn<'_, '_>, + slot: &Slot, + span: &Span, + ) -> heed::Result<()> { + let txn = DatabaseIndex::write_txn(self, txn)?; + self.blocks.put(txn, slot, span) + } + + /// Indexes a transaction and its execution details. + pub(crate) fn insert_transaction( + &self, + txn: OptRwTxn<'_, '_>, + signature: &Signature, + span: &TxSpan, + ) -> heed::Result<()> { + let txn = DatabaseIndex::write_txn(self, txn)?; + self.transactions.put(txn, signature, span) + } + + /// Adds account-to-execution entries for all static transaction accounts. + pub(crate) fn insert_accounts( + &self, + txn: OptRwTxn<'_, '_>, + accounts: &[Pubkey], + span: &Span, + ) -> heed::Result<()> { + let txn = DatabaseIndex::write_txn(self, txn)?; + for account in accounts { + self.accounts.put(txn, account, span)?; + } + Ok(()) + } + + /// Locates a transaction by its first signature. + pub(crate) fn transaction( + &self, + signature: &Signature, + txn: OptRoTxn<'_, '_>, + ) -> heed::Result> { + let txn = DatabaseIndex::read_txn(self, txn)?; + self.transactions.get(txn, signature) + } + + /// Locates a block boundary by slot. + pub(crate) fn block(&self, slot: &Slot, txn: OptRoTxn<'_, '_>) -> heed::Result> { + let txn = DatabaseIndex::read_txn(self, txn)?; + self.blocks.get(txn, slot) + } + + /// Returns execution spans that mention `pubkey`. + pub(crate) fn accounts<'t, 'e>( + &'e self, + pubkey: &Pubkey, + txn: OptRoTxn<'t, 'e>, + ) -> heed::Result>> { + let txn = DatabaseIndex::read_txn(self, txn)?; + self.accounts.get_duplicates(txn, pubkey) + } +} + +unsafe impl DatabaseIndex for Index { + fn env(&self) -> &Env { + &self.env + } +} + +impl> From for SignatureKey { + fn from(signature: S) -> Self { + Self(array::from_fn(|i| signature.as_ref().as_array()[i])) + } +} + +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<'a> BytesEncode<'a> for AccountKey { + type EItem = Pubkey; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + Ok(item.as_array()[..KEY_BYTES].into()) + } +} + +impl<'a> BytesDecode<'a> for SignatureKey { + type DItem = &'a Self; + + fn bytes_decode(bytes: &'a [u8]) -> CodecResult { + bytemuck::try_from_bytes(bytes).map_err(Into::into) + } +} + +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) + } +} + +impl<'a> BytesEncode<'a> for Span { + type EItem = Self; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + U64Le::bytes_encode(&item.0) + } +} + +impl<'a> BytesDecode<'a> for Span { + type DItem = Self; + + fn bytes_decode(bytes: &'a [u8]) -> CodecResult { + U64Le::bytes_decode(bytes).map(Self) + } +} + +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()) + } +} + +impl<'a> BytesDecode<'a> for AccountSpan { + type DItem = Span; + + fn bytes_decode(bytes: &'a [u8]) -> CodecResult { + U64Le::bytes_decode(bytes).map(|span| Span(!span)) + } +} + +impl<'a> BytesEncode<'a> for TxSpan { + type EItem = Self; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + Ok(bytemuck::bytes_of(item).into()) + } +} + +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) + } +} diff --git a/ledger/src/lib.rs b/ledger/src/lib.rs new file mode 100644 index 0000000..686b1b7 --- /dev/null +++ b/ledger/src/lib.rs @@ -0,0 +1,251 @@ +#![doc = include_str!("../README.md")] + +use std::{ + collections::BTreeMap, + fs::{self, File}, + path::{Path, PathBuf}, + sync::{Arc, atomic::Ordering::*}, + thread, +}; + +pub use crate::error::{LedgerError, LedgerRequestError}; +use derive_more::Deref; +use nucleus::{ + Slot, + shutdown::{Service, ShutdownManager}, +}; +use parking_lot::RwLock; +use tracing::info; + +mod appender; +mod error; +mod index; +mod metrics; +mod reader; +pub mod request; +pub mod schema; +mod storage; + +#[cfg(test)] +mod tests; + +use crate::{ + appender::{BLOCKSTORE_DB, EXECUTIONS_DB, LedgerAppender, SUPERBLOCK_META}, + error::Result, + index::{Index, Span}, + reader::LedgerReader, + request::ReaderSender, + schema::Event, + storage::{LedgerMeta, MetaMap, SuperblockMeta}, +}; + +/// Max allowed size for any ledger entry +pub const MAX_ENTRY_SIZE: usize = Span::MAX_SIZE as usize; +const LEDGER_META: &str = "ledger.meta"; +const SERVICE_QUEUE_CAPACITY: usize = 128; + +/// Top-level ledger handle. +/// +/// `head` in ledger metadata is the active superblock id. Older superblocks +/// stay open in `superblocks` until retention removes their directories. +pub struct Ledger { + /// Mmap-backed ledger metadata shared with the appender. + meta: MetaMap, + /// Retained superblocks keyed by id. + superblocks: RwLock>>, + /// Root ledger directory. + pub directory: PathBuf, + /// Maximum used bytes allowed on the ledger filesystem before retention runs. + size_limit: u64, +} + +impl Ledger { + /// Opens the ledger and starts one appender plus the reader worker pool. + pub fn init( + directory: impl AsRef, + size_limit: u64, + shutdown: &mut ShutdownManager, + ) -> Result { + let directory = directory.as_ref().to_owned(); + let ledger = Arc::new(Self::new(directory, size_limit)?); + metrics::init(&ledger); + let (appender_tx, rx) = flume::bounded(SERVICE_QUEUE_CAPACITY); + let appender = LedgerAppender::new(ledger.clone(), rx)?; + let sh = shutdown.handle(Service::LedgerAppender); + thread::Builder::new() + .name("ledger-appender".into()) + .spawn(|| appender.run(sh))?; + let (reader_tx, rx) = flume::bounded(SERVICE_QUEUE_CAPACITY); + for id in 0..num_cpus::get() as u32 { + let sh = shutdown.handle(Service::LedgerReader); + let reader = LedgerReader::new(ledger.clone(), rx.clone())?; + thread::Builder::new() + .name(format!("ledger-reader-{id}")) + .spawn(|| reader.run(sh))?; + } + Ok(LedgerHandle { + ledger, + reader: reader_tx, + appender: appender_tx, + }) + } + + /// Iterates retained superblocks from newest to oldest. + pub fn iter(&self) -> impl Iterator> { + let range = self.meta.superblocks(); + range.rev().filter_map(|id| self.superblocks.read().get(&id).cloned()) + } + + /// Returns the highest slot recorded across retained superblocks, or + /// `None` when none are retained. Taken over all superblocks rather than + /// just the head: right after a seal the active head has no blocks yet and + /// its persisted range is still zeroed. + pub fn tip(&self) -> Option { + self.iter().map(|s| s.meta.range.end.load(Acquire)).max() + } + + /// Iterates retained superblocks from `superblock` through the active head + /// (inclusive), so replay rebuilds the unsealed head's entries as well. + fn iter_from(&self, superblock: u64) -> impl Iterator> { + let range = superblock..=self.meta.head(); + range.filter_map(|id| self.superblocks.read().get(&id).cloned()) + } + + /// Opens ledger metadata and retained superblocks without starting services. + fn new(directory: PathBuf, size_limit: u64) -> Result { + fs::create_dir_all(&directory)?; + let meta = directory.join(LEDGER_META); + let meta = unsafe { MetaMap::::new(&meta) }?; + let mut superblocks = BTreeMap::new(); + for id in meta.superblocks() { + let superblock = Superblock::open(&directory, id)?; + superblocks.insert(id, superblock); + } + + info!(?directory, superblocks = superblocks.len(), "opened ledger"); + Ok(Self { + meta, + superblocks: superblocks.into(), + directory, + size_limit, + }) + } + + /// Returns true when the ledger filesystem has reached the configured limit. + /// + /// This assumes `directory` is on a filesystem dedicated to the ledger. Any + /// unrelated files on the same filesystem count toward the used byte total. + fn size_limit_reached(&self) -> Result { + let st = rustix::fs::statvfs(&self.directory)?; + let total = st.f_blocks * st.f_frsize; + let free = st.f_bfree * st.f_frsize; + Ok(total.saturating_sub(free) >= self.size_limit) + } + + /// 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. + fn truncate(&self) -> Result<()> { + let _timer = metrics::time(metrics::Operation::Truncate); + let Some((id, superblock)) = self + .superblocks + .read() + .first_key_value() + .map(|(id, superblock)| (*id, superblock.clone())) + else { + return Ok(()); + }; + if id >= self.meta.head() { + return Ok(()); + } + + let end = superblock.meta.range.end.load(Acquire); + superblock.purge()?; + self.superblocks.write().remove(&id); + self.meta.superblocks.fetch_sub(1, Release); + self.meta.range.start.store(end + 1, Release); + self.meta.flush()?; + info!(id, end, "purged oldest superblock for retention"); + Ok(()) + } +} + +/// Senders used by callers to append events and submit read requests. +#[derive(Deref)] +pub struct LedgerHandle { + /// Shared ledger state behind the request senders. + #[deref] + ledger: Arc, + /// Read request queue consumed by reader workers. + pub reader: ReaderSender, + /// Append event queue consumed by the appender worker. + pub appender: flume::Sender, +} + +impl LedgerHandle { + /// Returns the active superblock id. + pub fn head(&self) -> u64 { + self.ledger.meta.head() + } + + /// Returns the number of transactions committed into sealed block metadata. + pub fn transactions(&self) -> u64 { + self.ledger.meta.transactions.load(Acquire) + } +} + +/// Opened superblock files kept alive for active and retained readers. +pub struct Superblock { + id: u64, + /// Mmap-backed metadata for file cursors and slot range. + meta: MetaMap, + /// Transaction stream delimited by block entries and a superblock seal. + blockstore: File, + /// Transaction execution metadata file. + executions: File, + /// LMDB index for this superblock. + index: Arc, + /// Superblock directory path. + pub directory: PathBuf, +} + +impl Superblock { + /// Returns the directory path for `id` under `root`. + pub fn path(root: &Path, id: u64) -> PathBuf { + root.join(format!("superblock-{id:0>9}")) + } + + /// Opens a superblock directory, creating its data files when needed. + fn open(root: &Path, id: u64) -> Result> { + let directory = Self::path(root, id); + fs::create_dir_all(&directory)?; + let index = Arc::new(Index::new(&directory)?); + 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))?; + + Ok(Arc::new(Self { + id, + meta, + blockstore, + executions, + index, + directory, + })) + } + + /// Opens a superblock data file for random reads. + fn file(path: &Path) -> Result { + let f = File::options().create(true).truncate(false).read(true).write(true).open(path)?; + #[cfg(target_os = "linux")] + fs::fadvise(f.as_raw_fd(), 0, None, fs::Advice::RANDOM)?; + + Ok(f) + } + + /// Removes this superblock directory from the ledger root. + fn purge(&self) -> Result<()> { + fs::remove_dir_all(&self.directory).map_err(Into::into) + } +} diff --git a/ledger/src/metrics.rs b/ledger/src/metrics.rs new file mode 100644 index 0000000..cf3981c --- /dev/null +++ b/ledger/src/metrics.rs @@ -0,0 +1,151 @@ +//! Prometheus metrics for ledger. + +use std::sync::{OnceLock, atomic::Ordering::Relaxed}; + +use nucleus::metrics as metric; +use nucleus::metrics::{IntGauge, MetricOperation, MetricSpec, OperationCounters}; + +use crate::Ledger; + +/// Process-wide ledger metrics registered in the default Prometheus registry. +static METRICS: OnceLock = OnceLock::new(); + +/// Operation latency histogram recorded in microseconds. +const OPERATION_TIME: MetricSpec = MetricSpec { + name: "ledger_operation_duration_micros", + help: "Ledger operation duration distribution in microseconds.", +}; +/// Transactions waiting for execution metadata in the appender. +const PENDING_TRANSACTIONS: MetricSpec = MetricSpec { + name: "ledger_pending_transactions", + help: "Current ledger transactions waiting for execution metadata.", +}; +/// Total committed transactions from ledger metadata. +const TRANSACTIONS: MetricSpec = MetricSpec { + name: "ledger_transactions", + help: "Current total transactions committed into ledger metadata.", +}; +/// Total committed blocks from ledger metadata. +const BLOCKS: MetricSpec = MetricSpec { + name: "ledger_blocks", + help: "Current total blocks committed into ledger metadata.", +}; +/// Total superblocks allocated from ledger metadata. +const SUPERBLOCKS: MetricSpec = MetricSpec { + name: "ledger_superblocks", + help: "Current total ledger superblocks allocated from genesis.", +}; + +/// Ledger operation used as a low-cardinality operation label. +#[derive(Clone, Copy)] +pub(crate) enum Operation { + /// Append event processing path. + Append, + /// Transaction read request. + ReadTransaction, + /// Transaction status read request. + ReadTransactionStatus, + /// Account signature history read request. + ReadAccountSignatures, + /// Single-block read request. + ReadBlock, + /// Block range read request. + ReadBlockRange, + /// Retained blockstore replay request. + Replay, + /// Superblock rotation path. + Rotate, + /// Retention truncation path. + Truncate, + /// Data-file sync path. + FileSync, + /// Data-file finalization path. + FileFinalize, +} + +impl MetricOperation for Operation { + /// Returns the Prometheus label value for this operation. + fn label(self) -> &'static str { + match self { + Operation::Append => "append", + Operation::ReadTransaction => "read_transaction", + Operation::ReadTransactionStatus => "read_transaction_status", + Operation::ReadAccountSignatures => "read_account_signatures", + Operation::ReadBlock => "read_block", + Operation::ReadBlockRange => "read_block_range", + Operation::Replay => "replay", + Operation::Rotate => "rotate", + Operation::Truncate => "truncate", + Operation::FileSync => "file_sync", + Operation::FileFinalize => "file_finalize", + } + } +} + +/// Registers ledger metrics once and seeds gauges from opened ledger metadata. +pub(crate) fn init(ledger: &Ledger) { + METRICS.get_or_init(|| Metrics::new(ledger)); +} + +/// Starts an operation timer that records latency when the returned guard drops. +pub(crate) fn time(op: Operation) -> metric::OperationTimer<'static> { + op.time(METRICS.get().map(|m| &m.operations)) +} + +/// Refreshes the current pending transaction count. +pub(crate) fn pending_transactions(count: usize) { + metric::with_metrics(&METRICS, |m| m.pending_transactions.set(count as i64)); +} + +/// Refreshes ledger-wide count gauges from metadata. +pub(crate) fn ledger_counts(ledger: &Ledger) { + metric::with_metrics(&METRICS, |m| m.ledger_counts(ledger)); +} + +/// Owns all Prometheus collectors registered by ledger. +struct Metrics { + /// Runtime operation duration and completion counters. + operations: OperationCounters, + /// Runtime pending transaction count. + pending_transactions: IntGauge, + /// Runtime transaction total gauge. + transactions: IntGauge, + /// Runtime block total gauge. + blocks: IntGauge, + /// Runtime superblock total gauge. + superblocks: IntGauge, +} + +impl Metrics { + /// Builds collectors and registers them in the default Prometheus registry. + fn new(ledger: &Ledger) -> Self { + let metrics = Self { + operations: OperationCounters::new(OPERATION_TIME), + pending_transactions: metric::gauge(PENDING_TRANSACTIONS, 0), + transactions: metric::gauge( + TRANSACTIONS, + ledger.meta.transactions.load(Relaxed) as i64, + ), + blocks: metric::gauge(BLOCKS, ledger.meta.blocks.load(Relaxed) as i64), + superblocks: metric::gauge(SUPERBLOCKS, ledger.meta.head() as i64), + }; + metrics.register(); + metrics + } + + /// Refreshes count gauges from ledger metadata. + fn ledger_counts(&self, ledger: &Ledger) { + self.transactions.set(ledger.meta.transactions.load(Relaxed) as i64); + self.blocks.set(ledger.meta.blocks.load(Relaxed) as i64); + self.superblocks.set(ledger.meta.head() as i64); + } + + /// Registers all collectors in the default Prometheus registry. + fn register(&self) { + self.operations.register("ledger"); + metric::register("ledger", self.pending_transactions.clone()); + metric::register("ledger", self.transactions.clone()); + metric::register("ledger", self.blocks.clone()); + metric::register("ledger", self.superblocks.clone()); + } +} diff --git a/ledger/src/reader.rs b/ledger/src/reader.rs new file mode 100644 index 0000000..c476687 --- /dev/null +++ b/ledger/src/reader.rs @@ -0,0 +1,476 @@ +//! Ledger read-side worker implementation. + +use std::{ + collections::HashMap, + io::{BufRead, BufReader, Read}, + mem, + ops::Range, + os::unix::fs::FileExt, + sync::{Arc, atomic::Ordering::Acquire}, +}; + +use agave_transaction_view::transaction_view::SanitizedTransactionView; +use bitcode::Buffer; +use flume::Receiver; +use nucleus::{ + MB, Slot, + heed::OptRoTxn, + shutdown::{ShutdownHandle, ShutdownReason}, +}; +use solana_signature::Signature; +use tracing::error; +use wincode::{Error, io::Cursor}; +use zstd::bulk::Decompressor; + +use crate::{ + Ledger, LedgerError, Result, Superblock, + index::{AccountIter, Span}, + metrics::{self, Operation}, + request::{ + AccountSignature, AccountSignaturesParams, AccountSignaturesPayload, + AccountSignaturesReadResult, BlockDetails, BlockParams, BlockPayload, BlockReadResult, + BlockResponse, BlockWithSignatures, BlockWithTransactions, FullBlockInfo, ReadRequest, + ReplayParams, ReplayPayload, TransactionPayload, TransactionReadResult, + TransactionResponse, TransactionStatus, TransactionStatusPayload, + TransactionStatusReadResult, + }, + schema::{Block, BlockstoreEntry, Execution, ExecutionHeader, OwnedBlockestoreEntry}, +}; + +/// Stateful ledger reader with reusable decoder buffers. +pub(crate) struct LedgerReader { + /// Request stream shared by callers through `LedgerHandle`. + rx: Receiver, + /// Shared top-level ledger state. + ledger: Arc, + /// Reusable zstd decompressor. + decompressor: Decompressor<'static>, + /// Scratch buffers reused across requests. + buffers: ReadBuffers, +} + +impl LedgerReader { + /// Creates a reader over `ledger`. + pub(crate) fn new(ledger: Arc, rx: Receiver) -> Result { + let mut buffers = ReadBuffers::default(); + buffers.details.reserve(4 * MB); + Ok(Self { + rx, + ledger, + decompressor: Decompressor::new()?, + buffers, + }) + } + + /// Serves requests until every sender has been dropped. + pub(crate) fn run(mut self, mut shutdown: ShutdownHandle) { + while let Ok(request) = self.rx.recv() { + match request { + ReadRequest::Transaction(r) => { + let _timer = metrics::time(Operation::ReadTransaction); + let result = self.transaction(&r); + let _ = r.response.send(result); + } + ReadRequest::TransactionStatus(r) => { + let _timer = metrics::time(Operation::ReadTransactionStatus); + let result = self.transaction_status(&r); + let _ = r.response.send(result); + } + ReadRequest::AccountSignatures(r) => { + let _timer = metrics::time(Operation::ReadAccountSignatures); + let result = self.account_signatures(&r); + let _ = r.response.send(result); + } + ReadRequest::Block(r) => { + let _timer = metrics::time(Operation::ReadBlock); + let result = self.block(&r); + let _ = r.response.send(result); + } + ReadRequest::BlockRange(r) => { + let _timer = metrics::time(Operation::ReadBlockRange); + let result = self.blocks(r.params); + let _ = r.response.send(result); + } + ReadRequest::Replay(r) => { + let _timer = metrics::time(Operation::Replay); + let result = self.replay(&r); + let _ = r.response.send(result); + } + } + } + shutdown.terminate(ShutdownReason::Signalled); + } + + /// Reads a full transaction and its execution details. + fn transaction(&mut self, request: &TransactionPayload) -> TransactionReadResult { + for superblock in self.ledger.clone().iter() { + if request.cancelled() { + return Ok(None); + } + let Some(spans) = superblock.index.transaction(&request.params, &mut None)? else { + continue; + }; + return Ok(Some(TransactionResponse { + transaction: self.transaction_entry(&superblock, spans.blockstore)?, + execution: self.execution(&superblock, spans.execution)?, + })); + } + Ok(None) + } + + /// Reads the status header for a transaction. + fn transaction_status( + &mut self, + request: &TransactionStatusPayload, + ) -> TransactionStatusReadResult { + for superblock in self.ledger.clone().iter() { + if request.cancelled() { + return Ok(None); + } + let Some(spans) = superblock.index.transaction(&request.params, &mut None)? else { + continue; + }; + let header = self.header(&superblock, spans.execution)?; + return Ok(Some(TransactionStatus { + result: header.result, + slot: header.slot, + })); + } + Ok(None) + } + + /// Reads recent signatures that mention an account. + fn account_signatures( + &mut self, + request: &AccountSignaturesPayload, + ) -> AccountSignaturesReadResult { + let mut signatures = Vec::new(); + let AccountSignaturesParams { pubkey, limit, mut before, until } = request.params; + let mut blocktimes = HashMap::::new(); + for superblock in self.ledger.clone().iter() { + 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 mut upper = None; + if let Some(signature) = &before { + let Some(cutoff) = superblock.index.transaction(signature, &mut txn)? else { + continue; + }; + upper.replace(cutoff.execution); + before.take(); + } + + for result in iter { + let span = result?.1; + if let Some(s) = upper + && span >= s + { + continue; + }; + let header = self.header(&superblock, span)?; + let sig: Signature = header.signature; + if let Some(signature) = until + && sig == signature + { + return Ok(signatures); + } + // Several matching signatures can share a slot, so cache block timestamps per slot. + let blocktime = match blocktimes.get(&header.slot) { + Some(time) => *time, + None => { + let time = self.blocktime(&superblock, header.slot, &mut txn)?; + blocktimes.insert(header.slot, time); + time + } + }; + signatures.push(AccountSignature { + signature: sig, + result: header.result, + slot: header.slot, + blocktime, + }); + if signatures.len() >= limit { + return Ok(signatures); + } + } + } + Ok(signatures) + } + + /// Reads a block with the requested transaction detail level. + fn block(&mut self, request: &BlockPayload) -> BlockReadResult { + let slot = request.params.slot; + if !self.ledger.meta.range.contains(&slot) { + return Ok(None); + } + + let mut position = None; + for superblock in self.ledger.clone().iter() { + if request.cancelled() { + return Ok(None); + } + if !superblock.meta.range.contains(&slot) { + continue; + } + let Some(span) = superblock.index.block(&slot, &mut None)? else { + return Ok(None); + }; + position.replace((superblock, span)); + break; + } + let Some((superblock, span)) = position else { return Ok(None) }; + self.populate_block_response(&superblock, request, span).map(Some) + } + + /// 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(); + + 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; + }; + if let BlockstoreEntry::Block(b) = self.blockstore_entry(&superblock, span)? { + blocks.push(b); + } + } + } + blocks.reverse(); + Ok(blocks) + } + + fn replay(&mut self, request: &ReplayPayload) -> Result<()> { + let ReplayParams { from: slot, tx } = &request.params; + let mut from = None; + for superblock in self.ledger.iter() { + if superblock.meta.range.contains(slot) { + from.replace(superblock.id); + break; + } + } + let Some(from) = from else { return Ok(()) }; + for superblock in self.ledger.iter_from(from) { + let limit = superblock.meta.cursors.blockstore.load(Acquire); + let mut reader = BufReader::new((&superblock.blockstore).take(limit)); + while !reader.fill_buf()?.is_empty() { + let entry = wincode::deserialize_from::(&mut reader) + .map_err(Into::::into)?; + if tx.blocking_send(entry).is_err() { + break; + } + } + } + Ok(()) + } + + /// Reads and decodes a raw transaction entry. + fn transaction_entry(&mut self, superblock: &Superblock, span: Span) -> Result> { + let entry = self.blockstore_entry(superblock, span)?; + let BlockstoreEntry::Transaction(transaction) = entry else { + error!( + superblock = superblock.id, + "ledger index points to invalid transaction entry" + ); + return Err(LedgerError::Corruption( + "index points to invalid transaction entry", + )); + }; + Ok(transaction) + } + + /// Reads and decodes a complete execution entry. + fn execution(&mut self, superblock: &Superblock, span: Span) -> Result { + self.buffers.read_executions(superblock, span)?; + let header = self.decode_header()?; + let header_size = wincode::serialized_size(&header).map_err(Into::::into)? as usize; + + let compressed = &self.buffers.executions[header_size..]; + self.buffers.details.clear(); + self.decompressor.decompress_to_buffer(compressed, &mut self.buffers.details)?; + let details = self.buffers.decoder.decode(&self.buffers.details)?; + Ok(Execution { header, details }) + } + + /// Reads and decodes one blockstore entry at an indexed span. + fn blockstore_entry( + &mut self, + superblock: &Superblock, + span: Span, + ) -> Result { + self.buffers.read_blockstore(superblock, span)?; + let entry = wincode::deserialize(self.buffers.blockstore.as_slice()) + .map_err(Into::::into)?; + Ok(entry) + } + + /// Builds the block response requested by the caller. + fn populate_block_response( + &mut self, + superblock: &Superblock, + request: &BlockPayload, + span: Span, + ) -> Result { + let BlockParams { slot, details } = request.params; + let BlockstoreEntry::Block(block) = self.blockstore_entry(superblock, span)? else { + error!( + superblock = superblock.id, + slot, "ledger index points to invalid block entry" + ); + return Err(LedgerError::Corruption( + "index points to invalid block entry", + )); + }; + + 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)? { + Some(previous) => previous.offset() + previous.size(), + None => 0, + }; + self.buffers.read_blockstore_range(superblock, start..span.offset())?; + let len = self.buffers.blockstore.len(); + let mut cursor = Cursor::new(self.buffers.blockstore.clone()); + let mut full = Vec::new(); + let mut transactions = Vec::new(); + let mut signatures = Vec::new(); + while cursor.position() < len && !request.cancelled() { + let entry = wincode::deserialize_from::(&mut cursor) + .map_err(Into::::into)?; + let BlockstoreEntry::Transaction(transaction) = entry else { + error!( + superblock = superblock.id, + slot, "ledger blockstore includes invalid block entries" + ); + return Err(LedgerError::Corruption( + "blockstore includes invalid block entries", + )); + }; + if matches!(details, BlockDetails::Transactions) { + transactions.push(transaction); + continue; + } + let signature = signature(&transaction)?; + if matches!(details, BlockDetails::Signatures) { + signatures.push(signature); + continue; + } + let Some(spans) = superblock.index.transaction(&signature, &mut txn)? else { + continue; + }; + let execution = self.execution(superblock, spans.execution)?; + full.push(TransactionResponse { transaction, execution }); + } + + let response = match details { + BlockDetails::Full => BlockResponse::Full(FullBlockInfo { block, transactions: full }), + BlockDetails::Transactions => { + BlockResponse::WithTransactions(BlockWithTransactions { block, transactions }) + } + BlockDetails::Signatures => { + BlockResponse::WithSignatures(BlockWithSignatures { block, signatures }) + } + BlockDetails::None => BlockResponse::Bare(block), + }; + Ok(response) + } + + /// Reads the timestamp from a block boundary entry. + fn blocktime( + &mut self, + superblock: &Superblock, + slot: Slot, + txn: OptRoTxn<'_, '_>, + ) -> Result { + let Some(span) = superblock.index.block(&slot, txn)? else { + return Ok(0); + }; + let entry = self.blockstore_entry(superblock, span)?; + let BlockstoreEntry::Block(block) = entry else { + return Ok(0); + }; + Ok(block.time) + } + + /// Reads only the fixed execution header. + fn header(&mut self, superblock: &Superblock, span: Span) -> Result { + self.buffers.read_executions(superblock, span)?; + self.decode_header() + } + + /// Decodes the execution header currently loaded in the executions buffer. + fn decode_header(&self) -> Result { + let header = wincode::deserialize(self.buffers.executions.as_slice()) + .map_err(Into::::into)?; + Ok(header) + } +} + +/// Reusable read buffers owned by one reader task. +#[derive(Default)] +struct ReadBuffers { + /// Blockstore-file scratch bytes. + blockstore: Vec, + /// Executions-file scratch bytes. + executions: Vec, + /// Bitcode decoder state for execution details. + decoder: Buffer, + /// Decompressed execution-detail payload. + details: Vec, +} + +impl ReadBuffers { + /// Reads blockstore-file bytes at `span` into the reusable buffer. + fn read_blockstore(&mut self, superblock: &Superblock, span: Span) -> Result<()> { + self.blockstore.resize(span.size() as usize, 0); + superblock.blockstore.read_exact_at(&mut self.blockstore, span.offset())?; + Ok(()) + } + + /// Reads executions-file bytes at `span` into the reusable buffer. + fn read_executions(&mut self, superblock: &Superblock, span: Span) -> Result<()> { + self.executions.resize(span.size() as usize, 0); + superblock.executions.read_exact_at(&mut self.executions, span.offset())?; + Ok(()) + } + + /// Reads a byte range from the blockstore file into the reusable buffer. + fn read_blockstore_range(&mut self, superblock: &Superblock, range: Range) -> Result<()> { + self.blockstore.resize((range.end - range.start) as usize, 0); + superblock.blockstore.read_exact_at(&mut self.blockstore, range.start)?; + Ok(()) + } +} + +fn signature(transaction: &[u8]) -> Result { + SanitizedTransactionView::try_new_sanitized(transaction, false) + .map(|transaction| transaction.signatures()[0]) + .map_err(Into::into) +} + +// SAFETY: `LedgerReader` is moved into one background thread and keeps its +// decoder and decompressor state on that thread for the reader lifetime. +unsafe impl Send for LedgerReader {} diff --git a/ledger/src/request.rs b/ledger/src/request.rs new file mode 100644 index 0000000..ea1eefe --- /dev/null +++ b/ledger/src/request.rs @@ -0,0 +1,264 @@ +//! Ledger read-side request and response types. + +use std::{ + ops::Range, + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Duration, +}; + +use flume::Sender as RequestSender; +use nucleus::Slot; +use oneshot::{Receiver, Sender}; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use solana_transaction_error::TransactionResult; +use tokio::sync::mpsc; + +use crate::{ + Result, + error::RequestResult, + schema::{Block, Execution, OwnedBlockestoreEntry}, +}; + +/// Result returned by a full transaction lookup. +pub(crate) type TransactionReadResult = Result>; +/// Result returned by a transaction-status lookup. +pub(crate) type TransactionStatusReadResult = Result>; +/// Result returned by an account-signature history lookup. +pub(crate) type AccountSignaturesReadResult = Result>; +/// Result returned by a block lookup. +pub(crate) type BlockReadResult = Result>; + +/// Payload for a transaction lookup request. +pub(crate) type TransactionPayload = RequestPayload; +/// Payload for a transaction-status lookup request. +pub(crate) type TransactionStatusPayload = RequestPayload; +/// Payload for an account-signature history request. +pub(crate) type AccountSignaturesPayload = + RequestPayload; +/// Payload for a single-block lookup request. +pub(crate) type BlockPayload = RequestPayload; +/// Payload for a contiguous block-range request. +pub(crate) type BlockRangePayload = RequestPayload, Result>>; +/// Payload for replaying owned blockstore entries from a slot onward. +pub(crate) type ReplayPayload = RequestPayload>; + +/// Read request queue consumed by reader workers. +pub type ReaderSender = RequestSender; + +/// Handle for consuming a ledger replay stream. +pub struct ReplayHandle { + /// Entries streamed from retained blockstore data in on-disk order. + pub rx: mpsc::Receiver, + /// Completion result sent after the reader finishes streaming entries. + pub response: RequestHandle>, +} + +/// Signature summary returned for account history queries. +pub struct AccountSignature { + /// Transaction signature. + pub signature: Signature, + /// Slot where the transaction executed. + pub slot: Slot, + /// Runtime transaction result. + pub result: TransactionResult<()>, + /// Block timestamp for `slot`, or `0` when unavailable. + pub blocktime: i64, +} + +/// Full transaction response with optional execution metadata. +pub struct TransactionResponse { + /// Serialized transaction bytes from the blockstore file. + pub transaction: Vec, + /// Execution metadata when requested or available. + pub execution: Execution, +} + +/// Cheap transaction-status response. +#[derive(Clone)] +pub struct TransactionStatus { + /// Runtime transaction result. + pub result: TransactionResult<()>, + /// Slot where the transaction executed. + pub slot: Slot, +} + +/// Account-signature query parameters. +pub struct AccountSignaturesParams { + /// Account pubkey to search for. + pub pubkey: Pubkey, + /// Maximum number of signatures to return. + pub limit: usize, + /// Signature before which results should start. + pub before: Option, + /// Signature at which results should stop. + pub until: Option, +} + +/// Parameters for streaming retained blockstore entries into a replay consumer. +pub struct ReplayParams { + /// First slot whose containing superblock should be replayed. + pub from: Slot, + /// Channel receiving blockstore entries in on-disk order. + pub tx: mpsc::Sender, +} + +/// Read request sent to a ledger reader service. +pub enum ReadRequest { + /// Full transaction lookup by signature. + Transaction(TransactionPayload), + /// Transaction status lookup by signature. + TransactionStatus(TransactionStatusPayload), + /// Account-signature history lookup. + AccountSignatures(AccountSignaturesPayload), + /// Block lookup by slot and detail level. + Block(BlockPayload), + /// Contiguous block boundary lookup. + BlockRange(BlockRangePayload), + /// Replay retained blockstore entries from a slot onward. + Replay(ReplayPayload), +} + +/// Block lookup response at the requested detail level. +pub enum BlockResponse { + /// Block boundary only. + Bare(Block), + /// Block boundary with transaction signatures. + WithSignatures(BlockWithSignatures), + /// Block boundary with serialized transactions. + WithTransactions(BlockWithTransactions), + /// Block boundary with serialized transactions and execution metadata. + Full(FullBlockInfo), +} + +impl BlockResponse { + /// Returns the block boundary included in every response variant. + pub fn block(&self) -> &Block { + match self { + Self::Bare(b) => b, + Self::WithSignatures(b) => &b.block, + Self::WithTransactions(b) => &b.block, + Self::Full(b) => &b.block, + } + } +} + +/// Full block response with execution metadata. +pub struct FullBlockInfo { + /// Block boundary. + pub block: Block, + /// Transactions with matching execution metadata. + pub transactions: Vec, +} + +/// Block response containing serialized transactions only. +pub struct BlockWithTransactions { + /// Block boundary. + pub block: Block, + /// Serialized transactions in block order. + pub transactions: Vec>, +} + +/// Block response containing transaction signatures only. +pub struct BlockWithSignatures { + /// Block boundary. + pub block: Block, + /// Transaction signatures in block order. + pub signatures: Vec, +} + +/// Handle used by the request caller to await a reader response. +pub struct RequestHandle { + /// One-shot response receiver. + rx: Receiver, + /// Pending flag guard cleared when the response is no longer needed. + pending: PendingGuard, +} + +impl RequestHandle { + /// Wait up to one minute for a reader response. + /// + /// Dropping the handle before a response clears the request's pending flag + /// so long-running readers can stop work that no caller will receive. + pub async fn recv_timeout(mut self) -> RequestResult { + const ONE_MINUTE: Duration = Duration::from_secs(60); + let result = tokio::time::timeout(ONE_MINUTE, self.rx).await??; + self.pending.0.take(); + Ok(result) + } + + /// Wait for a reader response without applying a timeout. + pub async fn recv(mut self) -> RequestResult { + let result = self.rx.await?; + self.pending.0.take(); + Ok(result) + } +} + +/// Clears the request's pending flag when dropped. +struct PendingGuard(Option>); + +impl Drop for PendingGuard { + fn drop(&mut self) { + if let Some(pending) = self.0.take() { + pending.store(false, Ordering::Release); + } + } +} + +/// Request payload with cancellation and oneshot response channel. +pub struct RequestPayload { + /// Request parameters. + pub params: P, + /// Response channel for the result. + pub response: Sender, + /// Set while the caller is still waiting for the response. + pub pending: Arc, +} + +impl RequestPayload { + /// Creates a request payload and the handle that receives its response. + pub fn new(params: P) -> (Self, RequestHandle) { + let (response, rx) = oneshot::channel(); + let pending = Arc::new(AtomicBool::new(true)); + let payload = Self { + params, + response, + pending: pending.clone(), + }; + let handle = RequestHandle { + rx, + pending: PendingGuard(Some(pending)), + }; + (payload, handle) + } + + /// Returns whether the caller stopped waiting for this response. + pub(crate) fn cancelled(&self) -> bool { + !self.pending.load(Ordering::Acquire) + } +} + +/// Block query parameters. +pub struct BlockParams { + /// Slot to look up. + pub slot: Slot, + /// Amount of transaction data to include in the response. + pub details: BlockDetails, +} + +/// Transaction detail level for a block lookup. +#[derive(Clone, Copy)] +pub enum BlockDetails { + /// Include transactions and execution details. + Full, + /// Include transactions without execution details. + Transactions, + /// Include only transaction signatures. + Signatures, + /// Include only the block boundary entry. + None, +} diff --git a/ledger/src/schema.rs b/ledger/src/schema.rs new file mode 100644 index 0000000..1523e15 --- /dev/null +++ b/ledger/src/schema.rs @@ -0,0 +1,137 @@ +//! Ledger wire, event, and on-disk blockstore formats. +//! +//! The blockstore is a wincode stream of raw transactions in execution order. +//! Block entries delimit the transactions that belong to each Solana-like +//! block, and a superblock seal terminates a truncatable group of blocks. +//! Execution details live in a separate file as a wincode header followed by +//! a zstd-compressed bitcode payload. + +use std::sync::Arc; + +use bitcode::{Decode, Encode}; +use nucleus::Slot; +pub use nucleus::ledger::{Block, SuperblockSeal}; + +use solana_signature::Signature; +use solana_transaction_error::TransactionResult; +use wincode::{SchemaRead, SchemaWrite}; + +/// Byte offset into a ledger data file. +pub(crate) type Offset = u64; + +/// Owned blockstore entry used when replaying persisted ledger data. +pub type OwnedBlockestoreEntry = BlockstoreEntry>; + +/// Work item sent to the ledger appender. +// Keep execution inline to avoid one allocation per append event. +#[allow(clippy::large_enum_variant)] +pub enum Event { + /// Transaction bytes accepted for later execution indexing. + Transaction(TransactionEntry), + /// Runtime execution metadata for a previously appended transaction. + Execution(Execution), + /// Block boundary marker and block hash. + Block(Block), + /// Superblock boundary marker used to rotate the active segment. + Superblock(SuperblockSeal), +} + +/// Raw transaction payload queued for blockstore append. +pub struct TransactionEntry { + /// First transaction signature used for status and execution indexes. + pub signature: Signature, + /// Serialized sanitized transaction bytes shared with the appender. + pub payload: Arc>, +} + +/// One typed entry in the blockstore stream. +#[derive(SchemaRead, SchemaWrite)] +pub enum BlockstoreEntry { + /// Delimits the preceding transactions as one block and stores its hash. + Block(Block), + /// Serialized transaction bytes. + Transaction(T), + /// Superblock seal marker and snapshot checksum. + Superblock(SuperblockSeal), +} + +/// Fixed execution prefix stored before the compressed bitcode payload. +#[derive(SchemaRead, SchemaWrite)] +pub struct ExecutionHeader { + /// Transaction signature. + pub signature: Signature, + /// Runtime transaction error, if execution failed. + pub result: TransactionResult<()>, + /// Slot where the transaction executed. + pub slot: Slot, +} + +/// Complete transaction execution metadata. +pub struct Execution { + /// Fixed prefix used for cheap signature and error reads. + pub header: ExecutionHeader, + /// Execution details stored in the compressed payload. + pub details: Option, +} + +/// Transaction execution details stored after the execution header. +#[derive(Encode, Decode)] +pub struct ExecutionDetails { + /// Fee charged for the transaction. + pub fee: u64, + /// Pre/post balance deltas encoded in a compact prototype layout. + pub balances: Balances, + /// Log messages emitted during execution. + pub logs: Arc>, + /// Cross-program invocation trace. + pub cpi: Option>, + /// Compute units consumed. + pub compute_units: u64, + /// Program return data, if any. + pub return_data: Option, +} + +/// One cross-program invocation group. +#[derive(Encode, Decode)] +pub struct Cpis( + /// Inner instructions executed at from outer instruction level. + pub Vec, +); + +/// Runtime instruction metadata. +#[derive(Encode, Decode)] +pub struct Instruction { + /// Compiled Solana instruction. + pub compiled: CompiledInstruction, + /// Optional invocation stack height. + pub stack_height: u8, +} + +/// Compact compiled instruction form. +#[derive(Encode, Decode)] +pub struct CompiledInstruction { + /// Program account index. + pub program_index: u8, + /// Account indexes referenced by the instruction. + pub accounts: Vec, + /// Opaque instruction data. + pub data: Vec, +} + +/// Program return data. +#[derive(Encode, Decode)] +pub struct ReturnData { + /// Program that produced the return data. + pub program: [u8; 32], + /// Opaque return bytes. + pub data: Arc>, +} + +/// Pre/post balance vectors. +#[derive(Encode, Decode)] +pub struct Balances { + /// Balances before execution. + pub pre: Vec, + /// Balances after execution. + pub post: Vec, +} diff --git a/ledger/src/storage.rs b/ledger/src/storage.rs new file mode 100644 index 0000000..01e7b57 --- /dev/null +++ b/ledger/src/storage.rs @@ -0,0 +1,287 @@ +//! Superblock storage primitives. +//! +//! This module keeps the low-level storage surface in one place: buffered +//! append files and mmap-backed metadata headers. + +use std::{ + fs::File, + io::{self, Write}, + ops::{Deref, Range}, + os::{ + fd::{AsFd, AsRawFd}, + unix::fs::FileExt, + }, + path::Path, + ptr::NonNull, + sync::atomic::{AtomicBool, AtomicU64, Ordering::*}, +}; + +use memmap2::{MmapMut, MmapOptions}; +use nucleus::{MB, Slot}; +use rustix::fs::{self, FallocateFlags}; +use tracing::debug; +use zstd::bulk::Compressor; + +use crate::Result; + +/// Initial buffer size for append-heavy files. +const FILE_BUFFER_SIZE: usize = 128 * MB; +/// Physical space reserved when an append file is close to its allocated end. +// NOTE: fallocate is slow on macos, so we use tiny increments for tests +#[cfg(any(test, target_os = "macos"))] +const PREALLOCATION_SIZE: u64 = MB as u64; +#[cfg(not(target_os = "macos"))] +const PREALLOCATION_SIZE: u64 = 4 * nucleus::GB as u64; + +/// Remaining allocation that triggers another reservation. +const PREALLOCATION_THRESHOLD: u64 = PREALLOCATION_SIZE / 16; + +/// Buffered append writer that tracks logical file position. +pub(crate) struct AppendFile { + /// Backing file. + pub(crate) file: File, + /// Pending bytes not yet flushed to the backing file. + buffer: Vec, + /// Logical byte position including buffered bytes. + pub(crate) cursor: u64, + len: u64, +} + +impl AppendFile { + /// Opens an append file and resumes from the persisted cursor. + pub(crate) fn new(path: &Path, cursor: &AtomicU64) -> Result { + let file = + File::options().create(true).truncate(false).read(true).write(true).open(path)?; + #[cfg(target_os = "linux")] + fs::fadvise(file.as_raw_fd(), 0, None, fs::Advice::Sequential)?; + let cursor = cursor.load(Acquire); + let len = file.metadata()?.len(); + Ok(Self { + file, + buffer: Vec::with_capacity(FILE_BUFFER_SIZE), + cursor, + len, + }) + } + + /// Compresses `bytes` into the pending buffer and advances the logical cursor. + pub(crate) fn compress( + &mut self, + bytes: &[u8], + compressor: &mut Compressor<'static>, + ) -> io::Result<()> { + let buffered = self.buffer.len(); + compressor.compress_to_buffer(bytes, &mut self.buffer)?; + self.cursor += (self.buffer.len() - buffered) as u64; + if self.buffer.len() >= FILE_BUFFER_SIZE { + self.flush()?; + } + Ok(()) + } + + /// Flushes buffered bytes, syncs file data, and returns the durable cursor. + pub(crate) fn sync(&mut self) -> Result { + if self.len.saturating_sub(self.cursor) < PREALLOCATION_THRESHOLD { + self.preallocate()?; + } + self.flush()?; + self.file.sync_data()?; + Ok(self.cursor) + } + + /// Extends the physical file without changing the logical append cursor. + fn preallocate(&mut self) -> Result<()> { + let offset = self.len; + fs::fallocate( + self.file.as_fd(), + FallocateFlags::empty(), + offset, + PREALLOCATION_SIZE, + )?; + self.file.sync_all()?; + self.len = offset + PREALLOCATION_SIZE; + debug!(len = self.len, "preallocated ledger file space"); + Ok(()) + } + + /// Trims unused preallocated space after the superblock is sealed. + pub(crate) fn finalize(&mut self) -> Result<()> { + self.flush()?; + #[cfg(target_os = "linux")] + fs::fadvise(self.file.as_raw_fd(), 0, None, fs::Advice::RANDOM)?; + self.file.set_len(self.cursor)?; + self.file.sync_all()?; + self.len = self.cursor; + Ok(()) + } +} + +impl Write for AppendFile { + fn write(&mut self, buf: &[u8]) -> io::Result { + self.buffer.extend_from_slice(buf); + self.cursor += buf.len() as u64; + if self.buffer.len() >= FILE_BUFFER_SIZE { + self.flush()?; + } + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + if self.buffer.is_empty() { + return Ok(()); + } + // `cursor` includes the buffered bytes; indexes use these logical offsets. + let offset = self.cursor - self.buffer.len() as u64; + self.file.write_all_at(&self.buffer, offset)?; + self.buffer.clear(); + self.len = self.len.max(self.cursor); + Ok(()) + } +} + +/// Mmap-backed fixed-size metadata header. +pub(crate) struct MetaMap { + /// Typed pointer into the mapped header. + data: NonNull, + /// Mutable mapping kept alive for `data`. + mmap: MmapMut, +} + +impl MetaMap { + /// Opens `path` and initializes it with `T::default()` when empty. + /// + /// # Safety + /// + /// `T` must be a plain fixed-layout metadata header that can be safely + /// read from bytes previously written by this process. All shared mutation + /// inside `T` must use synchronization such as atomics. + #[allow(unsafe_op_in_unsafe_fn)] + pub(crate) unsafe fn new(path: &Path) -> Result { + let size = size_of::().div_ceil(512) * 512; + let file = + File::options().write(true).read(true).create(true).truncate(false).open(path)?; + if file.metadata()?.len() == 0 { + file.set_len(size as u64)?; + let mut mmap = MmapOptions::new().len(size).map_mut(file.as_raw_fd())?; + mmap.as_mut_ptr().cast::().write(T::default()); + mmap.flush()?; + let data = NonNull::new_unchecked(mmap.as_mut_ptr().cast::()); + return Ok(Self { data, mmap }); + } + + // SAFETY: `size` matches the fixed metadata header size used when the + // file was created, and `mmap` is owned by the returned `MetaMap`. + let mut mmap = MmapOptions::new().len(size).map_mut(file.as_raw_fd())?; + // SAFETY: mmap pointers are non-null for non-empty mappings. + let data = NonNull::new_unchecked(mmap.as_mut_ptr().cast::()); + Ok(Self { data, mmap }) + } + + /// Flushes metadata changes to disk. + #[inline] + pub(crate) fn flush(&self) -> Result<()> { + self.mmap.flush().map_err(Into::into) + } +} + +impl Deref for MetaMap { + type Target = T; + fn deref(&self) -> &Self::Target { + unsafe { self.data.as_ref() } + } +} + +// SAFETY: `MetaMap` owns the mapping that backs `data`; moving the wrapper +// does not invalidate the pointer. Cross-thread access is constrained by `T`. +unsafe impl Send for MetaMap {} + +// SAFETY: shared access exposes only `&T`. The metadata headers used by the +// ledger mutate shared state through atomics. +unsafe impl Sync for MetaMap {} + +/// Ledger-wide metadata header. +#[repr(C)] +pub(crate) struct LedgerMeta { + /// Total transactions committed since genesis. + pub(crate) transactions: AtomicU64, + /// Number of retained superblocks. + pub(crate) superblocks: AtomicU64, + /// Active superblock identifier. + pub(crate) head: AtomicU64, + /// Inclusive slot range covered by retained superblocks. + pub(crate) range: BlockRange, + /// Total blocks committed since genesis. + pub(crate) blocks: AtomicU64, +} + +impl Default for LedgerMeta { + fn default() -> Self { + Self { + transactions: 0.into(), + head: 1.into(), + superblocks: 1.into(), + range: Default::default(), + blocks: 0.into(), + } + } +} + +impl LedgerMeta { + /// Returns the active superblock id. + #[inline] + pub(crate) fn head(&self) -> u64 { + self.head.load(Acquire) + } + + /// Advances the active superblock id and returns the new id. + #[inline] + pub(crate) fn advance(&self) -> u64 { + self.head.fetch_add(1, Release) + 1 + } + + /// Returns the retained superblock id range, including the active head. + #[inline] + pub(crate) fn superblocks(&self) -> Range { + let head = self.head(); + let start = head - self.superblocks.load(Acquire).saturating_sub(1); + start..head + 1 + } +} + +/// Metadata header for one superblock directory. +#[derive(Default)] +#[repr(C)] +pub(crate) struct SuperblockMeta { + /// Durable append cursors for files in this superblock. + pub(crate) cursors: FileCursors, + /// Slot range stored in this segment. + pub(crate) range: BlockRange, + /// Whether this segment has been sealed. + pub(crate) complete: AtomicBool, +} + +/// Durable append cursors for superblock data files. +#[derive(Default)] +#[repr(C)] +pub(crate) struct FileCursors { + /// Durable byte cursor in `blockstore.db`. + pub(crate) blockstore: AtomicU64, + /// Durable byte cursor in `executions.db`. + pub(crate) executions: AtomicU64, +} + +/// Inclusive slot range covered by a superblock. +#[derive(Default)] +pub(crate) struct BlockRange { + /// First slot in the segment. + pub(crate) start: AtomicU64, + /// Last slot in the segment. + pub(crate) end: AtomicU64, +} + +impl BlockRange { + /// Returns true when `slot` is inside the range. + pub(crate) fn contains(&self, slot: &Slot) -> bool { + (self.start.load(Acquire)..=self.end.load(Acquire)).contains(slot) + } +} diff --git a/ledger/src/tests/index.rs b/ledger/src/tests/index.rs new file mode 100644 index 0000000..64d5951 --- /dev/null +++ b/ledger/src/tests/index.rs @@ -0,0 +1,109 @@ +//! Index unit tests. + +use heed::RwTxn; +use nucleus::{ + Slot, + testkit::{TempDir, init_tracing, tempdir}, +}; +use solana_pubkey::Pubkey; +use solana_signature::Signature; + +use crate::index::{Index, Span, TxSpan}; + +/// Opens a fresh index on a throwaway directory kept alive by the returned guard. +fn index() -> (TempDir, Index) { + init_tracing(); + let dir = tempdir(); + let index = Index::new(dir.path()).unwrap(); + (dir, index) +} + +/// Commits a lazily opened write transaction, if any inserts opened one. +fn commit(txn: Option>) { + if let Some(txn) = txn { + txn.commit().unwrap(); + } +} + +/// 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() +} + +#[test] +fn span_pack_and_order() { + let span = Span::new(1234, 56); + assert_eq!(span.offset(), 1234); + assert_eq!(span.size(), 56); + + // Boundary values: full size field, and an offset filling every high bit. + let max_size = Span::new(0, Span::MAX_SIZE); + assert_eq!((max_size.offset(), max_size.size()), (0, Span::MAX_SIZE)); + let max_offset = u64::MAX >> 25; + let high = Span::new(max_offset, 0); + assert_eq!((high.offset(), high.size()), (max_offset, 0)); + + // Ordering is offset-dominant: a later offset outranks any size at an + // earlier offset, and size only breaks ties within one offset. + assert!(Span::new(11, 0) > Span::new(10, Span::MAX_SIZE)); + assert!(Span::new(10, 7) > Span::new(10, 5)); +} + +#[test] +fn transaction_and_block_roundtrip() { + let (_dir, index) = index(); + let signature = Signature::from([7; 64]); + let txspan = TxSpan { + blockstore: Span::new(10, 20), + execution: Span::new(30, 40), + }; + let (slot_a, slot_b): (Slot, Slot) = (1, 2); + let block_a = Span::new(0, 8); + let block_b = Span::new(8, 16); + + let mut txn = None; + 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(); + commit(txn); + + let mut txn = None; + let got = index.transaction(&signature, &mut txn).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)); + + // 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); +} + +#[test] +fn account_signature_duplicates() { + let (_dir, index) = index(); + let account = Pubkey::new_unique(); + let other = Pubkey::new_unique(); + let spans = [Span::new(100, 10), Span::new(200, 20), Span::new(300, 30)]; + let other_span = Span::new(400, 40); + + let mut txn = None; + for span in &spans { + index.insert_accounts(&mut txn, &[account], span).unwrap(); + } + index.insert_accounts(&mut txn, &[other], &other_span).unwrap(); + commit(txn); + + // Account duplicate spans are newest-first, so later execution offsets are returned first. + assert_eq!( + account_spans(&index, &account), + vec![spans[2], spans[1], spans[0]] + ); + + // Duplicates stay partitioned per account key. + assert_eq!(account_spans(&index, &other), vec![other_span]); +} diff --git a/ledger/src/tests/integration.rs b/ledger/src/tests/integration.rs new file mode 100644 index 0000000..238a6bf --- /dev/null +++ b/ledger/src/tests/integration.rs @@ -0,0 +1,549 @@ +//! End-to-end tests over the append→seal→read pipeline. +//! +//! Each test drives real [`Event`]s through the [`LedgerAppender`] and reads +//! them back through the [`LedgerReader`], both run synchronously on the test +//! thread rather than through the service pool. Transactions are genuine +//! wincode-serialized Solana transactions so the 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`; +//! that also mirrors how a caller must frame writes. + +use std::{ + ops::Range, + sync::{ + Arc, + atomic::{AtomicBool, Ordering::Acquire}, + }, +}; + +use nucleus::{ + Slot, + ledger::{Block, SuperblockSeal}, + shutdown::{Service, ShutdownManager}, + testkit::{TempDir, init_tracing, tempdir, transaction}, +}; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use solana_transaction_error::TransactionResult; +use tokio::sync::mpsc; + +use crate::{ + Ledger, + appender::LedgerAppender, + reader::LedgerReader, + request::{ + AccountSignature, AccountSignaturesParams, BlockDetails, BlockParams, BlockResponse, + ReadRequest, ReplayParams, RequestPayload, TransactionResponse, TransactionStatus, + }, + schema::{ + Balances, Event, Execution, ExecutionDetails, ExecutionHeader, OwnedBlockestoreEntry, + TransactionEntry, + }, +}; + +/// Fresh ledger on a throwaway directory; the `TempDir` must outlive the ledger. +/// +/// `size_limit` gates retention: `u64::MAX` disables it, `0` forces `truncate` +/// to run at every block boundary (the whole ledger filesystem always counts as +/// "over budget"). +fn ledger(size_limit: u64) -> (TempDir, Arc) { + init_tracing(); + let dir = tempdir(); + let ledger = Arc::new(Ledger::new(dir.path().to_owned(), size_limit).unwrap()); + (dir, ledger) +} + +/// Feeds `events` through a freshly opened appender and runs it to completion. +/// +/// The appender resumes from on-disk cursors, so successive calls model both +/// continued appends and a restart of the write side. +fn append(ledger: &Arc, events: Vec) { + let (tx, rx) = flume::bounded(events.len().max(1)); + for event in events { + tx.send(event).unwrap(); + } + drop(tx); + let mut shutdown = ShutdownManager::default(); + LedgerAppender::new(ledger.clone(), rx) + .unwrap() + .run(shutdown.handle(Service::LedgerAppender)); +} + +/// Execution metadata carrying a recognizable `fee`/`logs` for read assertions. +fn execution(signature: Signature, slot: Slot, result: TransactionResult<()>) -> Execution { + Execution { + header: ExecutionHeader { signature, result, slot }, + details: Some(ExecutionDetails { + fee: slot * 1000, + balances: Balances { pre: vec![1], post: vec![2] }, + logs: Arc::new(vec![format!("log for {slot}")]), + cpi: None, + compute_units: slot * 7, + return_data: None, + }), + } +} + +/// The `Transaction` + paired `Execution` events that record one transaction in +/// a block at `slot` (execution result `Ok`). +fn recorded(sig: Signature, payload: Arc>, slot: Slot) -> [Event; 2] { + [ + Event::Transaction(TransactionEntry { signature: sig, payload }), + Event::Execution(execution(sig, slot, Ok(()))), + ] +} + +/// Builds a request payload keeping the response receiver on the caller side. +fn request_payload(params: P) -> (RequestPayload, oneshot::Receiver) { + let (response, rx) = oneshot::channel(); + let payload = RequestPayload { + params, + response, + pending: Arc::new(AtomicBool::new(true)), + }; + (payload, rx) +} + +/// Serves a single read request on a synchronously run reader and returns its +/// response. The reader terminates as soon as the one queued request is served. +fn serve(ledger: &Arc, request: ReadRequest, rx: oneshot::Receiver) -> R { + let (tx, reader_rx) = flume::bounded(1); + tx.send(request).unwrap(); + drop(tx); + let mut shutdown = ShutdownManager::default(); + LedgerReader::new(ledger.clone(), reader_rx) + .unwrap() + .run(shutdown.handle(Service::LedgerReader)); + rx.recv().unwrap() +} + +/// Reads a full transaction by signature. +fn read_transaction(ledger: &Arc, sig: Signature) -> Option { + let (p, rx) = request_payload(sig); + serve(ledger, ReadRequest::Transaction(p), rx).unwrap() +} + +/// Reads the cheap status header for a transaction. +fn read_status(ledger: &Arc, sig: Signature) -> Option { + let (p, rx) = request_payload(sig); + serve(ledger, ReadRequest::TransactionStatus(p), rx).unwrap() +} + +/// Reads a block at the requested detail level. +fn read_block(ledger: &Arc, slot: Slot, details: BlockDetails) -> Option { + let (p, rx) = request_payload(BlockParams { slot, details }); + serve(ledger, ReadRequest::Block(p), rx).unwrap() +} + +/// Reads account history with pagination parameters. +fn read_account_signatures( + ledger: &Arc, + params: AccountSignaturesParams, +) -> Vec { + let (p, rx) = request_payload(params); + serve(ledger, ReadRequest::AccountSignatures(p), rx).unwrap() +} + +/// Signatures of the block at `slot`; panics if the read returns any other +/// variant or nothing. +fn block_signatures(ledger: &Arc, slot: Slot) -> Vec { + match read_block(ledger, slot, BlockDetails::Signatures) { + Some(BlockResponse::WithSignatures(b)) => b.signatures, + _ => panic!("expected signatures response for slot {slot}"), + } +} + +/// Reads block boundaries for a slot range in ascending slot order. +fn read_block_range(ledger: &Arc, range: Range) -> Vec { + let (p, rx) = request_payload::, _>(range); + serve(ledger, ReadRequest::BlockRange(p), rx).unwrap() +} + +/// Streams every replayed blockstore entry from `slot`'s superblock onward. +/// +/// The reader runs on its own thread while the test drains the channel, so a +/// small mpsc buffer never deadlocks the `blocking_send` in the replay loop. +fn replay(ledger: &Arc, from: Slot) -> Vec { + let (tx, mut rx) = mpsc::channel(4); + let (p, _resp) = request_payload(ReplayParams { from, tx }); + let (reader_tx, reader_rx) = flume::bounded(1); + reader_tx.send(ReadRequest::Replay(p)).unwrap(); + drop(reader_tx); + let ledger = ledger.clone(); + let worker = std::thread::spawn(move || { + let mut shutdown = ShutdownManager::default(); + LedgerReader::new(ledger, reader_rx) + .unwrap() + .run(shutdown.handle(Service::LedgerReader)); + }); + let mut entries = Vec::new(); + while let Some(entry) = rx.blocking_recv() { + entries.push(entry); + } + worker.join().unwrap(); + entries +} + +/// Appends one block at `slot` made of `count` single-account transactions, +/// each paired with its execution, and returns their signatures in order. +fn block_of(ledger: &Arc, slot: Slot, count: usize) -> Vec { + let mut events = Vec::new(); + let mut signatures = Vec::new(); + for _ in 0..count { + let (sig, payload) = transaction(&[Pubkey::new_unique()]); + events.extend(recorded(sig, payload, slot)); + signatures.push(sig); + } + events.push(Event::Block(Block { + slot, + time: slot as i64 * 100, + ..Default::default() + })); + append(ledger, events); + signatures +} + +// A transaction and its execution written in one block come back whole through +// every read surface: full transaction bytes, decompressed execution details, +// and the cheap status header — the core append→index→read roundtrip. +#[test] +fn test_transaction_roundtrip() { + let (_dir, ledger) = ledger(u64::MAX); + let account = Pubkey::new_unique(); + let (sig, bytes) = transaction(&[account]); + + append( + &ledger, + vec![ + Event::Transaction(TransactionEntry { + signature: sig, + payload: bytes.clone(), + }), + Event::Execution(execution(sig, 5, Ok(()))), + Event::Block(Block { + slot: 5, + time: 500, + ..Default::default() + }), + ], + ); + + let response = read_transaction(&ledger, sig).expect("transaction present"); + assert_eq!(response.transaction, *bytes); + assert_eq!(response.execution.header.slot, 5); + let details = response.execution.details.expect("details decompressed"); + assert_eq!(details.fee, 5000); + assert_eq!(details.logs.as_slice(), &["log for 5".to_string()]); + + let status = read_status(&ledger, sig).expect("status present"); + assert_eq!(status.slot, 5); + assert!(status.result.is_ok()); + + // An unknown signature resolves to nothing on both surfaces. + let missing = Signature::from([9; 64]); + assert!(read_transaction(&ledger, missing).is_none()); + assert!(read_status(&ledger, missing).is_none()); +} + +// A transaction stays pending until its execution arrives: sealed into a block +// without one, it is never indexed (a record is never half-written); and an +// execution whose transaction never appeared is silently dropped. +#[test] +fn test_pending_requires_execution() { + let (_dir, ledger) = ledger(u64::MAX); + let (indexed, indexed_bytes) = transaction(&[Pubkey::new_unique()]); + let (orphan, orphan_bytes) = transaction(&[Pubkey::new_unique()]); + let stray = Signature::from([3; 64]); + + append( + &ledger, + vec![ + // Paired transaction: indexed and readable. + Event::Transaction(TransactionEntry { + signature: indexed, + payload: indexed_bytes, + }), + Event::Execution(execution(indexed, 1, Ok(()))), + // Transaction with no execution: written to the blockstore but never + // indexed, so no read surface can resolve it. + Event::Transaction(TransactionEntry { + signature: orphan, + payload: orphan_bytes, + }), + // Execution with no pending transaction: dropped without error. + Event::Execution(execution(stray, 1, Ok(()))), + Event::Block(Block { slot: 1, ..Default::default() }), + ], + ); + + assert!(read_transaction(&ledger, indexed).is_some()); + assert!(read_transaction(&ledger, orphan).is_none()); + assert!(read_transaction(&ledger, stray).is_none()); +} + +// Block reads reconstruct exactly the transactions between the previous block +// boundary and this one, at every detail level — the reader derives the block's +// transaction range from the `slot - 1` boundary, so later blocks must not leak +// earlier blocks' transactions. +#[test] +fn test_block_detail_levels_partition_transactions() { + let (_dir, ledger) = ledger(u64::MAX); + let first = block_of(&ledger, 1, 2); + let second = block_of(&ledger, 2, 3); + + // Each block reports only its own transactions, in append order. + assert_eq!(block_signatures(&ledger, 1), first); + assert_eq!(block_signatures(&ledger, 2), second); + + // Transactions-only and Full carry the same count without bleed-through. + match read_block(&ledger, 2, BlockDetails::Transactions) { + Some(BlockResponse::WithTransactions(b)) => assert_eq!(b.transactions.len(), 3), + _ => panic!("expected transactions response"), + } + match read_block(&ledger, 2, BlockDetails::Full) { + Some(BlockResponse::Full(b)) => { + assert_eq!(b.transactions.len(), 3); + assert!(b.transactions.iter().all(|t| t.execution.details.is_some())); + } + _ => panic!("expected full response"), + } + + // The bare boundary carries the block metadata only. + match read_block(&ledger, 1, BlockDetails::None) { + Some(BlockResponse::Bare(block)) => assert_eq!(block.time, 100), + _ => panic!("expected bare response"), + } +} + +// 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. +#[test] +fn test_superblock_rotation_and_retention() { + // size_limit 0 makes every block boundary trigger a retention pass. + let (_dir, ledger) = ledger(0); + let old = block_of(&ledger, 1, 1); + // Seal superblock 1 and rotate to superblock 2. + append( + &ledger, + vec![Event::Superblock(SuperblockSeal { checksum: 0, transactions: 1 })], + ); + assert_eq!(ledger.meta.head(), 2); + + // The sealed superblock is still readable through the newer head. + assert!(read_transaction(&ledger, old[0]).is_some()); + + // Writing a block into the new head triggers truncation of superblock 1. + let new = block_of(&ledger, 2, 1); + assert_eq!(ledger.meta.head(), 2, "active head is never purged"); + assert!( + ledger.superblocks.read().get(&1).is_none(), + "oldest sealed segment purged" + ); + // Its transactions are gone; the head's remain. + assert!(read_transaction(&ledger, old[0]).is_none()); + assert!(read_transaction(&ledger, new[0]).is_some()); + // Retention advances the retained range past the purged superblock's end. + assert_eq!(ledger.meta.range.start.load(Acquire), 2); +} + +// A single-block read resolves a slot living in an older sealed superblock, not +// only the active head: each segment's range must be pinned to its own slots so +// a newer segment does not shadow older ones, and the ledger-wide range must +// track the tip so in-range slots pass the guard and out-of-range ones do not. +#[test] +fn test_block_read_across_superblocks() { + let (_dir, ledger) = ledger(u64::MAX); + let first = block_of(&ledger, 1, 1); + append( + &ledger, + vec![Event::Superblock(SuperblockSeal { checksum: 0, transactions: 1 })], + ); + block_of(&ledger, 2, 1); + + // Slot 1 lives in the sealed superblock; slot 2 in the head. Both resolve. + assert_eq!(block_signatures(&ledger, 1), first); + assert!(read_block(&ledger, 2, BlockDetails::None).is_some()); + // A slot past the retained tip is rejected by the ledger-wide range guard. + assert!(read_block(&ledger, 9, BlockDetails::None).is_none()); +} + +// Replay streams superblocks in on-disk order, from the one containing the +// requested slot through the active head (inclusive) so nothing committed after +// the last seal is lost on recovery. Entries come back exactly as written: +// transactions, their block delimiter, then the seal — and the unsealed head's +// entries have no trailing seal. The read is bounded by each superblock's write +// cursor, so the active head's preallocated tail is not decoded. +#[test] +fn test_replay_streams_superblocks_through_active_head() { + let (_dir, ledger) = ledger(u64::MAX); + let seal = |txs| Event::Superblock(SuperblockSeal { checksum: 0, transactions: txs }); + // Two sealed superblocks (slots 1 and 2), then an unsealed head (slot 3). + block_of(&ledger, 1, 2); + append(&ledger, vec![seal(2)]); + block_of(&ledger, 2, 1); + append(&ledger, vec![seal(3)]); + block_of(&ledger, 3, 1); + + let entries = replay(&ledger, 1); + use crate::schema::BlockstoreEntry::*; + let shape: Vec<&str> = entries + .iter() + .map(|e| match e { + Transaction(_) => "tx", + Block(_) => "block", + Superblock(_) => "seal", + }) + .collect(); + // Superblock 1 (two txns) and superblock 2 (one txn), each ending in its + // block and seal, followed by the active head (superblock 3: one txn and its + // block, no seal). + assert_eq!( + shape, + ["tx", "tx", "block", "seal", "tx", "block", "seal", "tx", "block"] + ); +} + +// Ledger state survives reopening from disk: committed transactions remain +// readable and a reopened appender resumes at the persisted cursors, appending +// a new block without clobbering the old one. +#[test] +fn test_reopen_resumes_state() { + let dir = tempdir(); + let first = { + let ledger = Arc::new(Ledger::new(dir.path().to_owned(), u64::MAX).unwrap()); + let sigs = block_of(&ledger, 1, 2); + sigs[0] + }; + + // Reopen from the same directory. + let ledger = Arc::new(Ledger::new(dir.path().to_owned(), u64::MAX).unwrap()); + assert!( + read_transaction(&ledger, first).is_some(), + "prior block survives reopen" + ); + + // A resumed appender writes a second block after the first. + let second = block_of(&ledger, 2, 1)[0]; + assert!( + read_transaction(&ledger, first).is_some(), + "old block not clobbered" + ); + assert!(read_transaction(&ledger, second).is_some()); + assert_eq!(ledger.meta.blocks.load(Acquire), 2); +} + +// A block-range read returns every block in the range, in ascending slot order, +// even when the range straddles a superblock boundary. The reader walks slots +// descending across superblocks newest-first, so a boundary slot must be handed +// to the older segment instead of being consumed against the newer one. +#[test] +fn test_block_range_spans_superblocks() { + let (_dir, ledger) = ledger(u64::MAX); + // Slot 1 lands in superblock 1; the seal rotates slots 2 and 3 into + // superblock 2, so any range over 1..=2 crosses the segment boundary. + block_of(&ledger, 1, 1); + append( + &ledger, + vec![Event::Superblock(SuperblockSeal { checksum: 0, transactions: 1 })], + ); + block_of(&ledger, 2, 1); + block_of(&ledger, 3, 1); + + let slots = |range: Range| { + read_block_range(&ledger, range).iter().map(|b| b.slot).collect::>() + }; + // The full range comes back once each, in order, across the boundary. + assert_eq!(slots(1..4), vec![1, 2, 3]); + // A sub-range inside one segment returns only its blocks. + assert_eq!(slots(2..3), vec![2]); + // A tail past the retained tip yields the retained blocks without dropping + // the boundary slot. + assert_eq!(slots(1..9), vec![1, 2, 3]); +} + +// The account index keeps one entry per touching transaction, excludes unrelated +// transactions, and account-signature history pages newest-superblock first with +// exclusive `before`/`until` cursors. +#[test] +fn test_account_signatures_history_pagination_and_ordering() { + let (_dir, ledger) = ledger(u64::MAX); + let account = Pubkey::new_unique(); + + let block_for = |slot, count, unrelated| { + let mut events = Vec::new(); + let mut sigs = Vec::new(); + for _ in 0..count { + let (sig, payload) = transaction(&[account]); + events.extend(recorded(sig, payload, slot)); + sigs.push(sig); + } + if unrelated { + let (sig, payload) = transaction(&[Pubkey::new_unique()]); + events.extend(recorded(sig, payload, slot)); + } + events.push(Event::Block(Block { + slot, + time: slot as i64 * 100, + ..Default::default() + })); + append(&ledger, events); + sigs + }; + + // Two transactions touch `account` in superblock 1; one does not. + let sb1 = block_for(1, 2, true); + append( + &ledger, + vec![Event::Superblock(SuperblockSeal { checksum: 0, transactions: 3 })], + ); + let sb2 = block_for(2, 2, false); + + let read_history = |pubkey, limit, before, until| { + read_account_signatures( + &ledger, + AccountSignaturesParams { pubkey, limit, before, until }, + ) + }; + let signatures = |history: &[AccountSignature]| { + history.iter().map(|s| s.signature).collect::>() + }; + + let expected = vec![sb2[1], sb2[0], sb1[1], sb1[0]]; + let full = read_history(account, 10, None, None); + assert_eq!(signatures(&full), expected); + assert!(full.iter().all(|s| s.blocktime == s.slot as i64 * 100)); + + assert_eq!( + signatures(&read_history(account, 2, None, None)), + expected[..2], + "`limit` caps results" + ); + assert!( + read_history(Pubkey::new_unique(), 10, None, None).is_empty(), + "unmentioned account has no history" + ); + + // Newest-superblock first, newest execution first within each superblock. + assert_eq!( + full.iter().map(|s| s.slot).collect::>(), + vec![2, 2, 1, 1], + "newest account history first" + ); + + let history_signatures = |before, until| signatures(&read_history(account, 10, before, until)); + + // `before` is an exclusive upper bound. The newest cursor yields every + // older signature across the superblock boundary in newest-to-oldest order. + assert_eq!(history_signatures(Some(sb2[1]), None), expected[1..]); + // The oldest cursor yields nothing, separating an exclusive bound from an + // inclusive one. + assert!(history_signatures(Some(sb1[0]), None).is_empty()); + + // `until` stops at, and excludes, its signature: the first match yields + // nothing, the last yields everything above it. + assert!(history_signatures(None, Some(expected[0])).is_empty()); + assert_eq!(history_signatures(None, Some(expected[3])), expected[..3]); +} diff --git a/ledger/src/tests/mod.rs b/ledger/src/tests/mod.rs new file mode 100644 index 0000000..794fc57 --- /dev/null +++ b/ledger/src/tests/mod.rs @@ -0,0 +1,7 @@ +//! Ledger test modules. +//! +//! `index` covers the LMDB codec/index in isolation; `integration` drives the +//! append→seal→read pipeline end to end through the appender and reader. + +mod index; +mod integration;