From 4627a3d3cf1fef9c62306b0e54a11156d4334ede Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Wed, 6 May 2026 13:19:39 +0400 Subject: [PATCH] feat: implements keeper crate - state orchestrator --- Cargo.toml | 3 +- keeper/Cargo.toml | 83 ++++++++ keeper/README.md | 49 +++++ keeper/build.rs | 62 ++++++ keeper/src/accessor.rs | 312 ++++++++++++++++++++++++++++++ keeper/src/builder.rs | 291 ++++++++++++++++++++++++++++ keeper/src/cache.rs | 238 +++++++++++++++++++++++ keeper/src/error.rs | 45 +++++ keeper/src/lib.rs | 179 +++++++++++++++++ keeper/src/metrics.rs | 143 ++++++++++++++ keeper/src/subscriptions.rs | 119 ++++++++++++ keeper/src/testkit.rs | 256 ++++++++++++++++++++++++ keeper/src/tests/caches.rs | 111 +++++++++++ keeper/src/tests/mod.rs | 28 +++ keeper/src/tests/recovery.rs | 129 ++++++++++++ keeper/src/tests/subscriptions.rs | 63 ++++++ keeper/src/util.rs | 107 ++++++++++ nucleus/src/notifier.rs | 43 ++++ 18 files changed, 2260 insertions(+), 1 deletion(-) create mode 100644 keeper/Cargo.toml create mode 100644 keeper/README.md create mode 100644 keeper/build.rs create mode 100644 keeper/src/accessor.rs create mode 100644 keeper/src/builder.rs create mode 100644 keeper/src/cache.rs create mode 100644 keeper/src/error.rs create mode 100644 keeper/src/lib.rs create mode 100644 keeper/src/metrics.rs create mode 100644 keeper/src/subscriptions.rs create mode 100644 keeper/src/testkit.rs create mode 100644 keeper/src/tests/caches.rs create mode 100644 keeper/src/tests/mod.rs create mode 100644 keeper/src/tests/recovery.rs create mode 100644 keeper/src/tests/subscriptions.rs create mode 100644 keeper/src/util.rs create mode 100644 nucleus/src/notifier.rs diff --git a/Cargo.toml b/Cargo.toml index 2a3b251..84197b2 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "accountsdb", + "keeper", "ledger", "nucleus", "programs/magic-root-interface", @@ -25,6 +26,7 @@ version = "0.1.0" [workspace.dependencies] accountsdb = { path = "accountsdb", package = "magicblock-accountsdb" } +keeper = { path = "keeper", package = "magicblock-keeper" } ledger = { path = "ledger", package = "magicblock-ledger" } magic-root-interface = { path = "programs/magic-root-interface" } magic-root-program = { path = "programs/magic-root-program" } @@ -140,7 +142,6 @@ solana-transaction-context = { path = "solana/transaction-context" } missing_docs = "deny" rust_2018_idioms = { level = "warn", priority = -1 } unreachable_pub = "warn" -unsafe_op_in_unsafe_fn = "allow" unused_lifetimes = "warn" unused_macro_rules = "warn" unused_qualifications = "warn" diff --git a/keeper/Cargo.toml b/keeper/Cargo.toml new file mode 100644 index 0000000..52b633b --- /dev/null +++ b/keeper/Cargo.toml @@ -0,0 +1,83 @@ +[package] +name = "magicblock-keeper" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "keeper" + +[features] +# Exposes `keeper::testkit`, the shared keeper-level test harness, as normal code +# so downstream crates can import it via a dev-dependency (no new crate needed). +testkit = [ + "dep:solana-instruction", + "dep:tempfile", + "dep:v42-calculator-interface", + "nucleus/testkit" +] + +[dependencies] +accountsdb = { workspace = true } +ledger = { workspace = true } +nucleus = { workspace = true, features = [ + "ledger-schema", + "metrics", + "notifier", + "runtime", + "shutdown" +] } + + +tempfile = { workspace = true, optional = true } +v42-calculator-interface = { workspace = true, optional = true } + +ahash = { workspace = true } +arc-swap = { workspace = true } +derive_more = { workspace = true, features = ["from"] } +flume = { workspace = true } +parking_lot = { workspace = true } +scc = { workspace = true } +serde = { workspace = true } +tar = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["rt", "sync"] } +tracing = { workspace = true } +zstd = { workspace = true } + +agave-feature-set = { workspace = true, features = ["agave-unstable-api"] } +agave-transaction-view = { workspace = true } +solana-account = { workspace = true, features = ["bincode"] } +solana-feature-gate-interface = { workspace = true } +solana-hash = { workspace = true } +solana-instruction = { workspace = true, optional = true } +solana-keypair = { workspace = true } +solana-message = { workspace = true } +solana-program-runtime = { workspace = true } +solana-pubkey = { workspace = true } +solana-sdk-ids = { workspace = true } +solana-signature = { workspace = true } +solana-signer = { workspace = true } +solana-svm = { workspace = true } +solana-sysvar = { workspace = true } +solana-transaction-error = { workspace = true } + + +[dev-dependencies] +assert_matches = { workspace = true } +nucleus = { workspace = true, features = ["testkit"] } +tempfile = { workspace = true } +v42-calculator-interface = { workspace = true } + +solana-instruction = { workspace = true } +solana-keypair = { workspace = true } +solana-signer = { workspace = true } +tokio = { workspace = true, features = ["macros"] } + +[lints] +workspace = true diff --git a/keeper/README.md b/keeper/README.md new file mode 100644 index 0000000..4b3d805 --- /dev/null +++ b/keeper/README.md @@ -0,0 +1,49 @@ +# `magicblock-keeper` + +`accountsdb` and `ledger` each own one half of the engine's durable state, but +something has to open them together, keep them consistent, and present them as a +single thing the rest of the engine can use. That's the keeper. It owns both +stores plus the read-side caches and subscription fanout, and it deliberately +holds *no storage policy of its own* — every routing and retention decision stays +in the crates below it. The keeper is the seam, not a third store. + +## Startup and recovery + +`KeeperBuilder::build` opens the ledger and accountsdb and then seeds the accounts +the engine can't run without: active features, native builtins, the configured +upgradeable programs (with their ELF), caller-provided accounts, and sysvars. + +Recovery is the part worth understanding. If accountsdb validation reports +corruption on open, the keeper doesn't give up — it saves the active (corrupt) +tree for inspection, restores the latest archived snapshot from the retained +superblocks, and revalidates. This is why accountsdb snapshots are archived +*alongside* sealed superblocks: the ledger's retained history is also the supply +of restore points. + +The blockstore parameters carry the expected block time and a non-zero slot +interval, which is what drives superblock sealing. + +## Finalize + +`finalize` seals the next superblock with the current accounts checksum and +transaction count, then writes and archives an accountsdb snapshot under that +superblock. It requires **exclusive write access** while it runs — the snapshot +has to capture a single coherent point in time, and a concurrent write would let +it record a state that never actually existed. + +## Caches and subscriptions + +Caches are slot-based and lazily evicted on insertion, so stale slots fall out as +new ones arrive rather than needing a sweeper. Subscriptions (accounts, programs, +signatures, logs, transactions) are plain broadcast channels and hold no durable +state — they only fan out what's already happening. + +## `testkit` feature + +Enables `keeper::testkit`, the shared keeper-level test harness (building a real +`Keeper` over throwaway directories, plus the loadable v42 calculator program that +`build.rs` compiles). It also provides shared v42 account and transaction-view +helpers for keeper-backed integration tests. Downstream test suites enable it as +a dev-dependency — `keeper = { workspace = true, features = ["testkit"] }` — +instead of re-deriving the setup. The v42 ELF lives here, so only keeper builds +it. diff --git a/keeper/build.rs b/keeper/build.rs new file mode 100644 index 0000000..8f609ca --- /dev/null +++ b/keeper/build.rs @@ -0,0 +1,62 @@ +//! Builds the v42 calculator SBF program for runtime tests. + +use std::{env, io, path::PathBuf, process::Command}; + +const PROGRAM_DIR: &str = "programs/v42-calculator-program"; +const PROGRAM: &str = "programs/v42-calculator-program/Cargo.toml"; +const SO: &str = "v42_calculator_program.so"; + +fn main() -> Result<(), Box> { + let manifest_dir = PathBuf::from( + env::var_os("CARGO_MANIFEST_DIR") + .ok_or_else(|| io::Error::other("CARGO_MANIFEST_DIR is not set"))?, + ); + let workspace = manifest_dir.parent().ok_or_else(|| { + io::Error::other(format!( + "CARGO_MANIFEST_DIR has no parent: {}", + manifest_dir.display() + )) + })?; + let manifest = workspace.join(PROGRAM); + let artifact = workspace.join("target/deploy").join(SO); + println!( + "cargo:rerun-if-changed={}", + workspace.join(PROGRAM_DIR).display() + ); + + let output = Command::new("cargo") + .arg("build-sbf") + .arg("--manifest-path") + .arg(&manifest) + .arg("--arch") + .arg("v3") + .current_dir(workspace) + // `cargo clippy` exports these wrappers pointing at `clippy-driver`; left in + // place they hijack the SBF toolchain's rustc, which can't resolve the + // `sbpf*-solana` target. Strip them so `build-sbf` uses its own toolchain. + .env_remove("RUSTC_WRAPPER") + .env_remove("RUSTC_WORKSPACE_WRAPPER") + .output() + .map_err(|e| io::Error::other(format!("failed to run `cargo build-sbf`: {e}")))?; + + if !output.status.success() { + return Err(io::Error::other(format!( + "`cargo build-sbf` failed with {}\nstdout:\n{}\nstderr:\n{}", + output.status, + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + )) + .into()); + } + if !artifact.is_file() { + Err(io::Error::other(format!( + "missing SBF artifact: {}", + artifact.display() + )))?; + } + println!( + "cargo:rustc-env=V42_CALCULATOR_PROGRAM_SO={}", + artifact.display() + ); + Ok(()) +} diff --git a/keeper/src/accessor.rs b/keeper/src/accessor.rs new file mode 100644 index 0000000..f55cc95 --- /dev/null +++ b/keeper/src/accessor.rs @@ -0,0 +1,312 @@ +//! Namespaced keeper access APIs. + +use std::{ops::Deref, path::PathBuf, sync::Arc}; + +use accountsdb::{AccountLoader, AccountsDB, AccountsDBError}; +use ledger::{ + request::{ + AccountSignature, AccountSignaturesParams, BlockParams, BlockResponse, ReadRequest, + TransactionResponse, TransactionStatus, + }, + schema::{Event, TransactionEntry}, +}; +use nucleus::{ + Slot, + ledger::Block, + tls::{EncodedMessage, TlsManager}, +}; +use solana_account::{AccountSharedData, ReadableAccount}; +use solana_hash::Hash; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use solana_svm::{ + transaction_execution_result::ExecutedTransaction, + transaction_processing_result::TransactionProcessingResult, +}; +use solana_sysvar::{ + clock::Clock, + slot_hashes::{SlotHashes, SysvarId}, +}; +use tokio::sync::broadcast::Receiver; +use tracing::warn; + +use crate::{ + FullTransaction, Keeper, ResolvedTransaction, + cache::{AccountCache, MissingAccount}, + error::Result, + metrics, + util::{execution_commit, request}, +}; + +/// Account operations namespace. +pub struct AccountsAccessor<'a> { + keeper: &'a Keeper, +} + +/// Cursor over accounts missing from local storage. +/// +/// Each item either gives the caller load ownership or waits for another +/// caller that already owns the load. +pub struct MissingAccounts<'a> { + index: usize, + accounts: &'a [Pubkey], + loader: AccountLoader<'a>, + cache: &'a Arc, +} + +impl<'a> Iterator for MissingAccounts<'a> { + type Item = MissingAccount; + /// Returns the next account that still needs resolution. + fn next(&mut self) -> Option { + loop { + let pubkey = self.accounts.get(self.index)?; + self.index += 1; + if self.loader.contains(pubkey).unwrap_or_default() { + self.cache.promote(pubkey); + continue; + } + return Some(self.cache.reserve(*pubkey)); + } + } +} + +impl<'a> AccountsAccessor<'a> { + pub(crate) fn new(keeper: &'a Keeper) -> Self { + Self { keeper } + } + + /// Loads the current account state for `pubkey`. + pub fn get(&self, pubkey: &Pubkey) -> Result> { + self.loader().load(pubkey).map_err(Into::into) + } + + /// Coordinates loading of accounts missing from local storage. + pub fn ensure(&'a self, accounts: &'a [Pubkey]) -> MissingAccounts<'a> { + MissingAccounts { + index: 0, + accounts, + loader: self.keeper.accountsdb.loader(), + cache: &self.keeper.caches.accounts, + } + } + + /// Returns recent transaction signatures that mention the account. + pub async fn signatures( + &self, + params: AccountSignaturesParams, + ) -> Result> { + Ok(request(self.keeper, params, ReadRequest::AccountSignatures).await??) + } + + /// Subscribes to updates for one account pubkey. + pub async fn subscribe(&self, account: Pubkey) -> Receiver { + self.keeper.subscriptions.accounts.subscribe(account, 8).await + } + + /// Subscribes to account updates for accounts owned by `program`. + pub async fn subscribe_program(&self, program: Pubkey) -> Receiver { + self.keeper.subscriptions.programs.subscribe(program, 16).await + } + + /// Subscribes to account pubkeys evicted from the recent-load cache. + pub fn subscribe_evictions(&self) -> Receiver { + self.keeper.caches.accounts.evictions.subscribe() + } + + /// Subscribes to completed accountsdb snapshot archives. + pub fn subscribe_snapshots(&self) -> Receiver { + self.keeper.subscriptions.snapshots.subscribe() + } + + /// Updates durable `SlotHashes` and `Clock` sysvar accounts from `block`. + /// + /// If either sysvar account is absent, no account updates are stored. + pub fn update_sysvars(&self, block: Block) -> Result<()> { + let loader = self.loader(); + + let Some(mut hacc) = loader.load(&SlotHashes::id())? else { + return Ok(()); + }; + let mut hashes: SlotHashes = hacc.deserialize_data().map_err(AccountsDBError::from)?; + hashes.add(block.slot, block.hash); + hacc.serialize_data(&hashes).map_err(AccountsDBError::from)?; + let Some(mut cacc) = loader.load(&Clock::id())? else { + return Ok(()); + }; + let mut clock: Clock = cacc.deserialize_data().map_err(AccountsDBError::from)?; + clock.slot = block.slot; + clock.unix_timestamp = block.time; + cacc.serialize_data(&clock).map_err(AccountsDBError::from)?; + self.store(&[(SlotHashes::id(), hacc), (Clock::id(), cacc)]).map_err(Into::into) + } +} + +impl Deref for AccountsAccessor<'_> { + type Target = AccountsDB; + + fn deref(&self) -> &Self::Target { + &self.keeper.accountsdb + } +} + +/// Transaction operations namespace. +pub struct TransactionsAccessor<'a> { + keeper: &'a Keeper, +} + +impl<'a> TransactionsAccessor<'a> { + pub(crate) fn new(keeper: &'a Keeper) -> Self { + Self { keeper } + } + + /// Loads the full retained transaction for `signature`. + pub async fn get(&self, signature: Signature) -> Result> { + Ok(request(self.keeper, signature, ReadRequest::Transaction).await??) + } + + /// Loads the retained execution status for `signature`. + pub async fn status(&self, signature: Signature) -> Result> { + if let Some(status) = self.keeper.caches.signatures.get(&signature) { + return Ok(status); + } + Ok(request(self.keeper, signature, ReadRequest::TransactionStatus).await??) + } + + /// Subscribes to all processed transactions. + pub fn subscribe_processed(&self) -> Receiver> { + self.keeper.subscriptions.transactions.subscribe() + } + + /// Subscribes to status updates for one transaction signature. + pub async fn subscribe_signature(&self, signature: Signature) -> Receiver { + self.keeper.subscriptions.signatures.subscribe(signature, 1).await + } + + /// Subscribes to log batches mentioning `account`. + pub async fn subscribe_logs(&self, account: Pubkey) -> Receiver>> { + self.keeper.subscriptions.logs.subscribe(account, 8).await + } + + /// Subscribes to encoded service messages emitted by successful transactions. + pub fn subscribe_service_messages(&self) -> Receiver { + self.keeper.subscriptions.services.subscribe() + } + + /// Appends transaction bytes to the ledger, deduplicating by signature. + /// + /// Returns `Ok(true)` when the transaction was appended, or `Ok(false)` + /// without appending if its signature was already seen for this slot. + /// Execution details are appended later by `commit_execution`. + pub async fn append(&self, transaction: &ResolvedTransaction) -> Result { + let caches = &self.keeper.caches; + let slot = caches.blocks.latest.load().slot + 1; + let signature = transaction.signatures()[0]; + if transaction.data().len() >= ledger::MAX_ENTRY_SIZE { + warn!("transaction binary data size exceeded the limit"); + return Ok(false); + } + if !caches.signatures.push(signature, None, slot) { + return Ok(false); + } + metrics::signature_entries(caches.signatures.len()); + let event = Event::Transaction(TransactionEntry { + signature, + payload: transaction.inner_data().clone(), + }); + self.keeper.ledger.appender.send_async(event).await?; + Ok(true) + } + + /// Commits execution metadata and publishes resulting account changes. + pub fn commit_execution(&self, mut txn: FullTransaction) -> Result<()> { + let commit = execution_commit(&mut txn); + self.keeper.ledger.appender.send(commit.event)?; + + let subscriptions = &self.keeper.subscriptions; + if let Some(execution) = self.commit_state_transitions(&txn.execution.result)? { + let accounts = &execution.loaded_transaction.accounts; + for (pubkey, acc) in accounts { + subscriptions.logs.send(pubkey, &commit.logs, false); + if !acc.dirty() { + continue; + } + subscriptions.accounts.send(pubkey, acc, false); + subscriptions.programs.send(acc.owner(), acc, false); + } + if !subscriptions.transactions.is_empty() { + let _ = subscriptions.transactions.send(Arc::new(txn)); + } + if !subscriptions.services.is_empty() { + while let Some(msg) = TlsManager::dequeue() { + let _ = subscriptions.services.send(msg); + } + } + } + TlsManager::clear(); + subscriptions.signatures.send(&commit.signature, &commit.status, true); + self.keeper.caches.signatures.update(&commit.signature, Some(commit.status)); + Ok(()) + } + + /// Writes the dirty accounts of a successful execution back to accountsdb, + /// returning the execution (for downstream fanout) or `None` if it failed. + pub fn commit_state_transitions<'t>( + &self, + result: &'t TransactionProcessingResult, + ) -> Result> { + let Some(execution) = result.as_ref().ok().filter(|e| e.was_successful()) else { + return Ok(None); + }; + let accounts = &execution.loaded_transaction.accounts; + self.keeper.accountsdb.store(accounts.iter().filter(|(_, a)| a.dirty()))?; + Ok(Some(&**execution)) + } +} + +/// Block operations namespace. +pub struct BlocksAccessor<'a> { + keeper: &'a Keeper, +} + +impl<'a> BlocksAccessor<'a> { + pub(crate) fn new(keeper: &'a Keeper) -> Self { + Self { keeper } + } + + /// Loads a retained block at the requested detail level. + pub async fn get(&self, params: BlockParams) -> Result> { + Ok(request(self.keeper, params, ReadRequest::Block).await??) + } + + /// Returns the latest block boundary known to keeper. + pub fn latest(&self) -> Block { + **self.keeper.caches.blocks.latest.load() + } + + /// Returns the slot currently being built (one past the latest block). + pub fn current_slot(&self) -> Slot { + self.keeper.caches.blocks.latest.load().slot + 1 + } + + /// Returns whether `hash` is in the recent block hash cache (still valid). + pub fn is_valid(&self, hash: &Hash) -> bool { + self.keeper.caches.blocks.history.contains(hash) + } + + /// Subscribes to newly committed slots. + pub fn subscribe(&self) -> Receiver { + self.keeper.subscriptions.blocks.subscribe() + } + + /// Appends a sealed block to the ledger, publishes it to subscribers, and + /// updates the latest-block cache. + pub fn append(&self, block: Block) -> Result<()> { + let event = Event::Block(block); + self.keeper.ledger.appender.send(event)?; + if self.keeper.subscriptions.blocks.receiver_count() != 0 { + let _ = self.keeper.subscriptions.blocks.send(block); + } + self.keeper.caches.blocks.push(block); + Ok(()) + } +} diff --git a/keeper/src/builder.rs b/keeper/src/builder.rs new file mode 100644 index 0000000..662b805 --- /dev/null +++ b/keeper/src/builder.rs @@ -0,0 +1,291 @@ +//! Keeper construction, recovery, and startup account seeding. + +use std::{ + collections::HashMap, + fs::{self, File}, + num::NonZeroU64, + path::PathBuf, + sync::Arc, + time::Duration, +}; + +use accountsdb::{AccountEntry, AccountsDB, AccountsDBError, BackupOp, SnapshotError}; +use agave_feature_set::FeatureSet; +use ledger::{ + Ledger, LedgerHandle, + request::{BlockDetails, BlockParams, ReadRequest, RequestPayload}, +}; +use nucleus::{Slot, ledger::Block, shutdown::ShutdownManager}; +use serde::Serialize; +use solana_account::{AccountBuilder, AccountMode, AccountSharedData, ReadableAccount}; +use solana_feature_gate_interface::Feature; +use solana_keypair::Keypair; +use solana_program_runtime::invoke_context::BuiltinFunctionWithContext; +use solana_pubkey::Pubkey; +use solana_sdk_ids::sysvar; +use solana_signer::Signer; +use solana_sysvar::{ + clock::Clock, + epoch_schedule::EpochSchedule, + rent::Rent, + slot_hashes::{SlotHashes, SysvarId}, +}; +use tracing::{error, info, warn}; + +use crate::{ + Keeper, + cache::{AccountCache, BlocksCache, Caches, ExpiringCache}, + error::Result, + metrics, + subscriptions::Subscriptions, +}; + +const SIGNATURE_CACHE_WINDOW: Duration = Duration::from_secs(75); +const SLOTHASH_ENTRIES: usize = 512; +const BLOCK_CACHE_WINDOW: Duration = Duration::from_secs(60); + +/// Builder for keeper directories and cache timing. +pub struct KeeperBuilder { + /// Funded signer seeded at startup for engine-owned transactions. + pub authority: Arc, + /// Accounts database storage parameters. + pub accountsdb: AccountsDBParams, + /// Ledger storage parameters. + pub ledger: LedgerParams, + /// Block production timing and superblock sealing parameters. + pub blockstore: BlockstoreParams, + /// Native builtin program ids to seed as executable accounts. + pub builtins: HashMap, + /// Upgradeable program accounts to seed, paired as `(program id, ELF bytes)`. + pub programs: HashMap>, + /// Plain accounts to seed into storage before startup completes. + pub accounts: HashMap, + /// Rent parameters used to size seeded accounts and the Rent sysvar. + pub rent: Rent, +} + +/// Accounts database storage parameters. +pub struct AccountsDBParams { + /// Accounts database root directory. + pub directory: PathBuf, +} + +/// Block production timing used by the engine and keeper caches. +#[derive(Clone, Copy)] +pub struct BlockstoreParams { + /// Expected wall-clock interval between produced slots. + pub blocktime: Duration, + /// Number of blocks included into each superblock. + pub superblock: NonZeroU64, +} + +/// Ledger storage parameters. +pub struct LedgerParams { + /// Ledger root directory. + pub directory: PathBuf, + /// Maximum used bytes allowed on the ledger filesystem before eviction runs. + pub size_limit: u64, +} + +impl KeeperBuilder { + /// Open durable stores, recover accounts from the latest snapshot if needed, and wire caches. + pub async fn build(mut self, shutdown: &mut ShutdownManager) -> Result { + let ledger = Ledger::init(&self.ledger.directory, self.ledger.size_limit, shutdown)?; + let accountsdb = self.accountsdb(&ledger)?; + let (block, featureset) = self.prepopulate(&accountsdb, &ledger).await?; + let caches = self.caches(block); + metrics::init(); + Ok(Keeper { + authority: self.authority, + featureset, + rent: self.rent, + accountsdb, + ledger, + caches, + subscriptions: Subscriptions::new(shutdown), + }) + } + + /// Builds read-side caches using blocktime-derived slot TTLs. + fn caches(&self, latest: Block) -> Caches { + let blocktime = self.blockstore.blocktime; + let ttl = |window: Duration| window.div_duration_f64(blocktime).ceil() as Slot; + let blocks = BlocksCache::new(latest, ttl(BLOCK_CACHE_WINDOW)); + let signatures = ExpiringCache::new(ttl(SIGNATURE_CACHE_WINDOW)); + let accounts = Arc::new(AccountCache::default()); + + Caches { signatures, blocks, accounts } + } + + /// Activates the engine's required feature gates at slot 0, seeds a feature + /// account for each, and returns the resulting [`FeatureSet`]. + fn seed_featureset(&self, accounts: &mut Vec) -> Result { + let mut featureset = FeatureSet::default(); + [ + agave_feature_set::curve25519_syscall_enabled::ID, + agave_feature_set::curve25519_restrict_msm_length::ID, + agave_feature_set::enable_poseidon_syscall::ID, + agave_feature_set::enable_sbpf_v3_deployment_and_execution::ID, + agave_feature_set::virtual_address_space_adjustments::ID, + agave_feature_set::get_sysvar_syscall_enabled::ID, + agave_feature_set::ed25519_program_enabled::ID, + agave_feature_set::secp256k1_program_enabled::ID, + agave_feature_set::enable_secp256r1_precompile::ID, + ] + .iter() + .for_each(|f| featureset.activate(f, 0)); + for (&id, &slot) in featureset.active() { + let feature = &Feature { activated_at: Some(slot) }; + let account = self.account(feature, &solana_feature_gate_interface::ID)?; + accounts.push((id, account.build())); + } + Ok(featureset) + } + + /// Seeds accounts needed before the engine starts serving reads. + async fn prepopulate( + &mut self, + accountsdb: &AccountsDB, + ledger: &LedgerHandle, + ) -> Result<(Block, FeatureSet)> { + let mut accounts = Vec::new(); + let featureset = self.seed_featureset(&mut accounts)?; + self.seed_programs(&mut accounts)?; + let block = self.seed_sysvars(accountsdb, ledger, &mut accounts).await?; + let sponsor = AccountBuilder::default().lamports(u64::MAX / 2).mode(AccountMode::System); + accounts.push((self.authority.pubkey(), sponsor.build())); + accounts.extend(self.accounts.drain()); + accountsdb.store(&accounts)?; + Ok((block, featureset)) + } + + /// Seeds builtin and upgradeable program accounts. + fn seed_programs(&self, accounts: &mut Vec) -> Result<()> { + for &builtin in self.builtins.keys() { + let account = self.account(&(), &solana_sdk_ids::native_loader::ID)?; + let account = account.executable(true).build(); + accounts.push((builtin, account)); + } + + for (&program, elf) in &self.programs { + let lamports = self.rent.minimum_balance(elf.len()); + let account = AccountBuilder::default() + .lamports(lamports) + .mode(AccountMode::System) + .owner(solana_sdk_ids::loader_v4::ID) + .executable(true) + .data(elf.clone()); + accounts.push((program, account.build())); + } + Ok(()) + } + + /// Seeds sysvars derived from retained ledger state and keeper config. + /// + /// Returns the latest available block, resolved from accountsdb or ledger + async fn seed_sysvars( + &self, + accountsdb: &AccountsDB, + ledger: &LedgerHandle, + accounts: &mut Vec, + ) -> Result { + let slot = accountsdb.slot(); + let loader = accountsdb.loader(); + let mut last_block = None; + if let Some(hashes) = loader.load(&SlotHashes::id())? { + let hashes = hashes.deserialize_data::().map_err(AccountsDBError::from)?; + if let Some(&(slot, hash)) = hashes.last() { + let time = self.blocktime(ledger, slot).await?; + last_block.replace(Block { slot, hash, time }); + } + } else { + // SlotHashes stores 512 values + let range = slot.saturating_sub(SLOTHASH_ENTRIES as u64)..slot + 1; + let (payload, handle) = RequestPayload::new(range); + ledger.reader.send(ReadRequest::BlockRange(payload))?; + + let mut hashes = SlotHashes::new(&[Default::default(); SLOTHASH_ENTRIES]); + for block in handle.recv_timeout().await?? { + hashes.add(block.slot, block.hash); + last_block.replace(block); + } + let acc = self.account(&hashes, &sysvar::ID)?; + accounts.push((SlotHashes::id(), acc.build())); + } + + let block = last_block.unwrap_or_default(); + // Set the clock slot one ahead from the last + let clock = Clock { + slot: block.slot + 1, + unix_timestamp: block.time, + ..Default::default() + }; + accounts.push((Clock::id(), self.account(&clock, &sysvar::ID)?.build())); + accounts.push((Rent::id(), self.account(&self.rent, &sysvar::ID)?.build())); + accounts.push(( + EpochSchedule::id(), + self.account(&EpochSchedule::default(), &sysvar::ID)?.build(), + )); + Ok(block) + } + + /// Builds a rent-exempt system account containing a serialized sysvar-like state. + fn account(&self, state: &S, owner: &Pubkey) -> Result { + let account = + AccountSharedData::new_data(0, state, owner).map_err(AccountsDBError::from)?; + let lamports = self.rent.minimum_balance(account.data().len()); + Ok(AccountBuilder::from(account).lamports(lamports).mode(AccountMode::System)) + } + + /// Returns the retained block time for the given slot. + async fn blocktime(&self, ledger: &LedgerHandle, slot: Slot) -> Result { + let (payload, handle) = RequestPayload::new(BlockParams { + slot, + details: BlockDetails::None, + }); + ledger.reader.send(ReadRequest::Block(payload))?; + Ok(handle.recv_timeout().await??.map(|r| r.block().time).unwrap_or_default()) + } + /// Opens accountsdb, restoring the newest archived snapshot after corruption. + fn accountsdb(&self, ledger: &LedgerHandle) -> Result { + let mut backup = None; + loop { + let accountsdb = AccountsDB::new(&self.accountsdb.directory)?; + match accountsdb.validate() { + Ok(()) => { + info!("accountsdb validation succeeded"); + backup.map(fs::remove_dir_all).transpose()?; + return Ok(accountsdb); + } + Err(AccountsDBError::Corruption) => { + if backup.is_some() { + error!("accountsdb still corrupt after restore; reverting"); + accountsdb.backup(BackupOp::Restore)?; + return Err(SnapshotError::Missing.into()); + } + warn!("accountsdb corruption detected; restoring from latest snapshot"); + backup.replace(accountsdb.backup(BackupOp::Save)?); + self.unarchive(ledger)?; + } + Err(other) => return Err(other.into()), + } + } + } + + /// Restores the first retained accountsdb snapshot found, from newest to oldest. + fn unarchive(&self, ledger: &LedgerHandle) -> Result<()> { + for superblock in ledger.iter() { + let src = superblock.directory.join("accountsdb.tar.zst"); + if !src.exists() { + continue; + } + let dst = AccountsDB::path(&self.accountsdb.directory); + let file = File::open(src)?; + let mut tar = tar::Archive::new(zstd::Decoder::new(file)?); + tar.unpack(dst)?; + info!(directory = ?superblock.directory, "restored accountsdb snapshot"); + break; + } + Ok(()) + } +} diff --git a/keeper/src/cache.rs b/keeper/src/cache.rs new file mode 100644 index 0000000..3a56979 --- /dev/null +++ b/keeper/src/cache.rs @@ -0,0 +1,238 @@ +//! Read-side caches owned by keeper. + +use std::{collections::VecDeque, hash::Hash, sync::Arc}; + +use arc_swap::ArcSwap; +use ledger::request::TransactionStatus; +use nucleus::{Slot, ledger::Block, notifier::EventNotifier}; +use parking_lot::Mutex; +use scc::{HashCache, HashMap, hash_map::Entry}; +use solana_hash::Hash as SolanaHash; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use tokio::sync::broadcast::{self, Sender}; + +use crate::metrics; + +pub(crate) struct Caches { + /// Recent signature statuses keyed by transaction signature. + pub(crate) signatures: ExpiringCache>, + /// Recent block hashes and latest block boundary. + pub(crate) blocks: BlocksCache, + /// Account recency and missing-load coordination. + pub(crate) accounts: Arc, +} + +/// Account access cache along with missing-account load reservations. +pub(crate) struct AccountCache { + /// Committed account loads, ordered by recent access. + pub(crate) lru: HashCache, + /// In-flight account loads keyed by account pubkey. + pub(crate) reservations: HashMap>, + /// Pubkeys evicted when a committed load displaces a cold account. + pub(crate) evictions: Sender, +} + +impl Default for AccountCache { + fn default() -> Self { + let (evictions, _) = broadcast::channel(32); + Self { + lru: Default::default(), + reservations: Default::default(), + evictions, + } + } +} + +/// Load guard owned by the task responsible for one missing account. +pub struct AccountLoad { + /// Account this guard is responsible for loading. + pub pubkey: Pubkey, + /// Cache reservation released on commit or drop. + cache: Option>, +} + +/// Wait handle for an account currently being loaded by another task. +pub struct AccountWait(Pubkey, Arc); + +/// Coordination state for one account missing from local storage. +pub enum MissingAccount { + /// Caller owns the load; dropping the guard releases waiters. + Load(AccountLoad), + /// Another caller is already loading the account. + Wait(AccountWait), +} + +impl AccountLoad { + /// Marks the load complete and records the account as cached. + pub fn commit(mut self) { + let Some((cache, notifier)) = self.release() else { + return; + }; + if let Ok(Some(evicted)) = cache.lru.put_sync(self.pubkey, ()) { + metrics::account_cache_eviction(); + let _ = cache.evictions.send(evicted.0); + } + notifier.notify(); + } + + fn release(&mut self) -> Option<(Arc, Arc)> { + let cache = self.cache.take()?; + let (_, notifier) = cache.reservations.remove_sync(&self.pubkey)?; + Some((cache, notifier)) + } +} + +impl Drop for AccountLoad { + /// Cancels the load reservation and wakes waiters without caching. + fn drop(&mut self) { + let Some((_, notifier)) = self.release() else { + return; + }; + notifier.notify(); + } +} + +impl AccountWait { + /// Waits for the active loader and returns the loaded account pubkey. + pub async fn wait(self) -> Pubkey { + self.1.notified().await; + self.0 + } +} + +impl AccountCache { + /// Promotes `pubkey` only if it has already been committed to the cache. + pub(crate) fn promote(&self, pubkey: &Pubkey) { + self.lru.get_sync(pubkey); + } + + /// Reserves a missing account load or returns a waiter for the active load. + pub(crate) fn reserve(self: &Arc, pubkey: Pubkey) -> MissingAccount { + match self.reservations.entry_sync(pubkey) { + Entry::Occupied(e) => { + metrics::account_resolution_race(); + MissingAccount::Wait(AccountWait(pubkey, e.get().clone())) + } + Entry::Vacant(e) => { + let notifier = Arc::new(EventNotifier::default()); + e.insert_entry(notifier); + metrics::account_lru_entries(); + MissingAccount::Load(AccountLoad { + pubkey, + cache: Some(self.clone()), + }) + } + } + } +} + +/// Block lookup cache with a lock-free latest-block pointer. +pub(crate) struct BlocksCache { + /// Latest block boundary known at keeper startup or after updates. + pub(crate) latest: ArcSwap, + /// Recent block hash to slot lookups. + pub(crate) history: ExpiringCache, +} + +impl BlocksCache { + /// Creates a block cache seeded with the current latest block. + pub(crate) fn new(block: Block, ttl: Slot) -> Self { + let cache = Self { + latest: ArcSwap::new(block.into()), + history: ExpiringCache::new(ttl), + }; + cache.history.push(block.hash, block.slot, block.slot); + cache + } + + /// Records `block` as the latest and adds its hash to the recent history. + pub(crate) fn push(&self, block: Block) { + self.latest.store(block.into()); + self.history.push(block.hash, block.slot, block.slot); + metrics::block_hash_entries(self.history.len()); + } +} + +/// Concurrent cache with slot-based lazy eviction. +/// +/// Entries are evicted only when another entry is pushed. Re-inserting an +/// existing key leaves its value and expiry slot unchanged. +pub(crate) struct ExpiringCache { + /// Cached values by key. + index: HashMap, + /// Expiry order used for lazy eviction. + queue: Mutex>>, + /// Number of slots each entry lives after insertion. + ttl: Slot, +} + +struct ExpiringRecord { + key: K, + expires: Slot, +} + +impl ExpiringCache { + /// Creates a cache whose entries live for `ttl` slots after insertion. + pub(crate) fn new(ttl: Slot) -> Self { + Self { + index: HashMap::default(), + queue: Default::default(), + ttl, + } + } + + /// Insert a key and evict entries expired at `slot`. + /// + /// Returns `false` if `key` already exists. Existing values and expiry slots + /// are left unchanged. + pub(crate) fn push(&self, key: K, value: V, slot: Slot) -> bool { + let mut queue = self.queue.lock(); + // Lazily evict expired entries from the front of the queue. + while let Some(expired) = queue.pop_front_if(|e| e.expired(slot)) { + self.index.remove_sync(&expired.key); + } + + match self.index.entry_sync(key) { + Entry::Occupied(_) => false, + Entry::Vacant(v) => { + v.insert_entry(value); + queue.push_back(ExpiringRecord::new(key, slot + self.ttl)); + true + } + } + } + + /// Returns current cache entries. + pub(crate) fn len(&self) -> usize { + self.index.len() + } + + /// Returns a cloned cached value for `key`. + pub(crate) fn get(&self, key: &K) -> Option { + self.index.read_sync(key, |_, v| v.clone()) + } + + /// Returns whether a key is currently cached. + pub(crate) fn contains(&self, key: &K) -> bool { + self.index.contains_sync(key) + } + + /// Updates existing cache entry witha a new value. + pub(crate) fn update(&self, key: &K, value: V) { + let Some(mut entry) = self.index.get_sync(key) else { + return; + }; + entry.insert(value); + } +} + +impl ExpiringRecord { + fn new(key: K, expires: Slot) -> Self { + Self { key, expires } + } + + fn expired(&self, instant: Slot) -> bool { + instant >= self.expires + } +} diff --git a/keeper/src/error.rs b/keeper/src/error.rs new file mode 100644 index 0000000..8297a84 --- /dev/null +++ b/keeper/src/error.rs @@ -0,0 +1,45 @@ +//! Keeper error types. + +use accountsdb::{AccountsDBError, SnapshotError}; +use derive_more::From; +use flume::SendError; +use ledger::{LedgerError, LedgerRequestError, request::ReadRequest, schema::Event}; +use nucleus::shutdown::Service; + +/// Errors produced while initializing, finalizing, or serving keeper state. +#[derive(Debug, thiserror::Error, From)] +pub enum KeeperError { + /// Filesystem or archive IO failed. + #[error("io: {0}")] + IO(#[source] std::io::Error), + /// Accounts database operation failed. + #[error("accountsdb: {0}")] + AccountsDB(#[source] AccountsDBError), + /// Snapshot creation, restore, or archive operation failed. + #[error("snapshot: {0}")] + Snapshot(#[source] SnapshotError), + /// Ledger initialization or append failed. + #[error("ledger: {0}")] + Ledger(#[source] LedgerError), + /// A background service is no longer reachable, so the request was dropped. + #[error("service became unavailable: {0:?}")] + ServiceUnavailable(Service), + /// Ledger read request failed before a response was received. + #[error("ledger read request: {0}")] + LedgerRequest(#[source] LedgerRequestError), +} + +impl From> for KeeperError { + fn from(_: SendError) -> Self { + Self::ServiceUnavailable(Service::LedgerAppender) + } +} + +impl From> for KeeperError { + fn from(_: SendError) -> Self { + Self::ServiceUnavailable(Service::LedgerReader) + } +} + +/// Result type used by keeper APIs. +pub type Result = std::result::Result; diff --git a/keeper/src/lib.rs b/keeper/src/lib.rs new file mode 100644 index 0000000..4bd37c3 --- /dev/null +++ b/keeper/src/lib.rs @@ -0,0 +1,179 @@ +#![doc = include_str!("../README.md")] + +use std::{ + fs::{self, File}, + io::Write, + path::PathBuf, + sync::Arc, + thread, +}; + +use agave_feature_set::FeatureSet; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_signer::Signer; +use tokio::sync::mpsc; +use tracing::{error, info, warn}; + +use accountsdb::{AccountsDB, SnapshotError}; +use ledger::{ + LedgerHandle, Superblock, + request::{ReadRequest, ReplayHandle, ReplayParams, RequestPayload}, + schema::Event, +}; +use nucleus::ledger::SuperblockSeal; +use solana_sysvar::rent::Rent; + +use crate::{ + accessor::{AccountsAccessor, BlocksAccessor, TransactionsAccessor}, + cache::Caches, + error::Result, + metrics::Operation, + subscriptions::Subscriptions, +}; + +pub use cache::{AccountLoad, AccountWait, MissingAccount}; +pub use nucleus::runtime::{ + ExecutionRecord, FullTransaction, ResolvedTransaction, TransactionView, +}; + +mod accessor; +pub mod builder; +mod cache; +pub mod error; +mod metrics; +mod subscriptions; +mod util; + +#[cfg(any(test, feature = "testkit"))] +pub mod testkit; + +#[cfg(test)] +mod tests; + +/// Owns the durable state and live access helpers for the execution engine. +pub struct Keeper { + /// Funded signer used for engine-owned transactions and bootstrap accounts. + authority: Arc, + /// Active feature set governing runtime behavior. + featureset: FeatureSet, + /// Rent parameters applied during execution. + rent: Rent, + /// Account state store. + accountsdb: AccountsDB, + /// Ledger worker handles and append path. + ledger: LedgerHandle, + /// Read-side caches shared by accessors. + caches: Caches, + /// Subscription fanout maps for live updates. + subscriptions: Arc, +} + +impl Keeper { + /// Returns the account operations namespace. + pub fn accounts(&self) -> AccountsAccessor<'_> { + AccountsAccessor::new(self) + } + + /// Returns the transaction operations namespace. + pub fn transactions(&self) -> TransactionsAccessor<'_> { + TransactionsAccessor::new(self) + } + + /// Returns the block operations namespace. + pub fn blocks(&self) -> BlocksAccessor<'_> { + BlocksAccessor::new(self) + } + + /// Returns the authority pubkey without exposing signing capability. + pub fn authority(&self) -> Pubkey { + self.authority.pubkey() + } + + /// Returns the authority signer for code paths that must sign engine-owned transactions. + pub fn signer(&self) -> &Keypair { + &self.authority + } + + /// Streams retained ledger entries from the accountsdb slot up to the ledger + /// tip, used to rebuild volatile state on startup. Returns `None` when + /// accountsdb is already current (its slot is at or past the ledger tip). + pub async fn replay(&self) -> Result> { + let ledger_slot = self.ledger.tip().unwrap_or_default(); + let accountsdb_slot = self.accountsdb.slot(); + if accountsdb_slot >= ledger_slot { + return Ok(None); + }; + let (tx, rx) = mpsc::channel(16); + let params = ReplayParams { tx, from: accountsdb_slot }; + let (payload, response) = RequestPayload::new(params); + let handle = ReplayHandle { rx, response }; + self.ledger.reader.send_async(ReadRequest::Replay(payload)).await?; + warn!(accountsdb_slot, ledger_slot, "initiating the ledger replay"); + Ok(Some(handle)) + } + + /// Seal the current superblock and archive the matching accounts snapshot. + /// + /// Must run only when no account store can race the snapshot export; the + /// in-body `SAFETY` note relies on this exclusivity. + pub fn finalize_superblock(&self) -> Result<()> { + let _timer = metrics::time(Operation::FinalizeSuperblock); + let superblock = self.ledger.head() + 1; + // SAFETY: `snapshot` requires exclusive write access to accountsdb, + // i.e. no store operation may race the export. `finalize_superblock` + // is only run when there're no concurrent mutations taking place + let snapshot = unsafe { self.accountsdb.snapshot(superblock) }?; + let checksum = self.accountsdb.checksum(); + let transactions = self.ledger.transactions(); + let seal = SuperblockSeal { checksum, transactions }; + self.ledger.appender.send(Event::Superblock(seal))?; + let path = Superblock::path(&self.ledger.directory, superblock); + fs::create_dir_all(&path)?; + self.archive(snapshot, path)?; + info!(superblock, transactions, "finalized superblock"); + Ok(()) + } + + /// Returns the active feature set. + pub fn features(&self) -> &FeatureSet { + &self.featureset + } + + /// Returns the rent parameters. + pub fn rent(&self) -> &Rent { + &self.rent + } + + /// Flushes account storage and metadata to durable storage. + pub fn flush(&self) -> Result<()> { + self.accountsdb.flush(true).map_err(Into::into) + } + + /// Spawns a background thread that tars and zstd-compresses the accountsdb + /// snapshot at `snapshot` into `target`, removing the snapshot afterward. + fn archive( + &self, + snapshot: PathBuf, + target: PathBuf, + ) -> std::result::Result<(), SnapshotError> { + let path = target.join("accountsdb.tar.zst"); + let dst = File::options().write(true).create(true).truncate(true).open(&path)?; + let snapshots = self.subscriptions.snapshots.clone(); + thread::Builder::new().name("snapshot-archiver".into()).spawn(move || { + { + let _timer = metrics::time(Operation::ArchiveSnapshot); + let mut tar = tar::Builder::new(zstd::Encoder::new(dst, 0)?); + tar.append_dir_all(".", &snapshot)?; + tar.into_inner()?.finish()?.flush()?; + fs::remove_dir_all(snapshot)?; + if snapshots.receiver_count() != 0 { + let _ = snapshots.send(path); + } + Ok::<(), SnapshotError>(()) + } + .inspect_err(|error| error!(?error, "snapshot archival failed")) + })?; + Ok(()) + } +} diff --git a/keeper/src/metrics.rs b/keeper/src/metrics.rs new file mode 100644 index 0000000..a03e0b9 --- /dev/null +++ b/keeper/src/metrics.rs @@ -0,0 +1,143 @@ +//! Prometheus metrics for keeper. + +use std::sync::OnceLock; + +use nucleus::metrics::{self as metric, OperationTimer}; +use nucleus::metrics::{IntCounter, IntGauge, MetricOperation, MetricSpec, OperationCounters}; + +/// Process-wide keeper metrics registered in the default Prometheus registry. +static METRICS: OnceLock = OnceLock::new(); + +/// Operation latency histogram recorded in microseconds. +const OPERATION_TIME: MetricSpec = MetricSpec { + name: "keeper_operation_duration_micros", + help: "Keeper operation duration distribution in microseconds.", +}; +/// Account load cache entries. +const ACCOUNT_CACHE_ENTRIES: MetricSpec = MetricSpec { + name: "keeper_account_cache_entries", + help: "Current account load cache entries.", +}; +/// Signature cache entries. +const SIGNATURE_CACHE_ENTRIES: MetricSpec = MetricSpec { + name: "keeper_signature_cache_entries", + help: "Current signature cache entries.", +}; +/// Block hash cache entries. +const BLOCK_HASH_CACHE_ENTRIES: MetricSpec = MetricSpec { + name: "keeper_block_hash_cache_entries", + help: "Current block hash cache entries.", +}; +/// Account cache eviction counter. +const ACCOUNT_CACHE_EVICTIONS: MetricSpec = MetricSpec { + name: "keeper_account_cache_evictions", + help: "Account cache evictions.", +}; +/// Account resolution conflict counter. +const ACCOUNT_RESOLUTION_RACES: MetricSpec = MetricSpec { + name: "keeper_account_resolution_races", + help: "Account resolution race conditions.", +}; + +/// Keeper operation used as a low-cardinality operation label. +#[derive(Clone, Copy)] +pub(crate) enum Operation { + /// Superblock finalization path. + FinalizeSuperblock, + /// Idle subscription cleanup path. + Cleanup, + /// Snapshot tar/zstd archival path. + ArchiveSnapshot, +} + +impl MetricOperation for Operation { + /// Returns the Prometheus label value for this operation. + fn label(self) -> &'static str { + match self { + Operation::FinalizeSuperblock => "finalize_superblock", + Operation::Cleanup => "cleanup", + Operation::ArchiveSnapshot => "archive_snapshot", + } + } +} + +/// Registers keeper metrics once and seeds gauges from current caches. +pub(crate) fn init() { + METRICS.get_or_init(Default::default); +} + +/// Starts an operation timer that records latency when the returned guard drops. +pub(crate) fn time(op: Operation) -> OperationTimer<'static> { + op.time(METRICS.get().map(|m| &m.operations)) +} + +/// Refreshes account cache and reservation gauges. +pub(crate) fn account_lru_entries() { + metric::with_metrics(&METRICS, |m| m.account_cache_entries.inc()); +} + +/// Refreshes signature cache entry gauge. +pub(crate) fn signature_entries(count: usize) { + metric::with_metrics(&METRICS, |m| m.signature_cache_entries.set(count as i64)); +} + +/// Refreshes block hash cache entry gauge. +pub(crate) fn block_hash_entries(count: usize) { + metric::with_metrics(&METRICS, |m| m.block_hash_cache_entries.set(count as i64)); +} + +/// Records one account cache eviction. +pub(crate) fn account_cache_eviction() { + metric::with_metrics(&METRICS, |m| { + m.account_cache_evictions.inc(); + m.account_cache_entries.dec(); + }); +} + +/// Records one account resolution race condition. +pub(crate) fn account_resolution_race() { + metric::with_metrics(&METRICS, |m| m.account_resolution_race.inc()); +} + +/// Owns all Prometheus collectors registered by keeper. +struct Metrics { + /// Runtime operation duration and completion counters. + operations: OperationCounters, + /// Account cache entry gauge. + account_cache_entries: IntGauge, + /// Signature cache entry gauge. + signature_cache_entries: IntGauge, + /// Block hash cache entry gauge. + block_hash_cache_entries: IntGauge, + /// Account cache eviction counter. + account_cache_evictions: IntCounter, + /// Account resolution race conditions counter. + account_resolution_race: IntCounter, +} +impl Default for Metrics { + /// Builds collectors and registers them in the default Prometheus registry. + fn default() -> Self { + let metrics = Self { + operations: OperationCounters::new(OPERATION_TIME), + account_cache_entries: metric::gauge(ACCOUNT_CACHE_ENTRIES, 0), + signature_cache_entries: metric::gauge(SIGNATURE_CACHE_ENTRIES, 0), + block_hash_cache_entries: metric::gauge(BLOCK_HASH_CACHE_ENTRIES, 0), + account_cache_evictions: metric::counter(ACCOUNT_CACHE_EVICTIONS, 0), + account_resolution_race: metric::counter(ACCOUNT_RESOLUTION_RACES, 0), + }; + metrics.register(); + metrics + } +} + +impl Metrics { + /// Registers all collectors in the default Prometheus registry. + fn register(&self) { + self.operations.register("keeper"); + metric::register("keeper", self.account_cache_entries.clone()); + metric::register("keeper", self.signature_cache_entries.clone()); + metric::register("keeper", self.block_hash_cache_entries.clone()); + metric::register("keeper", self.account_cache_evictions.clone()); + metric::register("keeper", self.account_resolution_race.clone()); + } +} diff --git a/keeper/src/subscriptions.rs b/keeper/src/subscriptions.rs new file mode 100644 index 0000000..f9c160a --- /dev/null +++ b/keeper/src/subscriptions.rs @@ -0,0 +1,119 @@ +//! Broadcast channels for live read-side notifications. + +use std::{hash::Hash, path::PathBuf, sync::Arc, time::Duration}; + +use ahash::RandomState; +use ledger::{request::TransactionStatus, schema::Block}; +use nucleus::{ + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + tls::EncodedMessage, +}; +use scc::HashMap; +use solana_account::AccountSharedData; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use tokio::{ + sync::broadcast::{self, Receiver, Sender}, + time::{MissedTickBehavior, interval}, +}; + +use crate::{ + FullTransaction, + metrics::{self, Operation}, +}; + +/// Live notification channels owned by keeper. +pub(crate) struct Subscriptions { + /// Account updates keyed by account pubkey. + pub(crate) accounts: Subscribers, + /// Program account updates keyed by owner pubkey. + pub(crate) programs: Subscribers, + /// Signature status updates keyed by transaction signature. + pub(crate) signatures: Subscribers, + /// Log broadcasts keyed by mentioned program or account pubkey. + pub(crate) logs: Subscribers>>, + /// Broadcast channel for newly committed slots. + pub(crate) blocks: Sender, + /// Broadcast channel for all committed transactions. + pub(crate) transactions: Sender>, + /// Accountsdb snapshot archive completions, sent after compression finishes. + pub(crate) snapshots: Sender, + /// Encoded service messages emitted during successful transaction execution. + pub(crate) services: Sender, +} + +/// Lazily-created per-key broadcast channel map. +pub(crate) struct Subscribers(HashMap, RandomState>); + +impl Default for Subscribers { + fn default() -> Self { + Self(Default::default()) + } +} + +impl Subscribers { + /// Returns a receiver for `key`, creating its broadcast channel on demand. + pub(crate) async fn subscribe(&self, key: K, cap: usize) -> Receiver { + self.0 + .entry_async(key) + .await + .or_insert_with(|| broadcast::channel(cap).0) + .subscribe() + } + + /// Sends the provided updated to all of the active subscribers + #[inline] + pub(crate) fn send(&self, key: &K, value: &V, oneshot: bool) { + if self.0.is_empty() { + return; + } + let sender = |_: &K, tx: &Sender| tx.send(value.clone()).is_ok(); + let success = self.0.read_sync(key, sender).unwrap_or(true); + if !success || oneshot { + self.0.remove_sync(key); + } + } +} + +impl Subscriptions { + /// Builds subscription channels and starts cleanup for idle keyed entries. + pub(crate) fn new(shutdown: &mut ShutdownManager) -> Arc { + let (transactions, _) = broadcast::channel(1024); + let (blocks, _) = broadcast::channel(32); + let (snapshots, _) = broadcast::channel(4); + let (services, _) = broadcast::channel(64); + let subs = Arc::new(Self { + accounts: Default::default(), + programs: Default::default(), + signatures: Default::default(), + logs: Default::default(), + blocks, + transactions, + snapshots, + services, + }); + let shutdown = shutdown.handle(Service::SubscriptionsCleanup); + tokio::spawn(cleanup(subs.clone(), shutdown)); + subs + } +} + +/// Drops keyed broadcast channels after their last receiver is gone. +async fn cleanup(subscriptions: Arc, mut shutdown: ShutdownHandle) { + let mut ticker = interval(Duration::from_secs(60)); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + loop { + tokio::select! { + biased; + _ = shutdown.signalled() => break, + _ = ticker.tick() => { + let _timer = metrics::time(Operation::Cleanup); + subscriptions.accounts.0.retain_async(|_, s| s.receiver_count() != 0).await; + subscriptions.programs.0.retain_async(|_, s| s.receiver_count() != 0).await; + subscriptions.signatures.0.retain_async(|_, s| s.receiver_count() != 0).await; + subscriptions.logs.0.retain_async(|_, s| s.receiver_count() != 0).await; + } + } + } + shutdown.terminate(ShutdownReason::Signalled); +} diff --git a/keeper/src/testkit.rs b/keeper/src/testkit.rs new file mode 100644 index 0000000..eb20d8d --- /dev/null +++ b/keeper/src/testkit.rs @@ -0,0 +1,256 @@ +//! Keeper-level test harness shared by keeper and processor test suites. +//! +//! Builds a real [`Keeper`] over throwaway directories with the canonical test +//! parameters (retention disabled, 400 ms blocktime, superblock 16), and exposes +//! the loadable v42 calculator program guaranteed by `build.rs`. The low-level, +//! engine-agnostic builders (transactions, blocks, tempdirs) are re-exported from +//! [`nucleus::testkit`]. Compiled only under the `testkit` feature (or a crate's +//! own `cfg(test)`), so it never reaches release builds. +#![allow(clippy::expect_used)] + +use std::{ + collections::HashMap, + num::NonZeroU64, + path::{Path, PathBuf}, + sync::Arc, + time::Duration, +}; + +use derive_more::Deref; +use nucleus::{shutdown::ShutdownManager, testkit::sign_instruction}; +use solana_account::{AccountBuilder, AccountMode, OwnedAccount, ReadableAccount}; +use solana_hash::Hash; +use solana_instruction::{AccountMeta, Instruction}; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use solana_signer::Signer; +use solana_sysvar::rent::Rent; + +pub use nucleus::testkit::{TempDir, V42_ID, block, init_tracing, tempdir, transaction}; +use tokio::{sync::broadcast::Receiver, time}; + +use crate::{ + Keeper, ResolvedTransaction, TransactionView, + builder::{AccountsDBParams, BlockstoreParams, KeeperBuilder, LedgerParams}, +}; + +/// The v42 calculator ELF, built and located by `keeper/build.rs`. +pub const V42_PROGRAM_ELF: &[u8] = include_bytes!(env!("V42_CALCULATOR_PROGRAM_SO")); + +/// Throwaway on-disk homes for the accountsdb and ledger stores. +/// +/// The directories must outlive every keeper opened over them — the stores keep +/// their files open/mmapped — which is why the recovery tests hold `Dirs` across +/// a full close-and-reopen cycle. +pub struct Dirs { + /// Accounts database directory. + pub accounts: TempDir, + /// Ledger directory. + pub ledger: TempDir, +} + +impl Default for Dirs { + fn default() -> Self { + Self { + accounts: tempdir(), + ledger: tempdir(), + } + } +} + +/// A keeper builder over `dirs` with retention disabled and a 400 ms blocktime. +/// +/// `builtins`, `programs`, and `accounts` default to empty; individual tests fill +/// them as needed (e.g. via [`seed_v42`]). [`TestKeeper::new`] is the seeded path. +pub fn keeper_builder(dirs: &Dirs) -> KeeperBuilder { + let mut programs = HashMap::new(); + programs.insert(V42_ID, V42_PROGRAM_ELF.to_vec()); + init_tracing(); + + KeeperBuilder { + authority: Keypair::new().into(), + accountsdb: AccountsDBParams { + directory: dirs.accounts.path().to_owned(), + }, + ledger: LedgerParams { + directory: dirs.ledger.path().to_owned(), + size_limit: u64::MAX, + }, + blockstore: BlockstoreParams { + blocktime: Duration::from_millis(400), + superblock: NonZeroU64::new(16).expect("non-zero superblock"), + }, + builtins: Default::default(), + programs, + accounts: Default::default(), + rent: Rent::default(), + } +} + +/// A built keeper together with the directories and shutdown manager keeping its +/// background services alive. Derefs to [`Keeper`] for accessor calls. +/// +/// The keeper is seeded at construction with the v42 program and one funded +/// payer, so keeper-backed suites can build and run transactions immediately +/// instead of re-loading the ELF or re-storing a signer per test. +#[derive(Deref)] +pub struct TestKeeper { + _dirs: Dirs, + shutdown: ShutdownManager, + #[deref] + keeper: Arc, +} + +impl TestKeeper { + /// Builds a keeper on fresh directories seeded with v42 and a funded payer. + pub async fn new() -> Self { + let dirs = Dirs::default(); + let mut builder = keeper_builder(&dirs); + let payer = Keypair::new(); + builder.accounts.insert( + payer.pubkey(), + AccountBuilder::default().lamports(1_000_000).build(), + ); + let mut shutdown = ShutdownManager::default(); + let keeper = Arc::new(builder.build(&mut shutdown).await.expect("keeper builds")); + Self { _dirs: dirs, shutdown, keeper } + } + + /// Shared engine state, cloneable into processor services. + pub fn state(&self) -> Arc { + self.keeper.clone() + } + + /// The shutdown manager owning this keeper's background services, for wiring + /// additional services (sequencer/simulator) onto the same lifecycle. + pub fn shutdown(&mut self) -> &mut ShutdownManager { + &mut self.shutdown + } + + /// The funded fee payer seeded at construction. + pub fn payer(&self) -> &Keypair { + &self.authority + } + + /// Drops keeper-owned service channels and waits for background services. + pub async fn close(self) { + let Self { mut shutdown, keeper, .. } = self; + keeper.flush().expect("disk flush should succeed"); + drop(keeper); + shutdown.terminate().await; + } +} + +/// A signed, resolved no-op transaction and its first signature, built the same +/// way the sequencer resolves inbound transactions. +pub fn signed_tx() -> (Signature, ResolvedTransaction) { + let (signature, bytes) = transaction(&[]); + let view = TransactionView::try_new_sanitized(bytes, true).expect("transaction sanitizes"); + let resolved = + ResolvedTransaction::try_new(view, Some(Default::default()), &Default::default()) + .expect("transaction resolves"); + (signature, resolved) +} + +/// A resolved transaction whose account metadata matches `accounts`. +/// +/// Each tuple is `(pubkey, writable)`. The transaction is fully sanitized and +/// resolved so scheduling sees the same account flags the keeper resolution path +/// would produce. A fresh random program id per call keeps the referenced account +/// set disjoint from other transactions under test. +pub fn resolved(accounts: &[(Pubkey, bool)]) -> ResolvedTransaction { + let payer = Keypair::new(); + let program = Pubkey::new_unique(); + let metas = accounts + .iter() + .map(|(key, writable)| { + if *writable { + AccountMeta::new(*key, false) + } else { + AccountMeta::new_readonly(*key, false) + } + }) + .collect(); + let ix = Instruction::new_with_bytes(program, &[], metas); + let (_signature, bytes) = sign_instruction(&payer, ix, Hash::default()); + let view = TransactionView::try_new_sanitized(bytes, true).expect("transaction sanitizes"); + ResolvedTransaction::try_new(view, Some(Default::default()), &Default::default()) + .expect("transaction resolves") +} + +/// A delegated, v42-owned account carrying an 8-byte little-endian `u64`. +pub fn v42_owned(value: u64, mode: AccountMode) -> OwnedAccount { + AccountBuilder::default() + .lamports(Rent::default().minimum_balance(8)) + .owner(V42_ID) + .mode(mode) + .data(value.to_le_bytes().to_vec()) + .build() +} + +/// Stores a delegated v42 `u64` account and returns its pubkey. +pub fn store_v42(state: &Keeper, value: u64, mode: AccountMode) -> Pubkey { + let key = Pubkey::new_unique(); + let acc = v42_owned(value, mode).into(); + state.accounts().store(&[(key, acc)]).expect("account stores"); + key +} + +/// Reads the little-endian `u64` payload of a v42 account, or `None` if absent. +pub fn load_v42(state: &Keeper, key: Pubkey) -> Option { + state.accounts().get(&key).expect("account loads").as_ref().map(decode_v42) +} + +/// Decodes the little-endian `u64` payload stored in a v42 account. +pub fn decode_v42(account: &impl ReadableAccount) -> u64 { + u64::from_le_bytes(account.data()[..8].try_into().expect("u64 data")) +} + +/// Signs `instruction` into the sanitized transaction view consumed by services. +pub fn signed_view( + payer: &Keypair, + instruction: Instruction, + blockhash: Hash, +) -> (Signature, TransactionView) { + let (signature, bytes) = sign_instruction(payer, instruction, blockhash); + let view = TransactionView::try_new_sanitized(bytes, true).expect("transaction sanitizes"); + (signature, view) +} + +/// Poisons the persisted checksum so the next open reports corruption. +/// +/// The mmap-backed persisted store keeps its `DatabaseMeta` at the front of +/// `CURRENT/storage.db`; the `u64` format version sits at offset 0 and the +/// recorded checksum immediately after it at offset 8. Overwriting the checksum +/// word (leaving the version intact) makes it disagree with the recomputed value +/// on reopen — the exact `AccountsDBError::Corruption` the recovery path keys on, +/// rather than an `UnsupportedVersion` it does not handle. The caller must have +/// flushed a valid checksum first (e.g. via `finalize_superblock`, whose snapshot +/// flushes synchronously), else recovery would trip on pre-existing state instead. +pub fn corrupt(accounts_dir: &Path) { + use std::io::{Seek, SeekFrom, Write}; + let path = accounts_dir.join("CURRENT").join("storage.db"); + let mut file = std::fs::OpenOptions::new().write(true).open(&path).expect("open storage.db"); + file.seek(SeekFrom::Start(8)).expect("seek to checksum field"); + file.write_all(&[0xAB; 8]).expect("write garbage over checksum"); + file.flush().expect("flush corruption"); +} + +/// Returns the archived accountsdb snapshot path under any retained superblock, +/// or `None` when no superblock directory holds one yet. +pub fn archived_snapshot(ledger_dir: &Path) -> Option { + std::fs::read_dir(ledger_dir) + .expect("read ledger dir") + .filter_map(Result::ok) + .map(|e| e.path().join("accountsdb.tar.zst")) + .find(|p| p.exists()) +} + +/// Waits until a subscribed detached snapshot archiver reports completion. +pub async fn await_archive(mut tx: Receiver) { + let result = time::timeout(Duration::from_secs(8), tx.recv()).await; + result + .expect("timeout waiting for snapshot") + .expect("snapshot subscription failed"); +} diff --git a/keeper/src/tests/caches.rs b/keeper/src/tests/caches.rs new file mode 100644 index 0000000..21ef0da --- /dev/null +++ b/keeper/src/tests/caches.rs @@ -0,0 +1,111 @@ +//! Read-side cache primitives keeper owns: the slot-based `ExpiringCache` and the +//! `AccountCache` missing-load coordination. + +use std::sync::Arc; + +use solana_account::AccountBuilder; +use solana_pubkey::Pubkey; + +use super::TestKeeper; +use crate::cache::{AccountCache, AccountLoad, AccountWait, ExpiringCache, MissingAccount}; + +// `ExpiringCache` evicts lazily on push, never on read; re-inserting an existing +// key is a no-op; `update` replaces only present values. +#[test] +fn expiring_cache_lazy_eviction() { + // ttl = 2 slots: a key pushed at slot s expires at s + 2. + let cache: ExpiringCache = ExpiringCache::new(2); + + assert!(cache.push(1, 10, 0)); // inserted, expires at slot 2 + assert!(!cache.push(1, 99, 0)); // re-insert of an existing key is a no-op + assert_eq!( + cache.get(&1), + Some(10), + "value left unchanged by the re-insert" + ); + + // Eviction runs only on push: at slot 5 the entry is well past its expiry but + // stays readable until the next push sweeps the queue. + assert!(cache.contains(&1)); + assert_eq!(cache.get(&1), Some(10)); + + // A push at slot 5 first evicts everything expired at 5 (key 1), then inserts. + assert!(cache.push(2, 20, 5)); + assert!(!cache.contains(&1), "expired key swept on the next push"); + assert_eq!(cache.get(&2), Some(20)); + + // `update` replaces a present value and no-ops for an absent key. + cache.update(&2, 21); + assert_eq!(cache.get(&2), Some(21)); + cache.update(&404, 0); + assert!(!cache.contains(&404)); + + // A key re-admitted after expiry is a fresh insert again. + assert!(cache.push(1, 11, 5)); + assert_eq!(cache.get(&1), Some(11)); +} + +// Two callers racing on the same missing account get exactly one loader and one +// waiter; committing caches the account while dropping the load guard does not. +#[tokio::test] +async fn account_load_release_paths_wake_waiters() { + for commit in [true, false] { + let cache = Arc::new(AccountCache::default()); + let pk = Pubkey::new_unique(); + let (load, wait) = reserve_load_and_wait(&cache, pk); + + let waiter = tokio::spawn(async move { wait.wait().await }); + if commit { + load.commit(); + } else { + drop(load); + } + + assert_eq!(waiter.await.unwrap(), pk, "waiter wakes with the pubkey"); + assert_eq!(cache.lru.get_sync(&pk).is_some(), commit); + assert!(matches!(cache.reserve(pk), MissingAccount::Load(_))); + } +} + +// `ensure` is the production seam over `AccountCache`: it skips accounts already +// resident in storage (promoting them) and hands back a coordination item only +// for the ones missing, with the first caller owning the load. +#[tokio::test] +async fn ensure_reserves_only_missing_accounts() { + let keeper = TestKeeper::new().await; + let present = Pubkey::new_unique(); + let missing = Pubkey::new_unique(); + keeper + .accounts() + .store(&[(present, AccountBuilder::default().lamports(1).build())]) + .unwrap(); + + // The accessor must outlive the iterator that borrows it. + let accounts = keeper.accounts(); + let reserved: Vec<_> = accounts.ensure(&[present, missing]).collect(); + + // The resident account is skipped entirely; only the missing one surfaces, + // and the first caller to reach it owns the load. + assert_eq!(reserved.len(), 1, "only the missing account is reserved"); + let MissingAccount::Load(load) = &reserved[0] else { + panic!("first reservation of a missing account owns the load"); + }; + assert_eq!(load.pubkey, missing); + + // The reservation stays live while the load guard is held, so a concurrent + // `ensure` of the same account waits instead of racing a second load. + let again: Vec<_> = accounts.ensure(&[missing]).collect(); + assert!(matches!(again.as_slice(), [MissingAccount::Wait(_)])); + + keeper.close().await; +} + +fn reserve_load_and_wait(cache: &Arc, pk: Pubkey) -> (AccountLoad, AccountWait) { + let MissingAccount::Load(load) = cache.reserve(pk) else { + panic!("first reservation must own the load"); + }; + let MissingAccount::Wait(wait) = cache.reserve(pk) else { + panic!("concurrent reservation must wait"); + }; + (load, wait) +} diff --git a/keeper/src/tests/mod.rs b/keeper/src/tests/mod.rs new file mode 100644 index 0000000..8166765 --- /dev/null +++ b/keeper/src/tests/mod.rs @@ -0,0 +1,28 @@ +//! Keeper integration and unit tests. +//! +//! These cover the composition layer keeper owns — startup seeding, corruption +//! recovery, the read-side caches, and subscription fanout — and deliberately +//! avoid re-testing the accountsdb/ledger internals already covered below it. + +mod caches; +mod recovery; +mod subscriptions; + +use nucleus::shutdown::ShutdownManager; + +use crate::{ + Keeper, + testkit::{Dirs, TestKeeper, archived_snapshot, corrupt, keeper_builder, signed_tx}, +}; + +/// Closes the keeper and blocks until every ledger worker has drained and +/// exited, so a subsequent reopen observes a fully flushed ledger. +/// +/// Dropping the keeper closes the appender/reader channels, which drains any +/// buffered superblock seal and lets those workers finish; cancelling the shared +/// token releases the subscription-cleanup task. `wait` then joins all of their +/// termination reports (bounded by the manager's own termination timeout). +async fn teardown(mut shutdown: ShutdownManager, keeper: Keeper) { + drop(keeper); + shutdown.terminate().await; +} diff --git a/keeper/src/tests/recovery.rs b/keeper/src/tests/recovery.rs new file mode 100644 index 0000000..5ee483a --- /dev/null +++ b/keeper/src/tests/recovery.rs @@ -0,0 +1,129 @@ +//! Startup seeding, corruption recovery + +use nucleus::shutdown::ShutdownManager; +use solana_account::{AccountBuilder, AccountMode, ReadableAccount}; +use solana_pubkey::Pubkey; +use solana_sdk_ids::{loader_v4, sysvar}; +use solana_sysvar::{ + clock::Clock, epoch_schedule::EpochSchedule, rent::Rent, slot_hashes::SysvarId, +}; + +use crate::testkit::await_archive; + +use super::{Dirs, archived_snapshot, corrupt, keeper_builder, teardown}; + +// Startup seeds the engine's required feature gates, the configured upgradeable +// programs, and the sysvars, with the exact ownership/rent/clock-offset shape the +// rest of the engine assumes. +#[tokio::test] +async fn seeds_features_programs_and_sysvars() { + let dirs = Dirs::default(); + let mut builder = keeper_builder(&dirs); + let program = Pubkey::new_unique(); + let elf = vec![1u8, 2, 3, 4, 5, 6, 7, 8]; + builder.programs.insert(program, elf.clone()); + let mut shutdown = ShutdownManager::default(); + let keeper = builder.build(&mut shutdown).await.expect("keeper builds"); + let rent = Rent::default(); + + // The engine's required curve25519/precompile/sbpf/sysvar gates are all + // active at slot 0, and every active feature is backed by a rent-exempt + // feature-gate-owned account. + let required = [ + agave_feature_set::curve25519_syscall_enabled::ID, + agave_feature_set::enable_sbpf_v3_deployment_and_execution::ID, + agave_feature_set::get_sysvar_syscall_enabled::ID, + agave_feature_set::ed25519_program_enabled::ID, + agave_feature_set::secp256k1_program_enabled::ID, + ]; + for id in required { + assert_eq!( + keeper.features().active().get(&id), + Some(&0), + "required gate active at slot 0" + ); + } + for (&id, &slot) in keeper.features().active() { + assert_eq!(slot, 0, "features activate at slot 0"); + let acc = keeper.accounts().get(&id).unwrap().expect("feature account seeded"); + assert_eq!(acc.owner(), &solana_feature_gate_interface::ID); + assert!(acc.lamports() >= rent.minimum_balance(acc.data().len())); + } + + // The upgradeable program account carries its ELF verbatim, is executable, + // owned by loader_v4 (not the BPF upgradeable loader), and rent-exempt. + // Builtins are seeded through the same path with an executable native-loader + // account, so they share this shape. + let acc = keeper.accounts().get(&program).unwrap().expect("program seeded"); + assert!(acc.executable()); + assert_eq!(acc.owner(), &loader_v4::ID); + assert_eq!(acc.data(), elf.as_slice()); + assert_eq!(acc.lamports(), rent.minimum_balance(elf.len())); + + // The Clock is seeded one slot ahead of the last block; a fresh ledger's last + // block defaults to slot 0, so the clock starts at slot 1. + let clock: Clock = keeper + .accounts() + .get(&Clock::id()) + .unwrap() + .expect("clock seeded") + .deserialize_data() + .unwrap(); + assert_eq!(clock.slot, 1); + + // Rent and EpochSchedule sysvars are present and sysvar-owned. + for id in [Rent::id(), EpochSchedule::id()] { + let acc = keeper.accounts().get(&id).unwrap().expect("sysvar seeded"); + assert_eq!(acc.owner(), &sysvar::ID); + } + + teardown(shutdown, keeper).await; +} + +// A corrupt accountsdb on open is restored from the newest archived snapshot, +// and the saved corrupt tree is discarded once the restored store revalidates. +#[tokio::test] +async fn recovers_the_newest_snapshot() { + let dirs = Dirs::default(); + let marker = Pubkey::new_unique(); + + let mut shutdown = ShutdownManager::default(); + let keeper = keeper_builder(&dirs).build(&mut shutdown).await.expect("keeper builds"); + let state1 = AccountBuilder::default().lamports(1).mode(AccountMode::Delegated); + let state2 = AccountBuilder::default().lamports(2).mode(AccountMode::Delegated); + // First snapshot captures marker == 1. + keeper.accounts().store(&[(marker, state1.build())]).unwrap(); + let snapshots = keeper.accounts().subscribe_snapshots(); + keeper.finalize_superblock().expect("first finalize"); + await_archive(snapshots).await; + assert!( + archived_snapshot(dirs.ledger.path()).is_some(), + "snapshot archived under superblock" + ); + // Second snapshot, in a later superblock, captures marker == 2. + keeper.accounts().store(&[(marker, state2.build())]).unwrap(); + let snapshots = keeper.accounts().subscribe_snapshots(); + keeper.finalize_superblock().expect("second finalize"); + await_archive(snapshots).await; + teardown(shutdown, keeper).await; + + assert_eq!( + recover_lamports(&dirs, marker).await, + 2, + "newest snapshot wins" + ); +} + +async fn recover_lamports(dirs: &Dirs, marker: Pubkey) -> u64 { + corrupt(dirs.accounts.path()); + + let mut shutdown = ShutdownManager::default(); + let keeper = keeper_builder(dirs).build(&mut shutdown).await.expect("keeper recovers"); + keeper.accounts().validate().expect("restored store validates"); + let restored = keeper.accounts().get(&marker).unwrap().expect("marker restored"); + let lamports = restored.lamports(); + // The corrupt tree saved for inspection is removed on successful recovery. + assert!(!dirs.accounts.path().join("CURRENT.bkp").exists()); + teardown(shutdown, keeper).await; + lamports +} diff --git a/keeper/src/tests/subscriptions.rs b/keeper/src/tests/subscriptions.rs new file mode 100644 index 0000000..122d6c9 --- /dev/null +++ b/keeper/src/tests/subscriptions.rs @@ -0,0 +1,63 @@ +//! Subscription fanout primitives, transaction-append dedup + +use super::{TestKeeper, signed_tx}; +use crate::subscriptions::Subscribers; + +// A keyed broadcast map creates channels lazily, drops the channel after a +// oneshot send or once its last receiver is gone, and treats an absent key as a +// successful no-op. +#[tokio::test] +async fn subscribers_send_semantics() { + let subs: Subscribers = Subscribers::default(); + + // Sending to a key with no channel is a no-op (no panic, nothing removed). + subs.send(&1, &10, false); + + // A subscriber receives ordinary (non-oneshot) sends on its key. + let mut rx = subs.subscribe(1, 4).await; + subs.send(&1, &11, false); + assert_eq!(rx.recv().await.unwrap(), 11); + + // A oneshot send delivers, then removes the channel; a fresh subscribe gets a + // new channel that never sees the terminal value. + subs.send(&1, &12, true); + assert_eq!(rx.recv().await.unwrap(), 12); + let mut rx2 = subs.subscribe(1, 4).await; + subs.send(&1, &13, false); + assert_eq!(rx2.recv().await.unwrap(), 13); + + // A send to a key whose only receiver has been dropped removes the dead + // channel; the next subscribe recreates it without redelivering old values. + let mut rx3 = subs.subscribe(2, 4).await; + subs.send(&2, &20, false); + assert_eq!(rx3.recv().await.unwrap(), 20); + drop(rx3); + subs.send(&2, &21, false); // dead channel -> removed + let mut rx4 = subs.subscribe(2, 4).await; + subs.send(&2, &22, false); + assert_eq!(rx4.recv().await.unwrap(), 22); +} + +// Appending a transaction records a `Some(None)` sentinel in the signature cache, +// so a same-slot re-append is deduplicated and `status` is served from the cache +// as "seen, no status yet" without a ledger read. +#[tokio::test] +async fn append_dedup_and_status_sentinel() { + let keeper = TestKeeper::new().await; + let (signature, txn) = signed_tx(); + + // First append writes to the ledger; the duplicate is dropped. + assert!( + keeper.transactions().append(&txn).await.unwrap(), + "first append is accepted" + ); + assert!( + !keeper.transactions().append(&txn).await.unwrap(), + "duplicate is deduplicated" + ); + + // The sentinel makes status() return None from the cache. + assert!(keeper.transactions().status(signature).await.unwrap().is_none()); + + keeper.close().await; +} diff --git a/keeper/src/util.rs b/keeper/src/util.rs new file mode 100644 index 0000000..d73906e --- /dev/null +++ b/keeper/src/util.rs @@ -0,0 +1,107 @@ +//! Internal helpers shared by keeper accessors. + +use std::sync::Arc; + +use ledger::{ + request::{ReadRequest, RequestPayload, TransactionStatus}, + schema::{ + Balances, CompiledInstruction, Cpis, Event, Execution, ExecutionDetails, ExecutionHeader, + Instruction, ReturnData, + }, +}; +use solana_message::inner_instruction::{InnerInstruction, InnerInstructionsList}; +use solana_signature::Signature; +use solana_svm::{ + transaction_balances::BalanceCollector, transaction_execution_result::ExecutedTransaction, + transaction_processing_result::TransactionProcessingResultExtensions, +}; + +use crate::{FullTransaction, Keeper, Result}; + +/// Ledger event, status-cache entry, and logs derived from one execution result. +pub(crate) struct ExecutionCommit { + /// First transaction signature used for status notifications. + pub(crate) signature: Signature, + /// Status stored in the signature cache and sent to subscribers. + pub(crate) status: TransactionStatus, + /// Ledger event that pairs execution metadata with the appended transaction. + pub(crate) event: Event, + /// Execution logs shared with account log subscribers. + pub(crate) logs: Arc>, +} + +/// Sends a typed read request to the ledger reader and waits for its response. +pub(crate) async fn request(keeper: &Keeper, params: P, request: F) -> Result +where + F: FnOnce(RequestPayload) -> ReadRequest, +{ + let (payload, handle) = RequestPayload::::new(params); + let request = request(payload); + keeper.ledger.reader.send_async(request).await?; + Ok(handle.recv().await?) +} + +/// Builds the ledger and cache records for a completed transaction execution. +pub(crate) fn execution_commit(txn: &mut FullTransaction) -> ExecutionCommit { + let slot = txn.execution.slot; + let result = txn.execution.result.flattened_result(); + let signature = txn.transaction.signatures()[0]; + + let header = ExecutionHeader { + signature, + slot, + result: result.clone(), + }; + let status = TransactionStatus { result, slot }; + let details = txn + .execution + .result + .as_ref() + .ok() + .map(|execution| execution_details(execution, txn.execution.balances.take())); + let logs = details.as_ref().map(|d| Arc::clone(&d.logs)).unwrap_or_default(); + let event = Event::Execution(Execution { header, details }); + + ExecutionCommit { signature, status, event, logs } +} + +/// Projects SVM execution data into the retained ledger format. +fn execution_details( + execution: &ExecutedTransaction, + balances: Option, +) -> ExecutionDetails { + let (pre, post) = balances.map(|bc| bc.into_vecs()).unwrap_or_default(); + let details = &execution.execution_details; + + ExecutionDetails { + fee: execution.loaded_transaction.fee_details.total_fee(), + balances: Balances { pre, post }, + logs: details.log_messages.clone().unwrap_or_default(), + compute_units: details.executed_units, + return_data: details.return_data.as_ref().map(|rd| ReturnData { + program: rd.program_id.to_bytes(), + data: rd.data.clone().into(), + }), + cpi: details.inner_instructions.as_ref().map(cpis), + } +} + +/// Projects grouped SVM inner instructions into ledger CPI records. +fn cpis(groups: &InnerInstructionsList) -> Vec { + groups + .iter() + .map(|group| Cpis(group.iter().map(instruction).collect())) + .collect() +} + +/// Projects one SVM inner instruction into the ledger instruction format. +fn instruction(ix: &InnerInstruction) -> Instruction { + Instruction { + stack_height: ix.stack_height, + compiled: CompiledInstruction { + program_index: ix.instruction.program_id_index, + accounts: ix.instruction.accounts.clone(), + data: ix.instruction.data.clone(), + }, + } +} diff --git a/nucleus/src/notifier.rs b/nucleus/src/notifier.rs new file mode 100644 index 0000000..0166871 --- /dev/null +++ b/nucleus/src/notifier.rs @@ -0,0 +1,43 @@ +//! One-shot async event notification. + +use std::sync::atomic::{AtomicBool, Ordering::*}; + +use tokio::sync::Notify; + +/// A one-shot latch that wakes every waiter once notified. +/// +/// Waiters that arrive after notification return immediately. The event cannot +/// be reset. +#[derive(Default)] +pub struct EventNotifier { + /// Set after notification so future waiters can return immediately. + done: AtomicBool, + /// Wakes tasks that registered before notification. + notify: Notify, +} + +impl EventNotifier { + /// Marks the event complete and wakes all current waiters. + pub fn notify(&self) { + self.done.store(true, Release); + self.notify.notify_waiters(); + } + + /// Waits until the event is complete. + pub async fn notified(&self) { + loop { + if self.done.load(Acquire) { + return; + } + + let notified = self.notify.notified(); + // The waiter is created before the second load, so a concurrent + // notify cannot land between observing `false` and registering. + if self.done.load(Acquire) { + return; + } + + notified.await; + } + } +}