From 260b773e5fad0a90437290ff1034def9cfea3391 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 17 Apr 2026 18:39:34 +0400 Subject: [PATCH] feat: add optimized accountsdb crate --- Cargo.toml | 13 + README.md | 327 +++++++++++++++++++++ accountsdb/Cargo.toml | 41 +++ accountsdb/README.md | 65 +++++ accountsdb/src/lib.rs | 269 ++++++++++++++++++ accountsdb/src/metrics.rs | 216 ++++++++++++++ accountsdb/src/snapshot.rs | 102 +++++++ accountsdb/src/store/defrag.rs | 133 +++++++++ accountsdb/src/store/index.rs | 204 +++++++++++++ accountsdb/src/store/kv.rs | 123 ++++++++ accountsdb/src/store/mmap.rs | 297 +++++++++++++++++++ accountsdb/src/store/mod.rs | 292 +++++++++++++++++++ accountsdb/src/tests.rs | 489 ++++++++++++++++++++++++++++++++ accountsdb/src/volatile.rs | 121 ++++++++ solana/account/src/cow/tests.rs | 83 ------ 15 files changed, 2692 insertions(+), 83 deletions(-) create mode 100644 README.md create mode 100644 accountsdb/Cargo.toml create mode 100644 accountsdb/README.md create mode 100644 accountsdb/src/lib.rs create mode 100644 accountsdb/src/metrics.rs create mode 100644 accountsdb/src/snapshot.rs create mode 100644 accountsdb/src/store/defrag.rs create mode 100644 accountsdb/src/store/index.rs create mode 100644 accountsdb/src/store/kv.rs create mode 100644 accountsdb/src/store/mmap.rs create mode 100644 accountsdb/src/store/mod.rs create mode 100644 accountsdb/src/tests.rs create mode 100644 accountsdb/src/volatile.rs delete mode 100644 solana/account/src/cow/tests.rs diff --git a/Cargo.toml b/Cargo.toml index a4b4c44..b793825 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,6 @@ [workspace] members = [ + "accountsdb", "nucleus", "programs/magic-root-interface", "programs/magic-root-program", @@ -22,6 +23,7 @@ rust-version = "1.94.1" version = "0.1.0" [workspace.dependencies] +accountsdb = { path = "accountsdb" } magic-root-interface = { path = "programs/magic-root-interface" } magic-root-program = { path = "programs/magic-root-program" } nucleus = { path = "nucleus", package = "magicblock-engine-nucleus" } @@ -35,18 +37,28 @@ arc-swap = "1.9.1" assert_matches = "1.5.0" base64 = "0.22.1" bincode = "1.3.3" +bitcode = "0.6.9" bitflags = "2.11.1" blake3 = "1.8.5" +bytemuck = { version = "1.25", features = ["derive", "extern_crate_std"] } cfg-if = "1.0.4" +clonetree = "0.0.2" criterion = "0.7.0" derive_more = "2.1.1" env_logger = "0.11.8" futures = { version = "0.3.32", default-features = false } +heed = { version = "0.22.1", default-features = false } itertools = "0.14.0" log = "0.4.29" +memmap2 = "0.9.10" +num_cpus = "1.17.0" +oneshot = "0.2.1" +parking_lot = "0.12.5" +prometheus = { version = "0.14.0", default-features = false } qualifier_attr = "0.2.2" rand = "0.9.2" rustix = { version = "1.1.4" } +scc = "3.8.4" serde = "1.0.228" serde_bytes = "0.11.19" tar = "0.4.45" @@ -56,6 +68,7 @@ tokio = "1.52.1" tokio-util = "0.7.18" tracing = "0.1.44" tracing-subscriber = { version = "0.3.23", features = ["env-filter", "fmt"] } +twox-hash = { version = "2.1.2", default-features = false } wincode = "0.5.1" zstd = { version = "0.13.3", default-features = false } diff --git a/README.md b/README.md new file mode 100644 index 0000000..1789b79 --- /dev/null +++ b/README.md @@ -0,0 +1,327 @@ +# MagicBlock Engine + +A durable Solana execution engine for ephemeral rollups. + +MagicBlock Engine runs Solana transactions, owns account state, records execution +results, and gives callers a small async API for accounts, transactions, blocks, +and subscriptions. + +At the top of the stack is `engine::Engine`. A caller starts it once, then uses +the handle to mutate accounts, submit transactions, simulate execution, stream +updates, and query retained ledger state. + +```rust +use std::{num::NonZeroU64, path::PathBuf, sync::Arc, time::Duration}; + +use engine::Engine; +use keeper::builder::{AccountsDBParams, BlockstoreParams, KeeperBuilder, LedgerParams}; +use nucleus::shutdown::ShutdownManager; +use solana_keypair::Keypair; +use solana_sysvar::rent::Rent; + +async fn open_engine(home: PathBuf) -> engine::Result { + let mut shutdown = ShutdownManager::default(); + let authority = Arc::new(Keypair::new()); + + let builder = KeeperBuilder { + authority, + accountsdb: AccountsDBParams { + directory: home.join("accountsdb"), + }, + ledger: LedgerParams { + directory: home.join("ledger"), + size_limit: 256 * 1024 * 1024 * 1024, + }, + blockstore: BlockstoreParams { + blocktime: Duration::from_millis(400), + superblock: NonZeroU64::new(16).unwrap(), + }, + builtins: Default::default(), + programs: Default::default(), + accounts: Default::default(), + rent: Rent::default(), + }; + + Engine::new(builder, None, &mut shutdown).await +} +``` + +## Cooperative Shutdown + +The caller owns the shutdown lifecycle. `Engine::new` receives a +`ShutdownManager` so the engine can register its background services with one +ordered coordinator, but the embedding service keeps the manager and drives it +from the outside. + +`shutdown.wait().await` returns when the process receives SIGTERM or Ctrl-C, an +engine service requests shutdown internally, or a registered service terminates +early. After that, the host should stop accepting outside work, release or flush +its engine handle, and then call `shutdown.terminate().await` to drain the +engine's services by tier. + +```rust +use engine::Engine; +use nucleus::shutdown::ShutdownManager; +use tokio::sync::oneshot; + +async fn run_service( + engine: Engine, + mut shutdown: ShutdownManager, + stop_ingress: oneshot::Sender<()>, +) -> engine::Result<()> { + shutdown.wait().await; + + let _ = stop_ingress.send(()); + engine.flush()?; + drop(engine); + + shutdown.terminate().await; + Ok(()) +} +``` + +Shutdown is cooperative and ordered. The pacemaker is cancelled first so no new +block boundaries are produced. The sequencer is cancelled next and drains +in-flight work before joining its executors. Ledger, simulator, subscription +cleanup, and other backing workers are cancelled last. Each tier gets a bounded +window to report why it stopped before the manager moves on. + +Inside engine-owned services, registration is explicit: the manager hands the +service a `ShutdownHandle`, the service waits on its tier token, and it reports +why it stopped. + +```rust +use nucleus::shutdown::{Service, ShutdownManager, ShutdownReason}; + +fn spawn_cleanup(shutdown: &mut ShutdownManager) { + let mut handle = shutdown.handle(Service::SubscriptionsCleanup); + + tokio::spawn(async move { + handle.signalled().await; + // Drop subscriptions, release timers, or flush local cleanup state here. + handle.terminate(ShutdownReason::Signalled); + }); +} +``` + +External services integrate at the supervisor boundary today: stop their ingress +or worker tasks before calling `terminate()`, rather than reusing engine-owned +`Service` labels as external names. If an external service needs to be registered +into the same ordered drain later, add that service explicitly to the shutdown +API when the integration owns a real ordering requirement. + +## Account Model + +The engine is organized around one question: who owns an account right now? + +Mutable accounts are under the rollup's exclusive control. The engine is the +authoritative owner, so the account is persisted on disk and survives restart. +Read-only accounts are owned by chain or service state. The engine can always +re-fetch them, so they live in volatile memory. + +That distinction is dynamic. A transaction can delegate an account into the +rollup, hand it back, or create state that only exists inside the rollup. The +storage layer routes those transitions without leaving stale copies behind. + +## Ergonomic Account Operations + +Account mutation is privileged, so callers do not hand-write storage mutations. +`Engine::account(pubkey)` builds MagicRoot instruction sequences, signs them with +the engine authority, submits them, and waits for the committed result. + +```rust +use engine::Engine; +use solana_account::{AccountBuilder, AccountFieldPatch, AccountMode, OwnedAccount, ReadableAccount}; +use solana_pubkey::Pubkey; + +async fn account_flow(engine: &Engine, owner: Pubkey) -> engine::Result<()> { + let pubkey = Pubkey::new_unique(); + + let account = AccountBuilder::default() + .lamports(1_000_000) + .owner(owner) + .mode(AccountMode::Delegated) + .data(b"hello engine".to_vec()) + .build(); + + engine.account(pubkey).create(account, None).await?; + + let current = engine.accounts().get(&pubkey)?.expect("account exists"); + assert_eq!(current.data(), b"hello engine"); + + engine + .account(pubkey) + .patch(vec![ + AccountFieldPatch::Lamports(2_000_000), + AccountFieldPatch::DataAt { + offset: 6, + data: b"rollup".to_vec(), + }, + ]) + .await?; + + let replacement = AccountBuilder::default() + .lamports(3_000_000) + .owner(owner) + .mode(AccountMode::Ephemeral) + .data(b"local-only state".to_vec()) + .build(); + + engine.account(pubkey).update(replacement).await?; + engine.account(pubkey).delete().await?; + + Ok(()) +} +``` + +Callers that need chain-owned accounts can coordinate cache misses +with `engine.accounts().ensure()`. Stored accounts are skipped. The first caller +to see a missing account owns the load and calls `commit()` after storing it; +concurrent callers wait on the same pubkey. + +```rust +use std::future::Future; + +use engine::Engine; +use keeper::{MissingAccount, error::KeeperError}; +use solana_account::AccountSharedData; +use solana_pubkey::Pubkey; + +async fn ensure_accounts( + engine: &Engine, + pubkeys: &[Pubkey], + mut fetch_account: F, +) -> engine::Result<()> +where + F: FnMut(Pubkey) -> Fut, + Fut: Future>>, +{ + let accounts = engine.accounts(); + + for missing in accounts.ensure(pubkeys) { + match missing { + MissingAccount::Load(load) => { + if let Some(account) = fetch_account(load.pubkey).await? { + engine.account(load.key).create(account, None).await?; + load.commit(); + } + } + MissingAccount::Wait(wait) => { + wait.wait().await; + } + } + } + + Ok(()) +} +``` + +## Transactions + +The transaction API accepts the shape callers already have. A slice of +instructions becomes a signed transaction using the engine authority and latest +blockhash. A `Message` works the same way. A pre-sanitized `TransactionView` can +also be submitted directly. + +```rust +use engine::Engine; +use solana_instruction::{AccountMeta, Instruction}; +use solana_pubkey::Pubkey; + +async fn run_instruction(engine: &Engine, program_id: Pubkey, account: Pubkey) -> engine::Result<()> { + let ix = Instruction { + program_id, + accounts: vec![AccountMeta::new(account, false)], + data: vec![1, 2, 3, 4], + }; + + engine.transaction(&[ix])?.execute().await?.map_err(Into::into) +} +``` + +Execution commits account changes and records a status receipt. Simulation runs +against current state on owned account copies and returns the execution record +without committing anything. Scheduling queues work without waiting for the +result. + +```rust +use engine::Engine; +use solana_instruction::Instruction; + +async fn transaction_modes(engine: &Engine, instructions: &[Instruction]) -> engine::Result<()> { + let simulated = engine.transaction(instructions)?.simulate().await?; + if simulated.is_ok() { + engine.transaction(instructions)?.schedule().await?; + } + + Ok(()) +} +``` + +## Live Subscriptions + +Reads and subscriptions come from the keeper namespaces exposed by `Engine`. +They are plain broadcast streams for state that is already being committed: +accounts, program-owned accounts, transaction statuses, logs, processed +transactions, and blocks. + +```rust +use engine::Engine; +use keeper::TransactionView; +use solana_account::ReadableAccount; +use solana_pubkey::Pubkey; + +async fn watch(engine: &Engine, account: Pubkey, program: Pubkey) { + let accounts = engine.accounts().subscribe(account).await; + let program_accounts = engine.accounts().subscribe_program(program).await; + let logs = engine.transactions().subscribe_logs(account).await; + let processed = engine.transactions().subscribe_processed(); + let blocks = engine.blocks().subscribe(); + +} + +``` + +## Startup And Recovery + +A normal restart opens persisted state. The engine does not replay the full +ledger when accountsdb is already current. + +Replay is the corruption-recovery path. If accountsdb opens corrupt, keeper +restores the newest archived snapshot from retained superblocks. That can leave +accountsdb behind the ledger tip, so engine replay re-executes retained ledger +entries from the restored slot forward. At each sealed superblock it quiesces the +replay sequencer and compares the reconstructed account checksum with the sealed +checksum. A mismatch is `ReplayError::StateMismatch`. + +Volatile state can be dropped at runtime because it mirrors chain-owned accounts. +Persisted rollup-exclusive state is left intact. + +## Crates + +The workspace is layered bottom to top. Each crate knows only about the layers +below it. + +- `nucleus` holds shared primitives, ledger schema, heed helpers, runtime + barriers, and cooperative shutdown. +- `solana/*` contains the Solana forks reshaped around the engine's account + representation and execution model. +- `accountsdb` routes accounts between persisted and volatile storage and owns + snapshots, backup, and defrag. +- `ledger` records transactions, block boundaries, and execution metadata in + self-contained superblocks. +- `keeper` opens durable state, owns read-side caches, seeds startup state, and + fans out subscriptions. +- `processor` schedules non-conflicting transactions across SVM executors and + commits results through keeper. +- `engine` is the consumer-facing handle that wires keeper, processor, replay, + pacemaking, account mutation, and transaction submission together. + +## Transaction Flow + +Inbound transactions reach the processor, which resolves conflicts by account +locks and fans non-conflicting work across its executor pool. Executors run the +transaction through SVM, reading accounts through keeper/accountsdb. Results are +appended to the ledger, dirty accounts are written back, and subscriptions are +published. When a superblock seals, keeper snapshots accountsdb and archives it +beside the sealed ledger segment. diff --git a/accountsdb/Cargo.toml b/accountsdb/Cargo.toml new file mode 100644 index 0000000..55bb871 --- /dev/null +++ b/accountsdb/Cargo.toml @@ -0,0 +1,41 @@ +[package] +name = "magicblock-accountsdb" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "accountsdb" + +[dependencies] +nucleus = { workspace = true, features = ["heed", "metrics"] } + +ahash = { workspace = true } +bincode = { workspace = true } +bytemuck = { workspace = true } +clonetree = { workspace = true } +derive_more = { workspace = true, features = ["from"] } +heed = { workspace = true } +memmap2 = { workspace = true } +parking_lot = { workspace = true } +scc = { workspace = true, features = ["serde"] } +serde = { workspace = true, features = ["derive"] } +thiserror = { workspace = true } +tracing = { workspace = true } +twox-hash = { workspace = true, features = ["alloc", "xxhash3_64"] } + +solana-account = { workspace = true, features = ["serde"] } +solana-pubkey = { workspace = true, features = ["bytemuck"] } +solana-sysvar = { workspace = true } + +[dev-dependencies] +nucleus = { workspace = true, features = ["testkit"] } +assert_matches = { workspace = true } + +[lints] +workspace = true diff --git a/accountsdb/README.md b/accountsdb/README.md new file mode 100644 index 0000000..88e859c --- /dev/null +++ b/accountsdb/README.md @@ -0,0 +1,65 @@ +# `magicblock-accountsdb` + +This crate is where the engine's central invariant — *persisted vs. volatile, +keyed on current mutability* — actually lives. It answers two questions for every +account: which backend holds it, and what happens when its mutability changes. + +The reasoning is simple once you accept the premise. A **mutable** account is one +the rollup controls exclusively; the engine is its authoritative owner, so its +copy has to survive a restart. That copy goes in the **`PersistedStore`** — an +mmap'd file plus an LMDB index. An **immutable** account is owned by the chain or +a service; the engine only reads it and can always re-fetch it, so durability +would be wasted effort. It goes in the **`VolatileStore`**, in memory. + +The interesting case is the flip. When an account changes mutability, it has to +move to the other store — but the *old* copy is now stale and must not linger, +or a later read could resurrect dead state. So `AccountsDB` writes through to the +opposite backend on a flip purely to **evict** the stale copy. That write-through +is the eviction; there is no separate cleanup pass to forget. + +## Storage layout + +The persisted data is a single mapped `storage.db`. A meta header sits in front +(the current write cursor and the file size), followed by account images in the +borrowed layout from `solana-account`. Each image is prefixed with its full +pubkey, so a plain scan of the file can recover every key even without the index. + +The index, under `/index`, holds three tables: + +- `accounts` — key tag → offset + owner tag (the primary lookup) +- `programs` — owner tag → offset (find all accounts owned by a program) +- `freelist` — image size → reusable offset (so freed slots get reused) + +## Invariants + +These hold because the layout and the defrag logic depend on them; breaking one +corrupts the store rather than just slowing it down. + +- Persisted images are 8-byte aligned, and offsets are counted in `StorageUnit`, + not raw bytes. +- The mmap header reservation always stays in front of the images. +- Defrag sweeps the tail half of the hole list each pass, so repeated runs shrink + the remaining work geometrically rather than rescanning everything. +- `PersistedProgramIter` keeps its read transaction open for the whole iteration, + so the snapshot it walks stays consistent. + +## Batches & snapshots + +A persisted batch commits once, at the end. If the LMDB commit fails, the borrowed +images are rolled back in memory so the logical state matches what's on disk — +the batch is all-or-nothing. + +Snapshots clone the active database tree for a slot. They rely on the +no-concurrent-writes rule: the caller holds exclusive write access, persisted +state is flushed first, and the cloned volatile store is swapped for the current +in-memory one. Without that exclusivity the clone could capture a half-written +state, which is why the caller — not this crate — has to guarantee it. + +## Resetting volatile state + +The split is asymmetric on purpose: volatile state can be dropped, persisted state +cannot. `reset` clears the volatile store and leaves the persisted store untouched. +This is the recovery move for when the volatile mirror can no longer be trusted to +match chain — for example the chain-sync connection drops — since those accounts are +chain-owned and can simply be re-fetched. The persisted, ER-exclusive copy is +authoritative, so there's nothing to re-fetch and nothing to reset. diff --git a/accountsdb/src/lib.rs b/accountsdb/src/lib.rs new file mode 100644 index 0000000..1fbd3d1 --- /dev/null +++ b/accountsdb/src/lib.rs @@ -0,0 +1,269 @@ +#![doc = include_str!("../README.md")] + +use std::{ + cell::RefCell, + collections::BTreeSet, + path::{Path, PathBuf}, + sync::atomic::Ordering::*, +}; + +use derive_more::From; +use nucleus::Slot; +use nucleus::heed::RoTxnTls; +use solana_account::{AccountSharedData, CoWAccount}; +use solana_pubkey::Pubkey; +use tracing::{info, warn}; + +use crate::{ + store::{DatabaseVersion, PersistedProgramIter, PersistedStore}, + volatile::VolatileStore, +}; + +pub use snapshot::{BackupOp, SnapshotError, SnapshotResult}; + +mod metrics; +mod snapshot; +mod store; +mod volatile; + +#[cfg(test)] +mod tests; + +/// Active database subdirectory. +const ACTIVE_DIR: &str = "CURRENT"; + +/// Top-level account store backed by persisted and volatile backends. +pub struct AccountsDB { + /// Authoritative on-disk store for mutable accounts. + persisted: PersistedStore, + /// Authoritative in-memory store for immutable accounts. + volatile: VolatileStore, + /// Database root directory. + root: PathBuf, +} + +impl AccountsDB { + /// Opens or creates the database at `root`. + pub fn new(root: impl AsRef) -> Result { + let root = root.as_ref().to_owned(); + let path = Self::path(&root); + let persisted = PersistedStore::new(&path)?; + let volatile = VolatileStore::new(&path)?; + info!(?path, "opened accountsdb"); + let db = Self { persisted, volatile, root }; + metrics::init(&db); + Ok(db) + } + + /// Returns the active database directory under `root`. + pub fn path(root: &Path) -> PathBuf { + root.join(ACTIVE_DIR) + } + + /// Stores accounts in the backend that matches their current form. + /// + /// Mutable accounts are kept in persisted storage. Immutable accounts are + /// kept in volatile storage. Each batch also touches the opposite backend + /// so stale copies are removed after mode changes. Persisted failures roll + /// back borrowed images before the caller sees the error. + pub fn store<'a, AC>(&self, accounts: AC) -> Result<()> + where + AC: IntoIterator + Clone, + ::IntoIter: Clone, + { + let iter = accounts.clone().into_iter().filter(persisted); + self.persisted.upsert(iter)?; + + let iter = accounts.into_iter().filter(volatile); + self.volatile.upsert(iter); + + Ok(()) + } + + /// Creates a loader that reuses a read transaction for persisted lookups. + pub fn loader(&self) -> AccountLoader<'_> { + AccountLoader::new(self) + } + + /// Iterates program-owned accounts across both backends. + pub fn program(&self, owner: &Pubkey) -> Result> { + let persisted = self.persisted.program(*owner)?; + let volatile = self.volatile.program(owner); + Ok(ProgramIter { persisted, volatile, db: self }) + } + + /// Returns the latest slot persisted in the database metadata. + pub fn slot(&self) -> Slot { + self.persisted.meta().slot.load(Acquire) + } + + /// Sets the database slot and flushes dirty pages asynchronously. + pub fn set_slot(&self, slot: Slot) -> Result<()> { + self.persisted.meta().slot.store(slot, Release); + self.flush(false) + } + + /// Flushes persisted account storage, forcing synchronous durability when requested. + pub fn flush(&self, force: bool) -> Result<()> { + self.persisted.flush(force).map_err(Into::into) + } + + /// Validates the persisted store checksum and on-disk format version. + pub fn validate(&self) -> Result<()> { + self.persisted.validate() + } + + /// Returns the last checksum published on superblock boundary. + pub fn checksum(&self) -> u64 { + self.persisted.meta().checksum.load(Acquire) + } + + /// Drops all volatile account state; persisted state is left untouched. + /// + /// Volatile accounts are a mirror of chain-owned state the engine can + /// always re-fetch, so a caller that can no longer guarantee the mirror is in + /// sync with chain (e.g. the chain-sync connection dropped) can reset it and let + /// it be rebuilt from chain. Persisted, ER-exclusive state is authoritative and + /// is never reset. + pub fn reset(&self) { + self.volatile.reset(); + } +} + +/// Loader that caches a read transaction for persisted account lookups. +pub struct AccountLoader<'a> { + /// Cached read transaction for the persisted index. + txn: RefCell>>, + /// Database handle used for volatile and persisted lookups. + db: &'a AccountsDB, +} + +impl<'a> AccountLoader<'a> { + /// Creates a new loader bound to `db`. + pub fn new(db: &'a AccountsDB) -> Self { + Self { txn: Default::default(), db } + } + + /// Loads one account, checking both backends in turn + pub fn load(&self, pubkey: &Pubkey) -> Result> { + let txn = &mut self.txn.borrow_mut(); + if let Some(acc) = self.db.persisted.load(txn, pubkey)? { + metrics::load(StoreKind::Persisted); + return Ok(Some(acc.into())); + } + let account = self.db.volatile.load(pubkey).map(Into::into); + if account.is_some() { + metrics::load(StoreKind::Volatile); + } + Ok(account) + } + + /// Returns whether an account exists in either backend. + pub fn contains(&self, pubkey: &Pubkey) -> Result { + let txn = &mut self.txn.borrow_mut(); + if self.db.persisted.contains(txn, pubkey)? { + return Ok(true); + } + let contains = self.db.volatile.contains(pubkey); + Ok(contains) + } +} + +/// Iterates program-owned accounts across both backends. +pub struct ProgramIter<'a> { + /// Persisted program accounts. + persisted: Option>, + /// Volatile program pubkeys. + volatile: BTreeSet, + /// Database handle used to resolve volatile accounts. + db: &'a AccountsDB, +} + +impl<'a> Iterator for ProgramIter<'a> { + type Item = AccountEntry; + /// Yields persisted mutable accounts first, then volatile immutable ones. + fn next(&mut self) -> Option { + if let Some(persisted) = &mut self.persisted { + // Yield persisted entries first; mutable accounts live there. + if let Some(item) = persisted.next() { + return Some(item); + } + } + // Release the persisted read txn before draining volatile entries. + let _ = self.persisted.take(); + // Then drain the in-memory set of immutable accounts. + while let Some(pubkey) = self.volatile.pop_first() { + if let Some(account) = self.db.volatile.load(&pubkey) { + return Some((pubkey, account.into())); + } + warn!(%pubkey, "volatile program set references a missing account; skipping"); + } + None + } +} + +/// Errors returned by accountsdb. +#[derive(Debug, thiserror::Error, From)] +pub enum AccountsDBError { + /// LMDB key-value codec error. + #[error("LMDB key/value codec error: {0}")] + Codec(#[source] heed::BoxedError), + /// Filesystem error. + #[error("filesystem I/O error: {0}")] + IO(#[source] std::io::Error), + /// LMDB index access error. + #[error("LMDB index error: {0}")] + Index(#[source] heed::Error), + /// Storage allocation would exceed the maximum mapped size. + #[error("mapped storage exceeded the 32 GiB limit")] + Allocation, + /// Opened database version is not supported by current implementation. + #[error("unsupported database version: {0:?}")] + UnsupportedVersion(DatabaseVersion), + /// Database was corrupted during the shutdown/crash. + #[error("database integrity check failed")] + Corruption, + /// Volatile snapshot serialization error. + #[error("volatile snapshot serialization error: {0}")] + Serde(#[source] Box), +} + +/// Result type used by the accountsdb crate. +type Result = std::result::Result; +/// Account key plus shared account payload. +pub type AccountEntry = (Pubkey, AccountSharedData); + +/// Accountsdb backend identity. +#[derive(Clone, Copy)] +pub(crate) enum StoreKind { + /// Mmap-backed persisted storage. + Persisted, + /// In-memory volatile storage. + Volatile, +} + +impl StoreKind { + /// Returns the Prometheus label value for this backend. + pub(crate) fn label(self) -> &'static str { + match self { + StoreKind::Persisted => "persisted", + StoreKind::Volatile => "volatile", + } + } +} + +/// Returns `true` for entries that must touch persisted storage. +fn persisted(entry: &&AccountEntry) -> bool { + match entry.1.cow() { + CoWAccount::Borrowed(_) => true, + CoWAccount::Owned(_) => entry.1.mutable(), + } +} + +/// Returns `true` for entries that must touch volatile storage. +fn volatile(entry: &&AccountEntry) -> bool { + match entry.1.cow() { + CoWAccount::Borrowed(_) => !entry.1.mutable(), + CoWAccount::Owned(_) => true, + } +} diff --git a/accountsdb/src/metrics.rs b/accountsdb/src/metrics.rs new file mode 100644 index 0000000..b759b47 --- /dev/null +++ b/accountsdb/src/metrics.rs @@ -0,0 +1,216 @@ +//! Prometheus metrics for accountsdb. + +use std::sync::{OnceLock, atomic::Ordering::*}; + +use nucleus::metrics as metric; +use nucleus::metrics::{ + IntCounter, IntCounterVec, IntGaugeVec, MetricOperation, MetricSpec, OperationCounters, +}; + +use crate::{AccountsDB, StoreKind, store::Stats}; + +/// Process-wide accountsdb metrics registered in the default Prometheus registry. +static METRICS: OnceLock = OnceLock::new(); + +/// Persisted account image load counter. +const READS: MetricSpec = MetricSpec { + name: "accountsdb_persisted_reads", + help: "Persisted account image loads.", +}; +/// Borrowed account commit counter. +const COMMITS: MetricSpec = MetricSpec { + name: "accountsdb_persisted_commits", + help: "Borrowed account commits into persisted storage.", +}; +/// Fresh mapped-storage allocation counter. +const ALLOCS: MetricSpec = MetricSpec { + name: "accountsdb_persisted_allocs", + help: "Fresh allocations from the mapped persisted storage file.", +}; +/// Persisted freelist reuse counter. +const REALLOCS: MetricSpec = MetricSpec { + name: "accountsdb_persisted_reallocs", + help: "Allocations reused from the persisted freelist.", +}; +/// Defragmentation relocation counter. +const COMPACTIONS: MetricSpec = MetricSpec { + name: "accountsdb_persisted_compactions", + help: "Persisted account relocations during defragmentation.", +}; +/// Persisted account removal counter. +const REMOVALS: MetricSpec = MetricSpec { + name: "accountsdb_persisted_removals", + help: "Persisted account removals.", +}; +/// Persisted storage resize counter. +const RESIZES: MetricSpec = MetricSpec { + name: "accountsdb_persisted_resizes", + help: "Persisted storage file resizes.", +}; +/// Account load counter grouped by backend store. +const LOADS: MetricSpec = MetricSpec { + name: "accountsdb_loads", + help: "Account loads by backend store.", +}; +/// Operation latency histogram recorded in microseconds. +const OPERATION_TIME: MetricSpec = MetricSpec { + name: "accountsdb_operation_duration_micros", + help: "Accountsdb operation duration distribution in microseconds.", +}; +/// Account count gauge grouped by backend store. +const ACCOUNTS: MetricSpec = MetricSpec { + name: "accountsdb_accounts", + help: "Current accountsdb account count by backend store.", +}; + +/// Label used to separate persisted and volatile account counts. +const STORE_LABEL: &str = "store"; + +/// Accountsdb operation used as a low-cardinality operation label. +#[derive(Clone, Copy)] +pub(crate) enum Operation { + /// Persisted store flush path. + Flush, + /// Persisted checksum path. + Checksum, + /// Accountsdb snapshot path. + Snapshot, + /// Persisted store defragmentation path. + Defragmentation, +} + +impl MetricOperation for Operation { + /// Returns the Prometheus label value for this operation. + fn label(self) -> &'static str { + match self { + Operation::Flush => "flush", + Operation::Checksum => "checksum", + Operation::Snapshot => "snapshot", + Operation::Defragmentation => "defragmentation", + } + } +} + +/// Registers accountsdb metrics once, seeding durable counters from persisted stats. +pub(crate) fn init(db: &AccountsDB) { + METRICS.get_or_init(|| Metrics::new(db.persisted.storage.stats())); +} + +/// Records one persisted account image load. +pub(crate) fn read() { + metric::with_metrics(&METRICS, |m| m.reads.inc()); +} + +/// Records one borrowed account commit into persisted storage. +pub(crate) fn commit() { + metric::with_metrics(&METRICS, |m| m.commits.inc()); +} + +/// Records one fresh allocation from the mapped persisted storage file. +pub(crate) fn alloc() { + metric::with_metrics(&METRICS, |m| m.allocs.inc()); +} + +/// Records one allocation reuse from the persisted freelist. +pub(crate) fn realloc() { + metric::with_metrics(&METRICS, |m| m.reallocs.inc()); +} + +/// Records one persisted account relocation during defragmentation. +pub(crate) fn compaction() { + metric::with_metrics(&METRICS, |m| m.compactions.inc()); +} + +/// Records one persisted account removal. +pub(crate) fn removal() { + metric::with_metrics(&METRICS, |m| m.removals.inc()); +} + +/// Records one persisted storage file resize. +pub(crate) fn resize() { + metric::with_metrics(&METRICS, |m| m.resizes.inc()); +} + +/// Refreshes the current account count for `store`. +pub(crate) fn accounts(store: StoreKind, count: u64) { + let count = i64::try_from(count).unwrap_or(i64::MAX); + metric::with_metrics(&METRICS, |m| { + m.accounts.with_label_values(&[store.label()]).set(count); + }); +} + +/// 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)) +} + +/// Records one account load satisfied by `store`. +pub(crate) fn load(store: StoreKind) { + metric::with_metrics(&METRICS, |m| m.loads[store as usize].inc()); +} + +/// Owns all Prometheus collectors registered by accountsdb. +struct Metrics { + /// Durable persisted account image load counter. + reads: IntCounter, + /// Durable borrowed account commit counter. + commits: IntCounter, + /// Durable fresh allocation counter. + allocs: IntCounter, + /// Durable freelist reuse counter. + reallocs: IntCounter, + /// Durable defragmentation relocation counter. + compactions: IntCounter, + /// Durable persisted account removal counter. + removals: IntCounter, + /// Durable persisted storage resize counter. + resizes: IntCounter, + /// Registered load counter labeled by backend store; backs `loads`. + loads_vec: IntCounterVec, + /// Per-`StoreKind` load counters pre-resolved from `loads_vec`. + loads: [IntCounter; 2], + /// Runtime operation duration and completion counters. + operations: OperationCounters, + /// Runtime account count gauge labeled by backend store. + accounts: IntGaugeVec, +} + +impl Metrics { + /// Builds collectors and seeds durable counters from persisted mmap stats. + fn new(stats: &Stats) -> Self { + let loads_vec = metric::counter_vec(LOADS, &[STORE_LABEL]); + let loads = [ + loads_vec.with_label_values(&[StoreKind::Persisted.label()]), + loads_vec.with_label_values(&[StoreKind::Volatile.label()]), + ]; + let metrics = Self { + reads: metric::counter(READS, stats.reads.load(Relaxed)), + commits: metric::counter(COMMITS, stats.commits.load(Relaxed)), + allocs: metric::counter(ALLOCS, stats.allocs.load(Relaxed)), + reallocs: metric::counter(REALLOCS, stats.reallocs.load(Relaxed)), + compactions: metric::counter(COMPACTIONS, stats.compactions.load(Relaxed)), + removals: metric::counter(REMOVALS, stats.removals.load(Relaxed)), + resizes: metric::counter(RESIZES, stats.resizes.load(Relaxed)), + loads_vec, + loads, + operations: OperationCounters::new(OPERATION_TIME), + accounts: metric::gauge_vec(ACCOUNTS, &[STORE_LABEL]), + }; + metrics.register(); + metrics + } + + /// Registers all collectors in the default Prometheus registry. + fn register(&self) { + metric::register("accountsdb", self.reads.clone()); + metric::register("accountsdb", self.commits.clone()); + metric::register("accountsdb", self.allocs.clone()); + metric::register("accountsdb", self.reallocs.clone()); + metric::register("accountsdb", self.compactions.clone()); + metric::register("accountsdb", self.removals.clone()); + metric::register("accountsdb", self.resizes.clone()); + metric::register("accountsdb", self.loads_vec.clone()); + self.operations.register("accountsdb"); + metric::register("accountsdb", self.accounts.clone()); + } +} diff --git a/accountsdb/src/snapshot.rs b/accountsdb/src/snapshot.rs new file mode 100644 index 0000000..641b046 --- /dev/null +++ b/accountsdb/src/snapshot.rs @@ -0,0 +1,102 @@ +//! Snapshot export helpers. + +use std::{ + fs::{self, File}, + io::{self, BufWriter, Write}, + path::PathBuf, +}; + +use nucleus::MB; +use tracing::info; + +use crate::{ + ACTIVE_DIR, AccountsDB, + metrics::{self, Operation}, +}; + +/// Snapshot directory prefix. +const PREFIX: &str = "snapshot-"; +/// Snapshot payload filename for the volatile store. +pub(crate) const VOLATILE_DB_FILE: &str = "volatile.db"; + +/// Errors while writing a snapshot directory. +#[derive(thiserror::Error, Debug)] +pub enum SnapshotError { + /// I/O while writing the snapshot. + #[error("snapshot export I/O error")] + IO(#[from] io::Error), + /// Failed to flush the persisted store before copying the tree. + #[error("failed to flush persisted store")] + Flush(#[from] heed::Error), + /// Failed to serialize the volatile store into the snapshot. + #[error("failed to serialize volatile store")] + Serde(#[from] Box), + /// Failed to clone the active database tree into the snapshot slot. + #[error("failed to clone snapshot tree")] + FsClone(#[from] Box), + /// No archived snapshot could be restored. + #[error("no valid archived accountsdb snapshot found")] + Missing, +} + +/// Result type used by snapshot export and restore helpers. +pub type SnapshotResult = Result; + +/// Active database backup operation. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum BackupOp { + /// Move the active database tree to its backup path. + Save, + /// Move the saved backup tree back to the active database path. + Restore, +} + +impl AccountsDB { + /// Writes a superblock snapshot under `root`. + /// + /// # Safety + /// The caller must ensure exclusive write access while the snapshot is in + /// progress. The persisted backend is flushed first, then the active tree + /// is cloned, and finally the volatile store is rewritten in the clone. + /// That ordering keeps the exported state coherent only when no concurrent + /// writes can race with the export. + pub unsafe fn snapshot(&self, superblock: u64) -> SnapshotResult { + let _timer = metrics::time(Operation::Snapshot); + let src = self.root.join(ACTIVE_DIR); + let dst = self.root.join(format!("{PREFIX}{superblock:0>9}")); + // SAFETY: snapshot owns exclusive write access, so defrag cannot race + // with concurrent mutation and can compact the persisted store first. + unsafe { self.persisted.defragment() }?; + // Persisted state must reach disk before we copy the active tree. + self.persisted.flush(true)?; + // Clone the whole active tree, then replace the volatile payload below. + clonetree::clone_tree(src, &dst, &Default::default()).map_err(Box::new)?; + + let file = File::options() + .create(true) + .truncate(true) + .write(true) + .open(dst.join(VOLATILE_DB_FILE))?; + let mut buffered = BufWriter::with_capacity(4 * MB, file); + // The clone may still contain a stale volatile.db; overwrite it with + // the current in-memory volatile store. + bincode::serialize_into(&mut buffered, &self.volatile.accounts)?; + buffered.flush()?; + + info!(superblock, ?dst, "wrote accountsdb snapshot"); + Ok(dst) + } + + /// Saves or restores the active database tree. + pub fn backup(&self, op: BackupOp) -> SnapshotResult { + let active = self.root.join(ACTIVE_DIR); + let backup = self.root.join(format!("{ACTIVE_DIR}.bkp")); + match op { + BackupOp::Save => fs::rename(&active, &backup), + BackupOp::Restore => fs::rename(&backup, &active), + } + .map(|()| backup) + .inspect(|backup| info!(?op, ?backup, "accountsdb backup")) + .map_err(Into::into) + } +} diff --git a/accountsdb/src/store/defrag.rs b/accountsdb/src/store/defrag.rs new file mode 100644 index 0000000..e9124bf --- /dev/null +++ b/accountsdb/src/store/defrag.rs @@ -0,0 +1,133 @@ +#![allow(unsafe_op_in_unsafe_fn)] + +use heed::Result; +use solana_account::BorrowedAccount; +use tracing::{debug, info}; + +use crate::{ + metrics::{self, Operation}, + store::kv::{Offset, OwnerAndOffset}, +}; + +use super::PersistedStore; + +/// Free span in the persisted image file, measured in storage units and +/// ordered by offset for the sweep. +#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +struct Hole { + offset: Offset, + units: u32, +} + +impl Hole { + /// Builds one hole from the freelist entry tuple. + fn new((units, offset): (u32, Offset)) -> Self { + Self { offset, units } + } + + /// Returns the first offset past the hole. + fn end(&self) -> Offset { + self.offset + self.units + } +} + +impl PersistedStore { + /// Compacts the tail half of the freelist and repacks the storage. + /// + /// Caller must guarantee exclusive access while defrag runs. Holes are + /// swept in file order, contiguous holes are merged into one run, and the + /// following live span is reindexed before its bytes are slid left. Each + /// pass starts at the tail half of the sorted hole list so repeated runs + /// halve the remaining work. + /// + /// # Safety + /// + /// No concurrent mutation may touch the persisted index or mapped + /// storage while offsets are rewritten and bytes are moved. + pub(crate) unsafe fn defragment(&self) -> Result<()> { + let _timer = metrics::time(Operation::Defragmentation); + let mut holes = { + let txn = self.index.env.read_txn()?; + self.index + .freelist + .iter(&txn)? + .map(|r| r.map(Hole::new)) + .collect::>>()? + }; + holes.sort_unstable(); + + let mut i = holes.len() / 2; + // `dst` is the next write head in storage units. + let Some(mut dst) = holes.get(i).map(|h| h.offset) else { + debug!("nothing to defragment"); + return Ok(()); + }; + // Tail is the live end; if the last hole reaches it, there is no + // trailing live span to move. + let tail = Offset(self.storage.cursor()); + while i < holes.len() { + let run = i; + let mut end = holes[i].end(); + i += 1; + + while let Some(&hole) = holes.get(i) { + if hole.offset != end { + break; + } + end = hole.end(); + i += 1; + } + + let next = holes.get(i).map(|h| h.offset).unwrap_or(tail); + // Defrag runs exclusively, so rewrite and commit the index before + // moving the live bytes that now point at the future offsets. + let shift = end - dst; + self.reindex(end, next, shift, &holes[run..i])?; + // Source and destination overlap by design during left-compaction. + dst = self.slide(end, dst, next - end); + } + let reclaimed = tail - dst; + info!(reclaimed, "defragmented persisted storage"); + self.storage.shrink(dst.0).map_err(Into::into) + } + + /// Rewrites the index for a live span before its bytes are moved. + /// + /// # Safety + /// + /// `src..end` must cover whole serialized accounts in mapped storage. + /// The caller must guarantee exclusive access while offsets are updated. + unsafe fn reindex(&self, src: Offset, end: Offset, shift: u32, run: &[Hole]) -> Result<()> { + let mut txn = self.index.env.write_txn()?; + let mut pos = src; + while pos < end { + let ptr = self.storage.at(pos); + let span = BorrowedAccount::span(ptr); + let next = pos + span; + // The bytes are still at `pos`; compute the future offset before copying. + let pubkey = BorrowedAccount::pubkey(ptr); + let account = BorrowedAccount::init(ptr); + let owner = account.owner().into(); + let data = OwnerAndOffset { owner, offset: pos - shift }; + self.index.relocate(&pubkey, pos, data, &mut txn)?; + self.storage.stats().compact(); + pos = next; + } + for hole in run { + self.index.freelist.delete_one_duplicate(&mut txn, &hole.units, &hole.offset)?; + } + txn.commit() + } + + /// Slides a live region left after the index has been committed. + /// + /// # Safety + /// + /// `src..src+len` and `dst..dst+len` must be valid mapped storage + /// ranges. The ranges may overlap. + unsafe fn slide(&self, src: Offset, dst: Offset, len: u32) -> Offset { + let src = self.storage.at(src); + src.copy_to(self.storage.at(dst), len as usize); + dst + len + } +} diff --git a/accountsdb/src/store/index.rs b/accountsdb/src/store/index.rs new file mode 100644 index 0000000..6409004 --- /dev/null +++ b/accountsdb/src/store/index.rs @@ -0,0 +1,204 @@ +//! LMDB index for persisted accounts. +//! +//! The index maps compact pubkey tags to storage offsets and owner tags, +//! plus a freelist keyed by image size. + +use std::{fs, mem, path::Path}; + +use heed::{ + Database, DatabaseFlags, Env, EnvFlags, EnvOpenOptions, IntegerComparator, Result, RoIter, + RoTxn, RwTxn, iteration_method::MoveOnCurrentKeyDuplicates, +}; +use nucleus::heed::{DatabaseIndex, RoTxnTls}; +use solana_pubkey::Pubkey; + +use crate::store::kv::{KeyTail, Offset, OwnerAndOffset, PubkeyBytes, U32LE}; + +use super::mmap::INDEX_MAP_SIZE; + +/// Subdirectory used for the LMDB index. +const INDEX_SUBDIR: &str = "index"; +/// Accounts table name. +const ACCOUNTS_INDEX: &str = "accounts"; +/// Program ownership table name. +const PROGRAMS_INDEX: &str = "programs"; +/// Freelist table name. +const FREELIST_INDEX: &str = "freelist"; + +/// Iterator over all persisted accounts in pubkey order. +type RoAccountIter<'a> = RoIter<'a, PubkeyBytes, OwnerAndOffset>; +/// Duplicate iterator over program-owned persisted accounts. +type RoProgramIter<'a> = RoIter<'a, KeyTail, Offset, MoveOnCurrentKeyDuplicates>; +/// Iterator over persisted accounts. +pub(crate) struct AccountIter<'a> { + /// Iterator over `pubkey -> account` entries. + pub(super) inner: RoAccountIter<'a>, + /// Keeps the read transaction alive for the iterator lifetime. + pub(super) _txn: RoTxnTls<'a>, +} +/// Duplicate iterator over persisted accounts for one owner. +pub(crate) struct OwnerIter<'a> { + /// Duplicates iterator over `owner -> account` entries. + pub(crate) inner: RoProgramIter<'a>, + /// Keeps the read transaction alive for the iterator lifetime. + pub(crate) _txn: RoTxnTls<'a>, +} + +/// LMDB index over persisted account offsets and owners. +pub(crate) struct Index { + /// LMDB environment for the on-disk index. + pub(super) env: Env, + /// Account pubkey -> offset + owner keytag. + pub(super) accounts: Database, + /// Owner keytag -> offset. + pub(super) programs: Database, + /// Image size -> offset. + pub(super) freelist: Database, +} + +impl Index { + /// Opens or creates the index directory and databases. + pub(crate) fn new(path: &Path) -> crate::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(3) + .map_size(INDEX_MAP_SIZE) + .flags(EnvFlags::WRITE_MAP) + .flags(EnvFlags::NO_READ_AHEAD) + .flags(EnvFlags::NO_SYNC) + .open(path)? + }; + + let mut txn = env.write_txn()?; + let accounts = env.database_options().name(ACCOUNTS_INDEX).types().create(&mut txn)?; + let programs = env + .database_options() + .name(PROGRAMS_INDEX) + .flags(DatabaseFlags::DUP_SORT | DatabaseFlags::DUP_FIXED) + .types() + .create(&mut txn)?; + let freelist = env + .database_options() + .name(FREELIST_INDEX) + .flags(DatabaseFlags::DUP_SORT | DatabaseFlags::DUP_FIXED) + .key_comparator() + .types() + .create(&mut txn)?; + txn.commit()?; + Ok(Self { + env, + accounts, + programs, + freelist, + }) + } + + /// Returns the persisted offset for `pubkey`. + pub(crate) fn offset(&self, key: &Pubkey, txn: &RoTxn<'_>) -> Result> { + let entry = self.accounts.get(txn, key)?; + Ok(entry.map(|e| e.offset)) + } + + /// Takes a freed span from the freelist when one matches `units`. + pub(crate) fn allocate(&self, units: u32, txn: &mut RwTxn<'_>) -> Result> { + let offset = self.freelist.get(txn, &units)?; + if let Some(offset) = offset { + self.freelist.delete_one_duplicate(txn, &units, &offset)?; + Ok(Some(offset)) + } else { + Ok(None) + } + } + + /// Inserts an account and its owner mapping. + pub(crate) fn insert( + &self, + key: &Pubkey, + data: OwnerAndOffset, + txn: &mut RwTxn<'_>, + ) -> Result<()> { + self.accounts.put(txn, key, &data)?; + let OwnerAndOffset { owner, offset } = data; + self.programs.put(txn, &owner, &offset) + } + + /// Removes an account and returns its persisted offset. + pub(crate) fn delete(&self, key: &Pubkey, txn: &mut RwTxn<'_>) -> Result> { + let Some(entry) = self.accounts.get(txn, key)? else { + return Ok(None); + }; + + let OwnerAndOffset { owner, offset } = entry; + self.accounts.delete(txn, key)?; + + self.programs.delete_one_duplicate(txn, &owner, &offset)?; + Ok(Some(offset)) + } + + /// Returns the duplicate iterator for accounts owned by `owner`. + pub(crate) fn program<'a>(&'a self, owner: Pubkey) -> Result>> { + let owner = owner.into(); + let txn = self.env.read_txn()?; + let Some(iter) = self.programs.get_duplicates(&txn, &owner)? else { + return Ok(None); + }; + // The duplicate iterator borrows `txn`; storing it in the wrapper keeps + // the borrow alive for the iterator lifetime. + // SAFETY: the wrapper owns `txn`, so the duplicate iterator cannot outlive it. + let iter = unsafe { mem::transmute::, RoProgramIter<'a>>(iter) }; + Ok(Some(OwnerIter { _txn: txn, inner: iter })) + } + + /// Returns an iterator over all accounts in pubkey order. + pub(crate) fn accounts<'a>(&'a self) -> Result> { + let txn = self.env.read_txn()?; + let iter = self.accounts.iter(&txn)?; + // The iterator borrows `txn`; storing it in the wrapper keeps the + // transaction alive for the iterator lifetime. + // SAFETY: the wrapper owns `txn`, so the iterator cannot outlive it. + let iter = unsafe { mem::transmute::, RoAccountIter<'a>>(iter) }; + Ok(AccountIter { _txn: txn, inner: iter }) + } + + /// Moves an account entry to a new owner while preserving its offset. + pub(crate) fn update_owner( + &self, + acc: &Pubkey, + new: KeyTail, + txn: &mut RwTxn<'_>, + ) -> Result<()> { + let Some(val) = self.accounts.get(txn, acc)? else { + return Ok(()); + }; + let OwnerAndOffset { owner: old, offset } = val; + self.programs.delete_one_duplicate(txn, &old, &offset)?; + + let data = OwnerAndOffset { owner: new, offset }; + self.accounts.put(txn, acc, &data)?; + self.programs.put(txn, &new, &offset) + } + + /// Moves an account entry to a new offset while preserving its owner. + pub(crate) fn relocate( + &self, + key: &Pubkey, + old: Offset, + new: OwnerAndOffset, + txn: &mut RwTxn<'_>, + ) -> Result<()> { + self.accounts.put(txn, key, &new)?; + let OwnerAndOffset { owner, offset } = new; + self.programs.delete_one_duplicate(txn, &owner, &old)?; + self.programs.put(txn, &owner, &offset) + } +} + +unsafe impl DatabaseIndex for Index { + fn env(&self) -> &Env { + &self.env + } +} diff --git a/accountsdb/src/store/kv.rs b/accountsdb/src/store/kv.rs new file mode 100644 index 0000000..75b26a1 --- /dev/null +++ b/accountsdb/src/store/kv.rs @@ -0,0 +1,123 @@ +use std::{array, borrow::Cow, ops}; + +use bytemuck::{Pod, Zeroable}; +use heed::{BoxedError, BytesDecode, BytesEncode, byteorder::LittleEndian, types::U32}; +use solana_pubkey::Pubkey; + +/// Result type used by LMDB byte codecs. +pub(crate) type CodecResult = Result; +/// Little-endian `u32` value stored in the freelist. +pub(super) type U32LE = U32; +/// Offset into mapped storage, measured in storage units. +#[derive(Clone, Copy, Zeroable, Pod, PartialEq, Eq, PartialOrd, Ord)] +#[repr(C)] +pub(crate) struct Offset(pub(super) u32); + +/// Compact 16-byte LMDB tag derived from the tail half of a pubkey. +#[derive(Clone, Copy, Pod, Zeroable)] +#[repr(C)] +pub(crate) struct KeyTail([u8; 16]); + +impl From for KeyTail { + fn from(v: Pubkey) -> Self { + Self(array::from_fn(|i| v.as_array()[i + size_of::()])) + } +} + +/// Full 32-byte pubkey codec for the accounts table. +pub(super) struct PubkeyBytes; + +/// LMDB value for the accounts table. +#[derive(Clone, Copy, Pod, Zeroable)] +#[repr(C)] +pub(crate) struct OwnerAndOffset { + /// Owner key tag for the stored account image. + pub(crate) owner: KeyTail, + /// Offset into mapped storage. + pub(crate) offset: Offset, +} + +impl<'a> BytesEncode<'a> for KeyTail { + type EItem = Self; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + Ok(bytemuck::bytes_of(item).into()) + } +} + +impl<'a> BytesDecode<'a> for KeyTail { + 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 PubkeyBytes { + type EItem = Pubkey; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + Ok(item.as_array().into()) + } +} + +impl<'a> BytesDecode<'a> for PubkeyBytes { + type DItem = &'a Pubkey; + + fn bytes_decode(bytes: &'a [u8]) -> CodecResult { + bytemuck::try_from_bytes(bytes).map_err(Into::into) + } +} + +impl<'a> BytesEncode<'a> for OwnerAndOffset { + type EItem = Self; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + Ok(bytemuck::bytes_of(item).into()) + } +} + +impl<'a> BytesDecode<'a> for OwnerAndOffset { + type DItem = Self; + + fn bytes_decode(bytes: &'a [u8]) -> CodecResult { + bytemuck::try_pod_read_unaligned(bytes).map_err(Into::into) + } +} + +impl<'a> BytesEncode<'a> for Offset { + type EItem = Self; + + fn bytes_encode(item: &'a Self::EItem) -> CodecResult> { + U32LE::bytes_encode(&item.0) + } +} + +impl<'a> BytesDecode<'a> for Offset { + type DItem = Self; + + fn bytes_decode(bytes: &'a [u8]) -> CodecResult { + U32LE::bytes_decode(bytes).map(Self) + } +} + +impl ops::Add for Offset { + type Output = Self; + fn add(self, rhs: u32) -> Self::Output { + Self(self.0 + rhs) + } +} + +impl ops::Sub for Offset { + type Output = Self; + fn sub(self, rhs: u32) -> Self::Output { + Self(self.0 - rhs) + } +} + +impl ops::Sub for Offset { + type Output = u32; + fn sub(self, rhs: Self) -> Self::Output { + self.0 - rhs.0 + } +} diff --git a/accountsdb/src/store/mmap.rs b/accountsdb/src/store/mmap.rs new file mode 100644 index 0000000..b0cefc7 --- /dev/null +++ b/accountsdb/src/store/mmap.rs @@ -0,0 +1,297 @@ +//! Mapped storage for persisted account images. +//! +//! The file reserves a small meta header at the front, followed by the raw +//! account images written in `solana-account`'s borrowed layout. + +use std::{ + fs::File, + io::{self, Write}, + ops::Range, + os::fd::AsRawFd, + path::Path, + ptr::NonNull, + sync::atomic::{AtomicU32, AtomicU64, Ordering::*}, +}; + +use memmap2::{MmapMut, MmapOptions}; +use nucleus::{GB, MB}; +use parking_lot::Mutex; +use solana_account::{STORAGE_UNIT, StorageUnit}; +use tracing::{debug, error}; + +use crate::{ + AccountsDBError, Result, metrics, + store::{DatabaseVersion, VERSION, kv::Offset}, +}; + +/// Bytes reserved at the front of the mapped file for metadata. +const DATABASE_META_RESERVATION: usize = 256; +/// Filename used for the mapped storage file. +const STORAGE_FILE: &str = "storage.db"; +/// LMDB map size for the index database. +pub(crate) const INDEX_MAP_SIZE: usize = GB; +/// Growth block for the mapped storage file. +pub(crate) const STORAGE_BLOCK: u64 = 256 * MB as u64; +/// Initial file size: one storage block plus the metadata reservation. +const INIT_STORAGE_SIZE: u64 = STORAGE_BLOCK + DATABASE_META_RESERVATION as u64; +/// Maximum mapped storage size. +const MMAP_SIZE: usize = u32::MAX as usize * STORAGE_UNIT + DATABASE_META_RESERVATION; + +/// One allocation inside the mapped storage. +pub(crate) struct Allocation { + /// Offset from the start of the storage area, in storage units. + pub(crate) offset: Offset, + /// Pointer to the start of the allocated image. + pub(crate) ptr: NonNull, +} + +/// Mapped storage backing persisted account images. +pub(crate) struct MappedStorage { + /// Pointer to the reserved metadata header. + meta: NonNull, + /// Full file mapping. + mmap: MmapMut, + /// Start of the account image region. + head: NonNull, + /// File handle used for resizing. + file: Mutex, +} + +#[repr(C)] +#[derive(Default)] +/// Runtime counters for the persisted backend. +pub(crate) struct Stats { + /// Persisted image loads. + pub(crate) reads: AtomicU64, + /// `BorrowedAccount::commit` calls. + pub(crate) commits: AtomicU64, + /// Fresh allocations on backing storage. + pub(crate) allocs: AtomicU64, + /// Freelist allocation reuse. + pub(crate) reallocs: AtomicU64, + /// Account relocations during defrag. + pub(crate) compactions: AtomicU64, + /// Persisted deletes. + pub(crate) removals: AtomicU64, + /// File resizes. + pub(crate) resizes: AtomicU64, +} + +impl Stats { + /// Counts one persisted read. + pub(crate) fn read(&self) { + self.reads.fetch_add(1, Relaxed); + metrics::read(); + } + + /// Counts one borrowed account commit. + pub(crate) fn commit(&self) { + self.commits.fetch_add(1, Relaxed); + metrics::commit(); + } + + /// Counts one fresh allocation from the mapped file. + pub(crate) fn alloc(&self) { + self.allocs.fetch_add(1, Relaxed); + metrics::alloc(); + } + + /// Counts one freelist reuse. + pub(crate) fn realloc(&self) { + self.reallocs.fetch_add(1, Relaxed); + metrics::realloc(); + } + + /// Counts one relocation during defragmentation. + pub(crate) fn compact(&self) { + self.compactions.fetch_add(1, Relaxed); + metrics::compaction(); + } + + /// Counts one persisted removal. + pub(crate) fn remove(&self) { + self.removals.fetch_add(1, Relaxed); + metrics::removal(); + } + + /// Counts one file resize. + pub(crate) fn resize(&self) { + self.resizes.fetch_add(1, Relaxed); + metrics::resize(); + } +} + +#[repr(C)] +#[derive(Default)] +/// Metadata header stored at the front of the mapped file. +pub(crate) struct DatabaseMeta { + /// On-disk format version. + version: DatabaseVersion, + /// Last computed database checksum. + pub(crate) checksum: AtomicU64, + /// Current slot. + pub(crate) slot: AtomicU64, + /// Current backing file length in bytes. + len: AtomicU64, + /// Database statistics. + stats: Stats, + /// Next allocation cursor. + pub(super) cursor: AtomicU32, +} + +impl MappedStorage { + /// Opens or creates the mapped storage file. + pub(crate) fn new(path: &Path) -> Result { + let path = path.join(STORAGE_FILE); + let mut file = + File::options().create(true).truncate(false).read(true).write(true).open(path)?; + let fd = file.as_raw_fd(); + // SAFETY: the file is opened read/write and mapped for the full fixed size. + let mut mmap = unsafe { MmapOptions::new().len(MMAP_SIZE).map_mut(fd)? }; + if file.metadata()?.len() == 0 { + file.set_len(INIT_STORAGE_SIZE)?; + file.flush()?; + let meta = DatabaseMeta { + version: VERSION, + len: INIT_STORAGE_SIZE.into(), + slot: 1.into(), + ..Default::default() + }; + // SAFETY: the first bytes of the mapping are reserved for `DatabaseMeta`. + unsafe { mmap.as_mut_ptr().cast::().write(meta) }; + mmap.flush()?; + } + // SAFETY: the mapping is at least `DATABASE_META_RESERVATION` bytes long, + // so the meta header and account head pointers stay within the map. + let (meta, head) = unsafe { + let head = mmap.as_mut_ptr().add(DATABASE_META_RESERVATION); + let head = NonNull::new_unchecked(head.cast()); + let meta = NonNull::new_unchecked(mmap.as_mut_ptr().cast()); + (meta, head) + }; + let file = Mutex::new(file); + Ok(Self { meta, mmap, head, file }) + } + + /// Flushes dirty pages to durable storage. + pub(crate) fn flush(&self, sync: bool) -> io::Result<()> { + let range = self.active(); + if sync { + self.mmap.flush_range(range.start, range.len()) + } else { + self.mmap.flush_async_range(range.start, range.len()) + } + } + + /// Validates the opened storage format. + pub(crate) fn validate(&self) -> Result<()> { + let meta = self.meta(); + if meta.version != VERSION { + Err(AccountsDBError::UnsupportedVersion(meta.version)) + } else { + Ok(()) + } + } + + /// Returns a pointer inside the account image region. + pub(crate) fn at(&self, offset: Offset) -> NonNull { + // SAFETY: private call sites pass offsets from the index, allocator, or + // defrag cursor and uphold the mapped-region bounds. + unsafe { self.head.add(offset.0 as usize) } + } + + /// Returns the runtime counters. + pub(crate) fn stats(&self) -> &Stats { + &self.meta().stats + } + + /// Allocates a fresh span of `units` storage units. + pub(crate) fn allocate(&self, units: u32) -> Result { + let meta = self.meta(); + let mut offset = meta.cursor.load(Acquire); + loop { + let end = offset.checked_add(units).ok_or(AccountsDBError::Allocation)?; + let needed = Self::bytes(end as u64); + if needed > meta.len.load(Acquire) { + self.grow(needed)?; + } + if let Err(updated) = meta.cursor.compare_exchange(offset, end, AcqRel, Acquire) { + offset = updated; + } else { + break; + } + } + self.stats().alloc(); + let offset = Offset(offset); + let ptr = self.at(offset); + Ok(Allocation { offset, ptr }) + } + + /// Returns a mutable reference to the metadata header. + pub(crate) fn meta(&self) -> &DatabaseMeta { + // SAFETY: `meta` points to the reserved header at the front of the map. + unsafe { &*self.meta.as_ptr() } + } + + /// Returns the current allocation cursor in storage units. + pub(crate) fn cursor(&self) -> u32 { + self.meta().cursor.load(Acquire) + } + + /// Shrinks the file to the current cursor. + pub(super) fn shrink(&self, units: u32) -> io::Result<()> { + self.resize(Self::bytes(units as u64), u64::le)?; + self.meta().cursor.store(units, Release); + Ok(()) + } + + /// Returns the active byte range, including the metadata reservation. + fn active(&self) -> Range { + 0..Self::bytes(self.cursor() as u64) as usize + } + + /// Converts storage units into file bytes, including the metadata reservation. + fn bytes(units: u64) -> u64 { + units * STORAGE_UNIT as u64 + DATABASE_META_RESERVATION as u64 + } + + /// Grows the file to at least `len` bytes. + /// Rounds up to a storage block before resizing. + fn grow(&self, mut len: u64) -> Result<()> { + len = len.div_ceil(STORAGE_BLOCK) * STORAGE_BLOCK; + if len > MMAP_SIZE as u64 { + error!( + requested = len, + limit = MMAP_SIZE, + "mapped storage limit exceeded" + ); + return Err(AccountsDBError::Allocation); + } + self.resize(len, u64::ge).map_err(Into::into) + } + + /// Resizes the file when the current size does not satisfy `cmp`. + /// + /// The file is updated before the new size is published into metadata so + /// readers never observe a larger size than the actual mapping. + fn resize(&self, len: u64, cmp: fn(&u64, &u64) -> bool) -> io::Result<()> { + let mut file = self.file.lock(); + if cmp(&file.metadata()?.len(), &len) { + return Ok(()); + } + // Resize the file first, then publish the new size into metadata. + file.set_len(len)?; + file.flush()?; + self.meta().len.store(len, Release); + self.stats().resize(); + self.mmap.flush()?; + debug!(len, "resized storage file"); + Ok(()) + } +} + +// SAFETY: the `NonNull` pointers point into the owned `mmap` and are never +// reseated; concurrent access is synchronized through atomics in the metadata +// header and the `Mutex`, so the storage is safe to send and share. +unsafe impl Send for MappedStorage {} +unsafe impl Sync for MappedStorage {} diff --git a/accountsdb/src/store/mod.rs b/accountsdb/src/store/mod.rs new file mode 100644 index 0000000..dd5267b --- /dev/null +++ b/accountsdb/src/store/mod.rs @@ -0,0 +1,292 @@ +//! Persisted account load and write path. +//! +//! This module coordinates the mmap, LMDB index, and borrowed account layout. + +use core::{hash::Hasher, slice}; +use std::sync::atomic::Ordering::{Acquire, Release}; + +use solana_account::{ + AccountMode, AccountSharedData, BorrowedAccount, CoWAccount::*, DirtyMarkers, OwnedAccount, +}; +use solana_pubkey::Pubkey; +use tracing::{error, warn}; + +use nucleus::heed::{DatabaseIndex, OptRoTxn, OptRwTxn}; +use twox_hash::XxHash3_64; + +use crate::{ + AccountEntry, AccountsDBError, Result, StoreKind, + metrics::{self, Operation}, + store::{ + index::{Index, OwnerIter}, + kv::{Offset, OwnerAndOffset}, + mmap::{DatabaseMeta, MappedStorage}, + }, +}; + +mod defrag; +pub(crate) mod index; +mod kv; +mod mmap; + +pub(crate) use mmap::Stats; + +/// Current on-disk storage format version. +pub(crate) const VERSION: DatabaseVersion = 1; +/// Version tag stored in the metadata header. +pub(crate) type DatabaseVersion = u64; + +/// Persisted store backed by the mmap and LMDB index. +pub(crate) struct PersistedStore { + /// Mapped account storage. + pub(crate) storage: MappedStorage, + /// LMDB index over persisted accounts. + pub(crate) index: Index, +} + +/// Iterator over persisted program-owned accounts. +pub(crate) struct PersistedProgramIter<'a> { + /// Keeps the read transaction alive while iterating. + iter: OwnerIter<'a>, + /// Mapped storage backing the returned borrowed accounts. + mmap: &'a MappedStorage, +} + +impl PersistedStore { + /// Opens or creates the persisted store at `path`. + pub(crate) fn new(path: &std::path::Path) -> Result { + let index = Index::new(path)?; + let storage = MappedStorage::new(path)?; + Ok(Self { storage, index }) + } + + /// Loads the persisted image for `pubkey` from the mapped file. + pub(crate) fn load<'e>( + &'e self, + txn: OptRoTxn<'_, 'e>, + pubkey: &Pubkey, + ) -> Result> { + let txn = self.index.read_txn(txn)?; + let offset = self.index.offset(pubkey, txn)?; + offset.is_some().then(|| self.storage.stats().read()); + // SAFETY: offsets come from the persisted index and point into the map. + Ok(offset.map(|o| unsafe { BorrowedAccount::init(self.storage.at(o)) })) + } + + /// Returns whether a persisted account image exists for `pubkey`. + pub(crate) fn contains<'e>(&'e self, txn: OptRoTxn<'_, 'e>, pubkey: &Pubkey) -> Result { + let txn = self.index.read_txn(txn)?; + self.index.offset(pubkey, txn).map(|o| o.is_some()).map_err(Into::into) + } + + /// Applies a batch of account updates to the persisted store. + /// + /// Borrowed mutable accounts are committed in place. Owned mutable accounts + /// are serialized into the mmap. Immutable accounts delete stale persisted + /// entries. If the LMDB commit fails or database runs out of space, the borrowed + /// images are rolled back so in-memory state stays aligned with the durable index. + pub(crate) fn upsert<'a, AC>(&self, accounts: AC) -> Result<()> + where + AC: IntoIterator + Clone, + { + let mut applied = 0; + let mut result = Ok(()); + let mut txn = None; + for entry in accounts.clone() { + result = self.apply(entry, &mut txn); + if result.is_err() { + break; + } + applied += 1; + } + // Commit once after the batch so the index and mmap stay in sync. + if let Some(txn) = txn + && result.is_ok() + { + metrics::accounts(StoreKind::Persisted, self.index.accounts.len(&txn)?); + result = txn.commit().map_err(Into::into); + } + if let Err(error) = &result { + warn!(applied, ?error, "accounts persistence failed; rolling back"); + // Only borrowed accounts need rollback here: owned inserts never + // mutate an existing borrowed image in place. + let processed = accounts.into_iter().take(applied).map(|(_, a)| a); + Self::rollback(processed); + } + + result + } + + /// Returns the persisted program iterator for `owner`. + pub(crate) fn program(&self, owner: Pubkey) -> Result>> { + let i = self.index.program(owner)?; + Ok(i.map(|iter| PersistedProgramIter { iter, mmap: &self.storage })) + } + + /// Flushes the mapped storage and LMDB index to durable storage. + pub(crate) fn flush(&self, sync: bool) -> heed::Result<()> { + let _timer = metrics::time(Operation::Flush); + self.index.flush()?; + if sync { + let checksum = self.checksum()?; + self.meta().checksum.store(checksum, Release); + } + self.storage.flush(sync)?; + Ok(()) + } + + /// Validates the persisted store checksum and on-disk format version. + pub(crate) fn validate(&self) -> Result<()> { + self.storage.validate()?; + if self.storage.cursor() == 0 { + return Ok(()); + } + let expected = self.meta().checksum.load(Acquire); + let actual = self.checksum()?; + if expected != actual { + error!( + expected, + actual, "persisted store checksum mismatch; database corrupted" + ); + return Err(AccountsDBError::Corruption); + } + + Ok(()) + } + + /// Applies one account state transition to the persisted backend. + fn apply<'e>(&'e self, acc: &AccountEntry, txn: OptRwTxn<'_, 'e>) -> Result<()> { + let (pubkey, account) = acc; + // The persisted store is authoritative only for mutable accounts. An + // account that has flipped to immutable, or has been closed, no longer + // belongs here, so drop any stale persisted entry. + if !account.mutable() || account.is(AccountMode::Closed) { + return self.delete(pubkey, txn); + } + + let markers = account.markers(); + match account.cow() { + Borrowed(acc) => self.update(pubkey, acc, markers, txn), + Owned(acc) => self.insert(pubkey, acc, txn), + } + } + + /// Rolls back borrowed accounts that were already touched in the batch. + fn rollback<'a, AC>(accounts: AC) + where + AC: Iterator, + { + for acc in accounts { + let Borrowed(acc) = acc.cow() else { continue }; + // SAFETY: only borrowed accounts were updated before the failed commit. + unsafe { acc.rollback() }; + } + } + + /// Commits a borrowed image after updating its owner mapping if needed. + fn update<'e>( + &'e self, + pubkey: &Pubkey, + acc: &BorrowedAccount, + markers: &DirtyMarkers, + txn: OptRwTxn<'_, 'e>, + ) -> Result<()> { + if markers.contains(DirtyMarkers::OWNER) { + let txn = self.index.write_txn(txn)?; + let owner = acc.owner().into(); + self.index.update_owner(pubkey, owner, txn)?; + } + acc.commit(); + self.storage.stats().commit(); + Ok(()) + } + + /// Serializes an owned image into mapped storage and records its offset. + fn insert<'e>( + &'e self, + pubkey: &Pubkey, + acc: &OwnedAccount, + txn: OptRwTxn<'_, 'e>, + ) -> Result<()> { + let txn = self.index.write_txn(txn)?; + let units = acc.units(); + let owner = acc.owner().into(); + + let (ptr, offset) = if let Some(offset) = self.index.allocate(units, txn)? { + let ptr = self.storage.at(offset); + self.storage.stats().realloc(); + (ptr, offset) + } else { + let alloc = self.storage.allocate(units)?; + (alloc.ptr, alloc.offset) + }; + let data = OwnerAndOffset { owner, offset }; + if let Some(offset) = self.index.delete(pubkey, txn)? { + self.free(offset, txn)?; + } + self.index.insert(pubkey, data, txn)?; + // SAFETY: `ptr` points at a fresh span inside the mapped storage and + // `units` is the exact serialized size of this owned account. + unsafe { + let buffer = slice::from_raw_parts_mut(ptr.as_ptr(), units as usize); + acc.serialize(buffer, pubkey); + }; + Ok(()) + } + + /// Returns one persisted span to the freelist. + fn free(&self, offset: Offset, txn: &mut heed::RwTxn<'_>) -> Result<()> { + // SAFETY: `offset` was returned by the index and still points at a valid image. + let space = unsafe { BorrowedAccount::span(self.storage.at(offset)) }; + self.index.freelist.put(txn, &space, &offset)?; + Ok(()) + } + + /// Removes a persisted image and returns its storage span to the freelist. + fn delete<'e>(&'e self, pubkey: &Pubkey, txn: OptRwTxn<'_, 'e>) -> Result<()> { + let txn = self.index.write_txn(txn)?; + let Some(offset) = self.index.delete(pubkey, txn)? else { + return Ok(()); + }; + self.free(offset, txn)?; + self.storage.stats().remove(); + Ok(()) + } + + /// Returns the persisted storage metadata header. + pub(crate) fn meta(&self) -> &DatabaseMeta { + self.storage.meta() + } + + /// Computes a deterministic checksum over persisted accounts in pubkey order. + fn checksum(&self) -> heed::Result { + let _timer = metrics::time(Operation::Checksum); + let mut hasher = XxHash3_64::new(); + let mut iter = self.index.accounts()?; + for entry in &mut iter.inner { + let (pubkey, data) = entry?; + hasher.write(pubkey.as_array()); + // SAFETY: offsets come from the persisted accounts index and point + // into the mapped storage for this store. + let account = unsafe { BorrowedAccount::init(self.storage.at(data.offset)) }; + hasher.write(account.storage()); + } + Ok(hasher.finish()) + } +} + +impl<'a> Iterator for PersistedProgramIter<'a> { + type Item = AccountEntry; + + fn next(&mut self) -> Option { + let (_, offset) = self.iter.inner.next()?.ok()?; + let ptr = self.mmap.at(offset); + // The image prefix stores the full pubkey, so iteration can recover it + // without consulting LMDB again. + // SAFETY: the iterator yields offsets stored in the same mapped database. + self.mmap.stats().read(); + let pubkey = unsafe { BorrowedAccount::pubkey(ptr) }; + let account = unsafe { BorrowedAccount::init(ptr).into() }; + Some((pubkey, account)) + } +} diff --git a/accountsdb/src/tests.rs b/accountsdb/src/tests.rs new file mode 100644 index 0000000..5643520 --- /dev/null +++ b/accountsdb/src/tests.rs @@ -0,0 +1,489 @@ +//! Integration-style unit tests for the two-backend account store. +//! +//! Each test drives a realistic multi-step flow through the public `AccountsDB` +//! surface and reaches into `pub(crate)` internals only to assert *which* +//! backend a given account landed in — the crate's central persisted/volatile +//! invariant that no public method exposes directly. + +use std::sync::atomic::Ordering::{Relaxed, Release}; + +use assert_matches::assert_matches; +use nucleus::testkit::{TempDir, init_tracing, tempdir}; +use solana_account::{ + AccountBuilder, AccountMode, AccountSharedData, ReadableAccount, WritableAccount, +}; +use solana_pubkey::Pubkey; + +use super::*; +use crate::snapshot::VOLATILE_DB_FILE; + +/// Fresh database on a throwaway directory; the `TempDir` must outlive the db. +fn db() -> (TempDir, AccountsDB) { + init_tracing(); + let dir = tempdir(); + let db = AccountsDB::new(dir.path()).unwrap(); + (dir, db) +} + +/// Owned mutable (persisted) account carrying `data`; its size follows the data. +fn mutable_data(lamports: u64, data: Vec, owner: &Pubkey) -> AccountSharedData { + let mut a = AccountSharedData::new(lamports, data.len(), owner); + a.set_data_from_slice(&data); + a.set_mode(AccountMode::Delegated); + a +} + +/// Whether a persisted image exists for `pubkey`. +fn in_persisted(db: &AccountsDB, pubkey: &Pubkey) -> bool { + let mut txn = None; + db.persisted.contains(&mut txn, pubkey).unwrap() +} + +/// Whether a volatile entry exists for `pubkey`. +fn in_volatile(db: &AccountsDB, pubkey: &Pubkey) -> bool { + db.volatile.contains(pubkey) +} + +/// Pubkeys `owner` owns, in iteration order (persisted first, then volatile). +fn program(db: &AccountsDB, owner: &Pubkey) -> Vec { + db.program(owner).unwrap().map(|(k, _)| k).collect() +} + +/// Balance of the account currently loaded for `pubkey`. +fn lamports(db: &AccountsDB, pubkey: &Pubkey) -> u64 { + db.loader().load(pubkey).unwrap().unwrap().lamports() +} + +/// Loads the account currently stored for `pubkey`. +/// +/// A persisted account comes back as a *borrowed* image and a volatile one as +/// *owned*; storing the loaded value back is how the engine drives mode changes +/// through the routing layer (a freshly built owned account with a non-mutable +/// mode is filtered out of the persisted backend entirely). +fn reload(db: &AccountsDB, pubkey: &Pubkey) -> AccountSharedData { + db.loader().load(pubkey).unwrap().unwrap() +} + +/// Closes `pubkey`, deleting it from whichever backend currently holds it. +/// +/// Goes through the load→mutate→store path so the account is a *borrowed* image +/// the routing layer will actually evict (see [`reload`]). +fn close(db: &AccountsDB, pubkey: &Pubkey) { + let mut acc = reload(db, pubkey); + acc.set_mode(AccountMode::Closed); + db.store(&[(*pubkey, acc)]).unwrap(); +} + +/// Allocation high-water mark of the persisted store, in storage units. +fn cursor(db: &AccountsDB) -> u32 { + db.persisted.storage.cursor() +} + +/// Defragments the persisted store until the cursor stops moving. +/// +/// A single pass sweeps only the tail half of the hole list, so full compaction +/// needs repeated passes. +fn defrag_to_stable(db: &AccountsDB) { + loop { + let before = cursor(db); + // SAFETY: the test is the sole owner of the store during defrag. + unsafe { db.persisted.defragment() }.unwrap(); + if cursor(db) == before { + break; + } + } +} + +// Routing, both eviction directions, owner remap and Closed/reset handling in +// one flow — the persisted-vs-volatile invariant is what this whole crate +// exists to enforce. +#[test] +fn test_routing_and_mutability_flips() { + let (_dir, db) = db(); + let (p, q) = (Pubkey::new_unique(), Pubkey::new_unique()); + let (a, b) = (Pubkey::new_unique(), Pubkey::new_unique()); + + let aacc = AccountBuilder::default().lamports(10).owner(p).mode(AccountMode::Delegated); + let bacc = AccountBuilder::default().lamports(20).owner(p); + // `a` mutable → persisted, `b` immutable → volatile; both owned by `p`. + db.store(&[(a, aacc.build()), (b, bacc.build())]).unwrap(); + assert!(in_persisted(&db, &a) && !in_volatile(&db, &a)); + assert!(in_volatile(&db, &b) && !in_persisted(&db, &b)); + + // Loader reads across both backends; contains agrees. + let loader = db.loader(); + assert_eq!(loader.load(&a).unwrap().unwrap().lamports(), 10); + assert_eq!(loader.load(&b).unwrap().unwrap().lamports(), 20); + assert!(loader.contains(&a).unwrap() && loader.contains(&b).unwrap()); + assert!(!loader.contains(&Pubkey::new_unique()).unwrap()); + drop(loader); + + // Persisted account is yielded before the volatile one. + assert_eq!(program(&db, &p), vec![a, b]); + + // Loading a persisted account returns a borrowed image; mutating its owner + // and re-storing must commit in place and remap the program index. + let mut borrowed = reload(&db, &a); + borrowed.set_owner(q); + db.store(&[(a, borrowed)]).unwrap(); + assert_eq!(program(&db, &p), vec![b]); // `a` left p's set + assert_eq!(program(&db, &q), vec![a]); // and joined q's + + // Mutable → immutable evicts the persisted copy into volatile. + let mut flip = reload(&db, &a); + flip.set_mode(AccountMode::ReadOnly); + flip.set_lamports(30); + db.store(&[(a, flip)]).unwrap(); + assert!(!in_persisted(&db, &a) && in_volatile(&db, &a)); + assert_eq!(lamports(&db, &a), 30); + + // Immutable → mutable evicts it back into persisted. + let mut flip = reload(&db, &a); + flip.set_mode(AccountMode::Delegated); + db.store(&[(a, flip)]).unwrap(); + assert!(in_persisted(&db, &a) && !in_volatile(&db, &a)); + + // Closing removes it from both backends. + close(&db, &a); + assert!(!in_persisted(&db, &a) && !in_volatile(&db, &a)); + + // reset() drops volatile mirror only; persisted state is authoritative. + let c = Pubkey::new_unique(); + let cacc = AccountBuilder::default().lamports(50).owner(p).mode(AccountMode::Delegated); + db.store(&[(c, cacc.build())]).unwrap(); + db.reset(); + assert!(!in_volatile(&db, &b)); + assert!(in_persisted(&db, &c)); +} + +// Persisted state and metadata survive a close/reopen, and validate() accepts +// the synced checksum. +#[test] +fn test_persistence_reopen_and_validate() { + let dir = tempdir(); + let keys: Vec = (0..8).map(|_| Pubkey::new_unique()).collect(); + + let checksum = { + let db = AccountsDB::new(dir.path()).unwrap(); + for (i, k) in keys.iter().enumerate() { + let acc = + AccountBuilder::default().lamports(100 + i as u64).mode(AccountMode::Delegated); + db.store(&[(*k, acc.build())]).unwrap(); + } + db.set_slot(42).unwrap(); + // Sync the checksum into the header so a reopen can validate against it. + db.persisted.flush(true).unwrap(); + assert!(db.validate().is_ok()); + db.checksum() + }; + + let db = AccountsDB::new(dir.path()).unwrap(); + for (i, k) in keys.iter().enumerate() { + assert_eq!(lamports(&db, k), 100 + i as u64); + } + assert_eq!(db.slot(), 42); + assert_eq!(db.checksum(), checksum); + assert!(db.validate().is_ok()); +} + +// A freed span is reused for a same-sized insert instead of growing the file; +// genuinely new accounts still extend it. +#[test] +fn test_freelist_reuse_and_growth() { + let (_dir, db) = db(); + let stats = || { + let s = db.persisted.storage.stats(); + (s.allocs.load(Relaxed), s.reallocs.load(Relaxed)) + }; + + let k1 = Pubkey::new_unique(); + db.store(&[( + k1, + AccountBuilder::default().lamports(1).mode(AccountMode::Delegated).build(), + )]) + .unwrap(); + let base = cursor(&db); + let (_, reallocs) = stats(); + + // Close k1 (returns its span to the freelist), then insert a same-sized + // account: it should land in the freed span without advancing the cursor. + close(&db, &k1); + let k2 = Pubkey::new_unique(); + db.store(&[( + k2, + AccountBuilder::default().lamports(2).mode(AccountMode::Delegated).build(), + )]) + .unwrap(); + assert_eq!(cursor(&db), base); + assert_eq!(stats().1, reallocs + 1); + + // Fresh accounts have no reusable span, so the file grows. + let (allocs, _) = stats(); + for _ in 0..8 { + db.store(&[( + Pubkey::new_unique(), + AccountBuilder::default().lamports(1).mode(AccountMode::Delegated).build(), + )]) + .unwrap(); + } + assert!(cursor(&db) > base); + assert!(stats().0 > allocs); +} + +// Defragmentation reclaims interior holes while preserving every live account's +// content, ownership index, and checksum. +#[test] +fn test_defragment_preserves_live_accounts() { + let (_dir, db) = db(); + let owner = Pubkey::new_unique(); + let keys: Vec = (0..16).map(|_| Pubkey::new_unique()).collect(); + for (i, k) in keys.iter().enumerate() { + let acc = AccountBuilder::default() + .lamports(100 + i as u64) + .owner(owner) + .mode(AccountMode::Delegated); + db.store(&[(*k, acc.build())]).unwrap(); + } + + // Punch alternating holes; keep the survivors for later comparison. + let mut live = Vec::new(); + for (i, k) in keys.iter().enumerate() { + if i % 2 == 0 { + close(&db, k); + } else { + live.push((*k, 100 + i as u64)); + } + } + db.persisted.flush(true).unwrap(); + let checksum = db.checksum(); + let before = cursor(&db); + + defrag_to_stable(&db); + assert!(cursor(&db) < before); + + // Every survivor still loads unchanged and remains program-indexed. + for (k, lam) in &live { + let acc = db.loader().load(k).unwrap().unwrap(); + assert_eq!(acc.lamports(), *lam); + assert_eq!(acc.owner(), &owner); + } + let mut owned = program(&db, &owner); + owned.sort(); + let mut expected: Vec = live.iter().map(|(k, _)| *k).collect(); + expected.sort(); + assert_eq!(owned, expected); + + // Relocating images must not change the content checksum. + db.persisted.flush(true).unwrap(); + assert_eq!(db.checksum(), checksum); +} + +// A snapshot is a self-contained tree: reopening it restores persisted accounts +// and bootstraps the volatile store from volatile.db, which is then consumed. +#[test] +fn test_snapshot_export_and_volatile_restore() { + let src = tempdir(); + let (a, b) = (Pubkey::new_unique(), Pubkey::new_unique()); + + let snapshot = { + let db = AccountsDB::new(src.path()).unwrap(); + let aacc = AccountBuilder::default().lamports(10).mode(AccountMode::Delegated); + let bacc = AccountBuilder::default().lamports(20).mode(AccountMode::ReadOnly); + db.store(&[(a, aacc.build()), (b, bacc.build())]).unwrap(); + // SAFETY: the test holds exclusive access to the store. + unsafe { db.snapshot(1) }.unwrap() + }; + + // Adopt the snapshot as a new database's active tree. + let dst = tempdir(); + let active = AccountsDB::path(dst.path()); + std::fs::rename(&snapshot, &active).unwrap(); + let db = AccountsDB::new(dst.path()).unwrap(); + + assert_eq!(lamports(&db, &a), 10); + assert_eq!(lamports(&db, &b), 20); + assert!(in_persisted(&db, &a)); + assert!(in_volatile(&db, &b)); + // The volatile payload is single-sourced back into memory on open. + assert!(!active.join(VOLATILE_DB_FILE).exists()); + + // Backup renames the active tree out and back. + let saved = db.backup(BackupOp::Save).unwrap(); + assert!(saved.exists() && !active.exists()); + db.backup(BackupOp::Restore).unwrap(); + assert!(active.exists()); +} + +// validate() flags a persisted checksum that no longer matches the images. +#[test] +fn test_corruption_detection() { + let (_dir, db) = db(); + for _ in 0..4 { + let acc = AccountBuilder::default().lamports(1).mode(AccountMode::Delegated); + db.store(&[(Pubkey::new_unique(), acc.build())]).unwrap(); + } + db.persisted.flush(true).unwrap(); + assert!(db.validate().is_ok()); + + // Corrupt the recorded checksum; recomputation must no longer agree. + db.persisted.meta().checksum.store(0xDEAD_BEEF, Release); + assert_matches!(db.validate(), Err(AccountsDBError::Corruption)); +} + +// Several freed spans of one size accumulate as duplicates under a single +// freelist key and are all reissued before the file grows — the N>1 duplicate +// case a broken DUP config silently loses. +#[test] +fn test_freelist_multi_duplicate_reuse() { + let (_dir, db) = db(); + let reallocs = || db.persisted.storage.stats().reallocs.load(Relaxed); + + const N: usize = 6; + let keys: Vec = (0..N).map(|_| Pubkey::new_unique()).collect(); + for k in &keys { + let acc = AccountBuilder::default().lamports(1).mode(AccountMode::Delegated); + db.store(&[(*k, acc.build())]).unwrap(); + } + let base = cursor(&db); + // Close them all: N same-size spans return to the freelist as N duplicates. + for k in &keys { + close(&db, k); + } + let reused = reallocs(); + + // Each of N fresh same-size inserts must land in a freed span, so the cursor + // never advances and every insert is a reuse. + for _ in 0..N { + let acc = AccountBuilder::default().lamports(2).mode(AccountMode::Delegated); + db.store(&[(Pubkey::new_unique(), acc.build())]).unwrap(); + } + assert_eq!(cursor(&db), base); + assert_eq!(reallocs(), reused + N as u64); +} + +// An immutable account changing owner is re-homed in the volatile program index +// and the now-empty old owner set is pruned. +#[test] +fn test_volatile_owner_remap() { + let (_dir, db) = db(); + let (x, y) = (Pubkey::new_unique(), Pubkey::new_unique()); + let k = Pubkey::new_unique(); + + let acc = AccountBuilder::default().lamports(10).owner(x); + db.store(&[(k, acc.build())]).unwrap(); + assert_eq!(program(&db, &x), vec![k]); + + // Re-store the volatile account under a new owner. + let mut moved = reload(&db, &k); + moved.set_owner(y); + db.store(&[(k, moved)]).unwrap(); + + assert_eq!(program(&db, &x), Vec::::new()); // old set pruned + assert_eq!(program(&db, &y), vec![k]); + assert!(in_volatile(&db, &k)); +} + +// The freelist reuses a span only on an exact size match, and accounts of mixed +// sizes survive defragmentation with their data intact. +#[test] +fn test_variable_sizes_and_exact_freelist() { + let (_dir, db) = db(); + let owner = Pubkey::new_unique(); + let reallocs = || db.persisted.storage.stats().reallocs.load(Relaxed); + + // A freed large span cannot satisfy a smaller allocation: sizes differ, so + // the small insert allocates fresh rather than reusing the hole. + let big = Pubkey::new_unique(); + db.store(&[(big, mutable_data(1, vec![0; 4096], &owner))]).unwrap(); + close(&db, &big); + let before = reallocs(); + let small = Pubkey::new_unique(); + db.store(&[(small, mutable_data(2, vec![0; 64], &owner))]).unwrap(); + assert_eq!(reallocs(), before); // size mismatch -> no reuse + + // Store a spread of sizes with distinct data, punch an interior hole, then + // defragment and confirm every survivor keeps its exact bytes. + let sizes = [8usize, 512, 100, 4096, 1]; + let mut live = Vec::new(); + for (i, &space) in sizes.iter().enumerate() { + let k = Pubkey::new_unique(); + let data: Vec = (0..space).map(|b| (b as u8).wrapping_add(i as u8)).collect(); + db.store(&[(k, mutable_data(i as u64, data.clone(), &owner))]).unwrap(); + live.push((k, data)); + } + close(&db, &small); + + defrag_to_stable(&db); + for (k, data) in &live { + assert_eq!( + db.loader().load(k).unwrap().unwrap().data(), + data.as_slice() + ); + } +} + +// The checksum hashes accounts in pubkey order, so it depends only on content — +// not on insertion order or the resulting on-disk offsets. +#[test] +fn test_checksum_order_independent() { + let keys: Vec = (0..8).map(|_| Pubkey::new_unique()).collect(); + + let checksum = |order: &[usize]| { + let (_dir, db) = db(); + for &i in order { + let acc = + AccountBuilder::default().lamports(100 + i as u64).mode(AccountMode::Delegated); + db.store(&[(keys[i], acc.build())]).unwrap(); + } + db.persisted.flush(true).unwrap(); + db.checksum() + }; + + let forward: Vec = (0..keys.len()).collect(); + let reversed: Vec = (0..keys.len()).rev().collect(); + assert_eq!(checksum(&forward), checksum(&reversed)); +} + +// 8 MiB accounts overflow the 256 MiB initial storage block, forcing the file to +// grow; removing half then defragmenting reclaims the large holes and shrinks +// the cursor back — growth and compaction over multi-megabyte images. +#[test] +fn test_large_accounts_growth_and_defrag() { + const SIZE: usize = 8 << 20; // 8 MiB of data per account + const COUNT: usize = 40; // ~320 MiB total, past the 256 MiB initial block + + let (_dir, db) = db(); + let owner = Pubkey::new_unique(); + let resizes = || db.persisted.storage.stats().resizes.load(Relaxed); + let baseline = resizes(); + + // Distinct fill byte per account so content is verifiable without retaining + // the expected bytes. + let keys: Vec = (0..COUNT).map(|_| Pubkey::new_unique()).collect(); + for (i, k) in keys.iter().enumerate() { + db.store(&[(*k, mutable_data(i as u64, vec![i as u8; SIZE], &owner))]).unwrap(); + } + // Crossing the initial block must have grown the file. + assert!(resizes() > baseline); + + // Close every other account to punch large interior holes. + let mut live = Vec::new(); + for (i, k) in keys.iter().enumerate() { + if i % 2 == 0 { + close(&db, k); + } else { + live.push((*k, i as u8)); + } + } + let before = cursor(&db); + + defrag_to_stable(&db); + assert!(cursor(&db) < before); + + // Every survivor keeps its full 8 MiB image byte-for-byte. + for (k, fill) in &live { + let acc = db.loader().load(k).unwrap().unwrap(); + assert_eq!(acc.data().len(), SIZE); + assert!(acc.data().iter().all(|&b| b == *fill)); + } +} diff --git a/accountsdb/src/volatile.rs b/accountsdb/src/volatile.rs new file mode 100644 index 0000000..b6c0c25 --- /dev/null +++ b/accountsdb/src/volatile.rs @@ -0,0 +1,121 @@ +//! In-memory account cache and program ownership sets. + +use std::{ + collections::BTreeSet, + fs::{self, File}, + io::BufReader, + path::Path, +}; + +use ahash::RandomState; +use scc::HashMap; +use solana_account::{AccountMode, AccountSharedData, DirtyMarkers, OwnedAccount, ReadableAccount}; +use solana_pubkey::Pubkey; +use tracing::info; + +use crate::{Result, StoreKind, metrics, snapshot::VOLATILE_DB_FILE}; + +/// Owned accounts keyed by account pubkey. +type AccountsMap = HashMap; +/// Program ownership sets keyed by owner pubkey. +type ProgramsMap = HashMap, RandomState>; + +/// Volatile account store backed by concurrent hash maps. +pub(crate) struct VolatileStore { + /// Current owned accounts. + pub(crate) accounts: AccountsMap, + /// Program owner -> account pubkeys. + pub(crate) programs: ProgramsMap, +} + +impl VolatileStore { + /// Opens the volatile store, optionally bootstrapping from a snapshot file. + /// + /// If `volatile.db` exists, it is loaded into memory and then removed from + /// the snapshot directory so the active tree stays single-sourced. + pub(crate) fn new(path: &Path) -> Result { + const CAP: usize = 2048; + let snapshot = path.join(VOLATILE_DB_FILE); + let accounts: AccountsMap = if snapshot.exists() { + let mut r = BufReader::new(File::open(&snapshot)?); + let accs: AccountsMap = bincode::deserialize_from(&mut r)?; + fs::remove_file(snapshot)?; + info!( + count = accs.len(), + "restored volatile accounts from snapshot" + ); + accs + } else { + AccountsMap::with_capacity_and_hasher(CAP, Default::default()) + }; + + let programs = ProgramsMap::with_capacity_and_hasher(CAP, Default::default()); + accounts.iter_sync(|&pk, acc| { + BTreeSet::insert(&mut programs.entry_sync(acc.owner()).or_default(), pk) + }); + Ok(Self { accounts, programs }) + } + + /// Stores volatile accounts and keeps the program ownership sets in sync. + pub(crate) fn upsert<'a, AC>(&self, accounts: AC) + where + AC: IntoIterator, + { + for (pubkey, account) in accounts { + // The volatile store holds only immutable accounts. An account that + // has flipped to mutable (now lives in persisted storage), or has been + // closed, drops any stale volatile copy. + if account.mutable() || account.is(AccountMode::Closed) { + self.delete(pubkey); + continue; + } + // Immutable accounts stay volatile and update the program mapping. + let mut set = self.programs.entry_sync(*account.owner()).or_default(); + BTreeSet::insert(&mut set, *pubkey); + let Some(prev) = self.accounts.upsert_sync(*pubkey, account.owned()) else { + continue; + }; + if !account.markers().contains(DirtyMarkers::OWNER) { + continue; + } + // Only the old owner set needs cleanup; the new owner was inserted above. + self.programs.remove_if_sync(&prev.owner(), |set| { + set.remove(pubkey); + set.is_empty() + }); + } + metrics::accounts(StoreKind::Volatile, self.accounts.len() as u64); + } + + /// Returns the owned account currently cached for `pubkey`. + pub(crate) fn load(&self, pubkey: &Pubkey) -> Option { + let entry = self.accounts.get_sync(pubkey)?; + Some(entry.get().clone()) + } + + /// Returns whether a volatile account exists for `pubkey`. + pub(crate) fn contains(&self, pubkey: &Pubkey) -> bool { + self.accounts.contains_sync(pubkey) + } + + /// Returns the owned accounts currently mapped to `owner`. + pub(crate) fn program(&self, owner: &Pubkey) -> BTreeSet { + self.programs.read_sync(owner, |_, s| s.clone()).unwrap_or_default() + } + + /// Clears all volatile accounts and owner indexes, so the chain mirror is + /// rebuilt from scratch (see [`AccountsDB::reset`](crate::AccountsDB::reset)). + pub(crate) fn reset(&self) { + self.accounts.clear_sync(); + self.programs.clear_sync(); + } + + /// Removes the cached account and drops its owner mapping. + fn delete(&self, pubkey: &Pubkey) { + let Some(e) = self.accounts.remove_sync(pubkey) else { return }; + self.programs.remove_if_sync(&e.1.owner(), |set| { + set.remove(pubkey); + set.is_empty() + }); + } +} diff --git a/solana/account/src/cow/tests.rs b/solana/account/src/cow/tests.rs deleted file mode 100644 index 89ad2b6..0000000 --- a/solana/account/src/cow/tests.rs +++ /dev/null @@ -1,83 +0,0 @@ -use super::borrowed::BorrowedAccount; -use super::{StorageUnit, init, serialize_buf}; -use crate::AccountBuilder; -use solana_pubkey::Pubkey; -use std::sync::atomic::Ordering::Acquire; - -const BORROWED_LAMPORTS: u64 = 5; -const ACTIVE_DATA: &[u8] = &[1, 2, 3]; -const COMMIT_DATA: &[u8] = &[4, 5, 6]; -const COMMITTED_DATA: &[u8] = &[9, 2, 3]; -const ACTIVE_WRITE: u8 = 9; -const ROLLBACK_WRITE: u8 = 8; -const INITIAL_SEQUENCE: u32 = 0; -const COMMITTED_SEQUENCE: u32 = 1; - -// Serializes an owned account into a borrowed buffer image. -fn make_buf(data: &[u8]) -> Vec { - let owner = Pubkey::new_unique(); - let owned = AccountBuilder::default() - .lamports(BORROWED_LAMPORTS) - .data(data.to_vec()) - .owner(owner) - .build(); - serialize_buf(&owned) -} - -// Reads the active sequence counter. -fn seq(acc: &BorrowedAccount) -> u32 { - // SAFETY: test helpers only call this on a live borrowed buffer. - unsafe { acc.header.as_ref().sequence.load(Acquire) } -} - -// Returns the active image bytes for direct assertions. -fn data(acc: &BorrowedAccount) -> &[u8] { - &acc.data -} - -#[test] -// `init` should read the active image without changing the sequence. -fn test_init_reads_active_image() { - let mut buf = make_buf(ACTIVE_DATA); - let borrowed = init(&mut buf); - - assert_eq!(data(&borrowed), ACTIVE_DATA); - assert_eq!(seq(&borrowed), INITIAL_SEQUENCE); -} - -#[test] -// `translate` should copy the active image into the shadow view, and `commit` should publish it. -fn test_translate_commit_publishes_shadow_image() { - let mut buf = make_buf(ACTIVE_DATA); - let mut borrowed = init(&mut buf); - - // SAFETY: `borrowed` still points at the live borrowed image selected by `init`. - unsafe { borrowed.translate() }; - assert_eq!(seq(&borrowed), INITIAL_SEQUENCE); - - borrowed.data[0] = ACTIVE_WRITE; - borrowed.commit(); - assert_eq!(seq(&borrowed), COMMITTED_SEQUENCE); - - let borrowed = init(&mut buf); - assert_eq!(data(&borrowed), COMMITTED_DATA); -} - -#[test] -// `rollback` should discard shadow writes and restore the active view. -fn test_translate_rollback_discards_shadow_writes() { - let mut buf = make_buf(COMMIT_DATA); - let mut borrowed = init(&mut buf); - - // SAFETY: `borrowed` still points at the live borrowed image selected by `init`. - unsafe { borrowed.translate() }; - borrowed.data[0] = ROLLBACK_WRITE; - - // SAFETY: `reset` is paired with the preceding `translate`. - unsafe { borrowed.reset() }; - assert_eq!(seq(&borrowed), INITIAL_SEQUENCE); - assert_eq!(data(&borrowed), COMMIT_DATA); - - let borrowed = init(&mut buf); - assert_eq!(data(&borrowed), COMMIT_DATA); -}