diff --git a/Cargo.toml b/Cargo.toml index 227e07a9..33e98531 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,10 +1,12 @@ [workspace] members = [ "accountsdb", + "keeper", "ledger", "nucleus", "programs/magic-root-interface", "programs/v42-calculator-interface", + "programs/v42-calculator-program", "solana/account", "solana/program-runtime", "solana/svm", @@ -24,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" } @@ -64,6 +67,7 @@ scc = "3.8.4" serde = "1.0.228" serde_bytes = "0.11.19" serde_with = { version = "3.21.0", default-features = false } +smallvec = "1.15.2" snedfile = "0.1" tar = "0.4.45" tempfile = "3" @@ -138,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 00000000..76a8a348 --- /dev/null +++ b/keeper/Cargo.toml @@ -0,0 +1,66 @@ +[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 = ["accountsdb/testkit", "dep:solana-instruction", "ledger/testkit", "nucleus/testkit"] + +[dependencies] +accountsdb = { workspace = true } +ledger = { workspace = true } +nucleus = { workspace = true, features = ["config", "notifier", "runtime"] } + +ahash = { workspace = true } +arc-swap = { workspace = true } +derive_more = { workspace = true, features = ["from"] } +flume = { workspace = true } +oneshot = { workspace = true } +parking_lot = { workspace = true } +scc = { workspace = true } +serde = { workspace = true } +smallvec = { 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"] } +solana-account = { workspace = true, features = ["bincode"] } +solana-feature-gate-interface = { workspace = true, features = ["bincode"] } +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] +keeper = { workspace = true, features = ["testkit"] } +nucleus = { workspace = true, features = ["testkit"] } + +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 00000000..074a0957 --- /dev/null +++ b/keeper/README.md @@ -0,0 +1,79 @@ +# `magicblock-keeper` + +Keeper opens accountsdb and the ledger as one durable state boundary. It also +owns startup account seeding, read-side caches, and live subscription fanout. +Account routing remains in accountsdb and ledger retention remains in ledger. + +## Startup and recovery + +`KeeperBuilder::build` opens both stores and seeds active feature accounts, +native builtins, configured loader-v4 programs, caller-provided accounts, +authority funding, and sysvars. + +Accountsdb is restored from the newest retained snapshot when validation finds +corruption, its sealed superblock trails the ledger head, or its committed +transaction count trails the ledger's durable count. The accountsdb count is a +checkpoint high-water mark and may exceed the locally retained ledger count, +including after replication snapshot bootstrap. The original active tree is +saved until the restored snapshot validates. Engine replay is responsible for +advancing restored state from the successor of its sealed superblock through the +ledger tip and must finish with matching transaction counts when replay runs. + +`nucleus::config::BlockstoreParams` supplies the expected block time and +non-zero superblock interval used by pacing and cache TTL calculation. The +shared accountsdb, blockstore, and ledger parameters are defined by nucleus; +keeper consumes them when opening its durable stores and caches. + +## Authority + +`nucleus::config::Authority::local` is the keypair used for locally signed +messages. When `Authority::remote` is set, `Keeper::authority` returns that +immediate upstream identity instead of the local pubkey, while `Keeper::signer` +continues to return the local signer. Replication followers retain both values +across restart. + +The effective authority also identifies the engine's sponsor account. Keeper +creates this engine-local account only for an empty ledger, persists its spent +balance across restarts, and restores its initial balance on reset. Startup +rejects a non-empty deployment whose configured authority account is absent. + +## Superblock finalization + +`Keeper::finalize_superblock` snapshots accountsdb at the current ledger head, +computes the persisted-account checksum, appends the corresponding +`SuperblockSeal`, and archives the snapshot in the successor superblock +directory. + +Finalization requires exclusive account-store access. Engine obtains that +exclusivity through the sequencer and simulator barriers before calling it. + +## Synchronization + +`Keeper::sync(false)` flushes queued appends and accountsdb while keeping ledger +workers available, as required by replay and replication. `Keeper::sync(true)` +is the irreversible shutdown fence: it closes every reader after earlier queued +requests, flushes and closes the appender, then flushes accountsdb. + +## Caches and subscriptions + +Signature and recent-block caches use slot-based TTLs with lazy eviction on +insertion. The account cache is an LRU that also coordinates concurrent loads of +missing accounts. Only non-authoritative modes enter the eviction LRU; +delegated, ephemeral, and unresolved transient state remains outside it. + +Dedicated channels publish account and program updates, signature results, logs, +processed transactions, blocks, cache evictions, completed snapshots, and +service messages. Signatures have terminal oneshot fanout; persistent multicast +streams give each receiver a bounded queue and disconnect a receiver that falls +behind. Processed transactions, service messages, and cache evictions each have +one process-lifetime receiver and apply producer backpressure when full. +Append rejection notifies only its newest signature waiter, preserving older +waiters for an already accepted transaction; invalid-blockhash status is cached. + +## `testkit` + +The `testkit` feature exposes a keeper backed by throwaway directories plus v42 +account and transaction helpers, including persisted-metadata fault injection. +When enabled, Keeper's build script builds the v42 SBF artifact consumed by the +harness. Downstream tests enable the feature on their dev-dependency instead of +duplicating the setup. diff --git a/keeper/build.rs b/keeper/build.rs new file mode 100644 index 00000000..d285f3d9 --- /dev/null +++ b/keeper/build.rs @@ -0,0 +1,67 @@ +//! 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> { + // The embedded program is only compiled by `keeper::testkit`. + if env::var_os("CARGO_FEATURE_TESTKIT").is_none() { + return Ok(()); + } + + 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 00000000..5d729b96 --- /dev/null +++ b/keeper/src/accessor.rs @@ -0,0 +1,372 @@ +//! Namespaced keeper access APIs. + +use std::{ops::Deref, path::PathBuf, sync::Arc}; + +use accountsdb::{AccountEntry, AccountLoader, AccountsDB, AccountsDBError}; +use ledger::{ + LedgerRequestError, + request::*, + schema::{Event, SuperblockSeal, TransactionEntry}, +}; +use nucleus::{ + Slot, + ledger::{Block, BlockstorePosition}, + tls::{EncodedMessage, TlsManager}, +}; +use solana_account::{AccountSharedData, ReadableAccount}; +use solana_hash::Hash; +use solana_pubkey::Pubkey; +use solana_sdk_ids::sysvar; +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 solana_transaction_error::TransactionError; +use tokio::sync::mpsc::Receiver; + +use crate::{ + FullTransaction, Keeper, ResolvedTransaction, + cache::{AccountCache, MissingAccount}, + error::Result, + subscriptions::TransactionLogs, + util::{execution_commit, request}, +}; + +/// Account operations namespace. +pub struct AccountsAccessor<'a> { + pub(crate) 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> { + /// 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).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).await + } + + /// Subscribes as the sole receiver of account pubkeys evicted from the recent-load cache. + /// + /// Returns an error if the process-lifetime eviction receiver was already registered. + pub fn subscribe_evictions(&self) -> Result> { + self.keeper.caches.accounts.evictions.subscribe() + } + + /// Subscribes to completed accountsdb snapshot archives. + pub fn subscribe_snapshots(&self) -> Receiver { + self.keeper.subscriptions.snapshots.subscribe_sync(()) + } + + /// 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> { + pub(crate) keeper: &'a Keeper, +} + +impl<'a> TransactionsAccessor<'a> { + /// 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 as the sole receiver of all processed transactions, including failures. + /// + /// Returns an error if the process-lifetime transaction receiver was already registered. + pub fn subscribe_processed(&self) -> Result> { + self.keeper.subscriptions.transactions.subscribe() + } + + /// Subscribes to status updates for one transaction signature. + pub async fn subscribe_signature( + &self, + signature: Signature, + ) -> oneshot::Receiver { + self.keeper.subscriptions.signatures.subscribe(signature).await + } + + /// Subscribes to log batches mentioning `account`. + pub async fn subscribe_logs(&self, account: Pubkey) -> Receiver> { + self.keeper.subscriptions.logs.subscribe(account).await + } + + /// Subscribes as the sole receiver of encoded service messages. + /// + /// Returns an error if the process-lifetime service receiver was already registered. + pub fn subscribe_service_messages(&self) -> Result> { + self.keeper.subscriptions.services.subscribe() + } + + /// Appends transaction bytes to the ledger, deduplicating by signature. + /// + /// Returns `Ok(true)` when the transaction was appended. On `Ok(false)`, + /// the latest signature subscriber receives `AlreadyProcessed` or + /// `BlockhashNotFound`. 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]; + let mut result = Ok(()); + if !caches.signatures.push(signature, None, slot) { + result = Err(TransactionError::AlreadyProcessed); + } else if !self.keeper.blocks().is_valid(transaction.recent_blockhash()) { + result = Err(TransactionError::BlockhashNotFound); + let status = TransactionStatus { result: result.clone(), slot }; + caches.signatures.update(&signature, Some(status)); + } + if result.is_err() { + let status = TransactionStatus { result, slot }; + self.keeper.subscriptions.signatures.send_last(&signature, &status); + return Ok(false); + } + 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 subs = &self.keeper.subscriptions; + let commit = execution_commit(&mut txn); + self.keeper.ledger.appender.send(commit.event)?; + + if let Some(execution) = self.commit_state_transitions(&txn.execution.result)? { + let accounts = &execution.loaded_transaction.accounts; + let mut logs = None; + for (pubkey, acc) in accounts { + if subs.logs.contains(pubkey) { + let logs = logs.get_or_insert_with(|| { + Arc::new(TransactionLogs { + signature: commit.signature, + result: commit.status.result.clone(), + logs: Arc::clone(&commit.logs), + }) + }); + subs.logs.send(pubkey, logs); + } + if !acc.dirty() { + continue; + } + subs.accounts.send(pubkey, acc); + if subs.programs.contains(acc.owner()) { + let account = &(*pubkey, acc.clone()); + subs.programs.send(acc.owner(), account); + } + } + while let Some(msg) = TlsManager::dequeue() { + subs.services.blocking_send(msg); + } + } + subs.transactions.blocking_send(txn); + // Clear TLS unconditionally so unsent messages cannot leak into the next transaction. + TlsManager::clear(); + subs.signatures.send(&commit.signature, &commit.status); + self.keeper.caches.signatures.update(&commit.signature, Some(commit.status)); + Ok(()) + } + + /// Commits one accepted transaction to accountsdb, writing dirty accounts + /// only for successful execution and returning it for downstream fanout. + pub fn commit_state_transitions<'t>( + &self, + result: &'t TransactionProcessingResult, + ) -> Result> { + let execution = result.as_ref().ok().filter(|e| e.was_successful()); + let accounts = execution + .into_iter() + .flat_map(|execution| execution.loaded_transaction.accounts.iter()) + .filter(|(id, a)| a.dirty() && !sysvar::instructions::check_id(id)); + self.keeper.accountsdb.commit(accounts)?; + Ok(execution.map(|execution| &**execution)) + } +} + +/// Block operations namespace. +pub struct BlocksAccessor<'a> { + pub(crate) keeper: &'a Keeper, +} + +impl<'a> BlocksAccessor<'a> { + /// 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_sync(()) + } + + /// Publishes a completed block and advances block-derived account state. + /// + /// Replay skips the ledger append because the block is already stored. + pub fn append(&self, block: Block, replay: bool) -> Result<()> { + let event = Event::Block(block); + if !replay { + self.keeper.ledger.appender.send(event)?; + self.keeper.subscriptions.blocks.send(&(), &block); + } + self.keeper.caches.blocks.push(block); + self.keeper.accounts().update_sysvars(block)?; + self.keeper.accounts().set_slot(block.slot)?; + Ok(()) + } +} + +/// Superblock operations namespace +pub struct SuperblockAccessor<'a> { + pub(crate) keeper: &'a Keeper, +} + +impl SuperblockAccessor<'_> { + /// Id of the superblock accountsdb last sealed. + pub fn sealed(&self) -> SuperblockSeal { + SuperblockSeal { + id: self.keeper.accountsdb.superblock(), + checksum: self.keeper.accountsdb.checksum(), + transactions: self.keeper.accountsdb.transactions(), + } + } + + /// Returns the ledger root containing retained superblock directories. + pub fn directory(&self) -> &PathBuf { + &self.keeper.ledger.directory + } + + /// Follower's current durable blockstore position, reported to the leader at handshake. + pub fn position(&self) -> BlockstorePosition { + self.keeper.ledger.position() + } + + /// Enqueues a seal onto the append stream, sealing the current superblock and + /// rotating to the next. Used by a follower applying a seal received from the leader. + pub fn append(&self, seal: SuperblockSeal) -> Result<()> { + let event = Event::Superblock(seal); + self.keeper.ledger.appender.send(event)?; + self.sync(false) + } + + /// Installs a snapshot seal and adopts its cumulative transaction count. + pub fn bootstrap(&self, seal: SuperblockSeal) -> Result<()> { + let event = Event::Bootstrap(seal); + self.keeper.ledger.appender.send(event)?; + self.sync(false) + } + + /// Blocks until every queued append event has been flushed and made durable. + pub fn sync(&self, is_final: bool) -> Result<()> { + let (response, ack) = oneshot::channel(); + let event = Event::Sync { response, is_final }; + self.keeper.ledger.appender.send(event)?; + ack.recv().map_err(LedgerRequestError::from)?.map_err(Into::into) + } +} diff --git a/keeper/src/builder.rs b/keeper/src/builder.rs new file mode 100644 index 00000000..d50f48c8 --- /dev/null +++ b/keeper/src/builder.rs @@ -0,0 +1,325 @@ +//! Keeper construction, recovery, and startup account seeding. + +use std::{ + collections::HashMap, + fs::{self, File}, + 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, + config::{AccountsDBParams, Authority, BlockstoreParams, LedgerParams}, + ledger::{ACCOUNTSDB_SNAPSHOT_FILE, Block}, + shutdown::ShutdownManager, +}; +use serde::Serialize; +use solana_account::{AccountBuilder, AccountMode, AccountSharedData, ReadableAccount}; +use solana_feature_gate_interface::Feature; +use solana_program_runtime::invoke_context::BuiltinFunctionWithContext; +use solana_pubkey::Pubkey; +use solana_sdk_ids::sysvar; +#[allow(deprecated)] +use solana_sysvar::fees::Fees; +use solana_sysvar::{ + clock::Clock, + epoch_rewards::EpochRewards, + epoch_schedule::EpochSchedule, + last_restart_slot::LastRestartSlot, + rent::Rent, + slot_hashes::{SlotHashes, SysvarId}, +}; +use tracing::{error, info, warn}; + +use crate::{ + Keeper, + cache::{AccountCache, BlocksCache, Caches, ExpiringCache}, + error::Result, + metrics, + subscriptions::Subscriptions, +}; + +/// Initial balance assigned to the authority account that sponsors account creation. +pub(crate) const SPONSOR_INIT_BALANCE: u64 = u64::MAX / 2; +/// Maximum number of recent hashes retained by the `SlotHashes` sysvar. +const SLOTHASH_ENTRIES: usize = 512; +/// Wall-clock retention window for recently processed signatures. +const SIGNATURE_CACHE_WINDOW: Duration = Duration::from_secs(75); +/// Wall-clock retention window for recently produced blocks. +const BLOCK_CACHE_WINDOW: Duration = Duration::from_secs(60); + +/// Builder for keeper directories and cache timing. +#[derive(Clone)] +pub struct KeeperBuilder { + /// Local signer and optional remote authority represented by the engine. + pub authority: Authority, + /// 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, +} + +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), + }) + } + + /// 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 authority = self.authority.pubkey(); + if accountsdb.loader().load(&authority)?.is_none() { + let sponsor = AccountBuilder::default() + .lamports(SPONSOR_INIT_BALANCE) + .mode(AccountMode::Ephemeral); + accounts.push((authority, sponsor.build())); + } + accounts.extend(self.accounts.drain()); + accountsdb.store(&accounts)?; + Ok((block, featureset)) + } + + /// 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::new(self.accountsdb.lru_capacity)); + + 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::syscall_parameter_address_restrictions::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 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)?; + // `SlotHashes` is ordered newest-first, so the latest block is `first` + if let Some(&(slot, hash)) = hashes.first() { + let parent = &slot.saturating_sub(1); + let parent = hashes.get(parent).copied().unwrap_or_default(); + let time = self.blocktime(ledger, slot).await?; + last_block.replace(Block { slot, hash, time, parent }); + } + } else { + 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())); + #[allow(deprecated)] + accounts.push(( + Fees::id(), + self.account(&Fees::default(), &sysvar::ID)?.build(), + )); + accounts.push(( + sysvar::last_restart_slot::id(), + self.account(&LastRestartSlot::default(), &sysvar::ID)?.build(), + )); + accounts.push(( + sysvar::instructions::id(), + self.account(&(), &sysvar::ID)?.build(), + )); + accounts.push(( + EpochSchedule::id(), + self.account(&EpochSchedule::default(), &sysvar::ID)?.build(), + )); + accounts.push(( + EpochRewards::id(), + self.account(&EpochRewards::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. + /// + /// A restored store trails the ledger tip — snapshots are archived at sealed + /// superblocks, not at the tip — so the returned accountsdb is only + /// guaranteed to validate, not to be current. Catching it back up is the + /// caller's job. + fn accountsdb(&self, ledger: &LedgerHandle) -> Result { + let mut backup = None; + loop { + let mut accountsdb = AccountsDB::new(&self.accountsdb.directory)?; + // Seal N opens ledger head N+1, so accountsdb is current at head-1. + let expected = ledger.head().saturating_sub(1); + let restored = backup.is_some(); + let lagging = accountsdb.superblock() < expected; + let count_lagging = accountsdb.transactions() < ledger.transactions(); + match accountsdb.validate() { + Ok(()) if restored || (!lagging && !count_lagging) => { + let reclaimed = accountsdb.compact()?; + info!( + lagging, + count_lagging, reclaimed, "accountsdb validation succeeded" + ); + backup.map(fs::remove_dir_all).transpose()?; + return Ok(accountsdb); + } + validation @ (Err(AccountsDBError::Corruption) | Ok(())) => { + if restored { + error!(?validation, "restored accountsdb is corrupt"); + accountsdb.backup(BackupOp::Restore)?; + return Err(SnapshotError::Missing.into()); + } + warn!( + ?validation, + lagging, count_lagging, "state inconsistency detected" + ); + backup.replace(accountsdb.backup(BackupOp::Save)?); + if let Err(error) = self.unarchive(ledger) { + accountsdb.backup(BackupOp::Restore)?; + return Err(error); + } + } + Err(other) => return Err(other.into()), + } + } + } + + /// Restores the first retained accountsdb snapshot found, from newest to oldest. + fn unarchive(&self, ledger: &LedgerHandle) -> Result<()> { + info!("restoring accountsdb from latest available snapshot"); + for superblock in ledger.iter() { + let src = superblock.directory.join(ACCOUNTSDB_SNAPSHOT_FILE); + if !src.exists() { + continue; + } + let dst = AccountsDB::directory(&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"); + return Ok(()); + } + Err(SnapshotError::Missing.into()) + } +} diff --git a/keeper/src/cache.rs b/keeper/src/cache.rs new file mode 100644 index 00000000..13526cac --- /dev/null +++ b/keeper/src/cache.rs @@ -0,0 +1,250 @@ +//! Read-side caches owned by keeper. + +use std::{collections::VecDeque, hash::Hash, sync::Arc}; + +use ahash::RandomState; +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_account::AccountMode; +use solana_hash::Hash as SolanaHash; +use solana_pubkey::Pubkey; +use solana_signature::Signature; + +use crate::{ + metrics, + subscriptions::{Subscription, Unicast}, +}; + +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, RandomState>, + /// Pubkeys evicted when a committed load displaces a cold account. + pub(crate) evictions: Unicast, +} + +impl AccountCache { + /// Creates missing-load coordination with the requested recent-load capacity. + pub(crate) fn new(capacity: usize) -> Self { + let lru = HashCache::with_capacity_and_hasher(256, capacity, Default::default()); + Self { + lru, + reservations: Default::default(), + evictions: Unicast::new(32, Subscription::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 { + /// Completes the load, wakes waiters, and tracks non-authoritative modes for + /// eviction. + pub async fn complete(mut self, mode: AccountMode) { + let Some((cache, notifier)) = self.release() else { + return; + }; + notifier.notify(true); + if mode.authoritative() { + return; + } + if let Ok(Some(evicted)) = cache.lru.put_sync(self.pubkey, ()) { + metrics::account_cache_eviction(); + cache.evictions.send(evicted.0).await; + } else { + metrics::account_cache_insert() + } + } + + 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(false); + } +} + +impl AccountWait { + /// Waits for the active loader and returns the pubkey plus whether it committed. + pub async fn wait(self) -> (Pubkey, bool) { + let result = self.1.notified().await; + (self.0, result) + } +} + +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); + 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 + } + } + } + + /// Entry count, including expired entries no sweep has reached yet, so the + /// metrics gauges fed from this can read above the live entry count. + pub(crate) fn len(&self) -> usize { + self.index.len() + } + + /// May yield an entry already past its expiry slot but not yet swept. + pub(crate) fn get(&self, key: &K) -> Option { + self.index.read_sync(key, |_, v| v.clone()) + } + + /// May report an entry already past its expiry slot but not yet swept. + pub(crate) fn contains(&self, key: &K) -> bool { + self.index.contains_sync(key) + } + + /// Replaces the value for `key`, leaving its expiry slot untouched: an + /// update does not extend the entry's life. No-op if `key` is absent. + 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 00000000..d8c67c59 --- /dev/null +++ b/keeper/src/error.rs @@ -0,0 +1,48 @@ +//! 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), + /// A process-lifetime unicast stream was already registered. + #[error("subscription already registered: {0}")] + SubscriptionRegistered(&'static str), +} + +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 00000000..522073ce --- /dev/null +++ b/keeper/src/lib.rs @@ -0,0 +1,245 @@ +#![doc = include_str!("../README.md")] + +use std::{ + fs::{self, File}, + path::PathBuf, + sync::Arc, + thread, +}; + +use agave_feature_set::FeatureSet; +use solana_account::AccountBuilder; +use solana_hash::Hash; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +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::{ + Slot, + config::Authority, + ledger::{ACCOUNTSDB_SNAPSHOT_FILE, SuperblockSeal}, +}; +use solana_sysvar::rent::Rent; + +use crate::{ + accessor::{AccountsAccessor, BlocksAccessor, SuperblockAccessor, TransactionsAccessor}, + builder::SPONSOR_INIT_BALANCE, + cache::Caches, + error::Result, + metrics::Operation, + subscriptions::Subscriptions, +}; + +pub use cache::{AccountLoad, AccountWait, MissingAccount}; +/// Re-exported so callers can name what `Keeper::transactions().status()` returns. +pub use ledger::request::TransactionStatus; +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 { + /// Local signer and optional remote authority represented by this engine. + authority: Authority, + /// 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 { keeper: self } + } + + /// Returns the transaction operations namespace. + pub fn transactions(&self) -> TransactionsAccessor<'_> { + TransactionsAccessor { keeper: self } + } + + /// Returns the block operations namespace. + pub fn blocks(&self) -> BlocksAccessor<'_> { + BlocksAccessor { keeper: self } + } + + /// Returns the superblock operations namespace. + pub fn superblocks(&self) -> SuperblockAccessor<'_> { + SuperblockAccessor { keeper: self } + } + + /// Returns the configured remote authority, or the local identity when unset. + pub fn authority(&self) -> Pubkey { + self.authority.pubkey() + } + + /// Returns the local signer, which may differ from [`Self::authority`]. + pub fn signer(&self) -> &Keypair { + &self.authority.local + } + + /// Returns the latest block hash + pub fn blockhash(&self) -> Hash { + self.blocks().latest().hash + } + + /// Borrows the handle used for direct ledger reads, appends, and + /// durable-position subscriptions. + pub fn ledger(&self) -> &LedgerHandle { + &self.ledger + } + + /// Streams retained ledger entries after accountsdb's sealed superblock up to + /// the ledger tip, used to rebuild state after snapshot restoration. Returns + /// `None` when accountsdb is already current by slot and transaction count. + pub async fn replay(&self) -> Result> { + let ledger_slot = self.ledger.tip().unwrap_or_default(); + let accountsdb_slot = self.accountsdb.slot(); + let ledger_txns = self.ledger.transactions(); + let accountsdb_txns = self.accountsdb.transactions(); + if accountsdb_slot >= ledger_slot && accountsdb_txns >= ledger_txns { + return Ok(None); + }; + let (tx, rx) = mpsc::channel(16); + let params = ReplayParams { + tx, + superblock: self.accountsdb.superblock(), + }; + 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, accountsdb_txns, ledger_txns, "starting 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 head = self.ledger.head(); + let next = 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(head) }?; + let checksum = self.accountsdb.checksum(); + let transactions = self.accountsdb.transactions(); + let seal = SuperblockSeal { id: head, checksum, transactions }; + self.superblocks().append(seal)?; + let dir = Superblock::init_dir(&self.ledger.directory, next)?; + self.archive(snapshot, dir)?; + info!(head, "finalized superblock"); + Ok(()) + } + + /// Resolved once from the seeded feature accounts at startup and fixed for + /// the engine's lifetime — features never activate mid-run. + pub fn features(&self) -> &FeatureSet { + &self.featureset + } + + /// Supplied by the builder at startup and fixed for the engine's lifetime; + /// the same parameters that sized the seeded accounts. + pub fn rent(&self) -> &Rent { + &self.rent + } + + /// Waits for queued ledger work to become durable, then synchronously + /// flushes persisted account storage. Volatile accounts are not serialized. + /// + /// A final sync closes every ledger reader and the appender. It is + /// irreversible and must only be used during coordinated shutdown. + pub fn sync(&self, is_final: bool) -> Result<()> { + if is_final { + for _ in 0..self.ledger.reader.receiver_count() { + self.ledger.reader.send(ReadRequest::Shutdown)?; + } + } + self.superblocks().sync(is_final)?; + self.accountsdb.flush(true).map_err(Into::into) + } + + /// Appends a reset marker before discarding chain-synchronized volatile + /// accounts and restoring the authority sponsor's initial balance. + /// + /// Internal system accounts and persisted engine-authoritative state remain + /// available. + pub fn reset(&self, slot: Slot) -> Result<()> { + self.ledger.appender.send(Event::Reset(slot))?; + self.accountsdb.reset(); + let authority = self.authority(); + let account = self.accounts().loader().load(&authority)?; + if let Some(account) = account { + let acc = AccountBuilder::from(account).lamports(SPONSOR_INIT_BALANCE); + self.accounts().store(&[(authority, acc.build())])?; + } + info!(slot, "reset volatile state"); + Ok(()) + } + + /// 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_SNAPSHOT_FILE); + let tmp = target.join(format!("{ACCOUNTSDB_SNAPSHOT_FILE}.tmp")); + let dst = File::options().write(true).create(true).truncate(true).open(&tmp)?; + let subscriptions = self.subscriptions.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)?; + { + let archive = tar.into_inner()?.finish()?; + metrics::snapshot_size(archive.metadata()?.len()); + archive.sync_data()?; + } + // Rename only after sync so replication cannot serve a partial archive. + fs::rename(tmp, &path)?; + fs::remove_dir_all(snapshot)?; + subscriptions.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 00000000..aa9654be --- /dev/null +++ b/keeper/src/metrics.rs @@ -0,0 +1,152 @@ +//! Prometheus metrics for keeper. + +use std::sync::OnceLock; + +use nucleus::metrics::{self as metric, OperationTimer}; +use nucleus::metrics::{ + IntCounter, IntCounterVec, IntGauge, MetricOperation, MetricSpec, OperationCounters, +}; + +use crate::subscriptions::Subscription; + +/// 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.", +}; +/// Block hash cache entries. +const BLOCK_HASH_CACHE_ENTRIES: MetricSpec = MetricSpec { + name: "keeper_block_hash_cache_entries", + help: "Current block hash cache entries.", +}; +/// Most recently completed snapshot archive size. +const SNAPSHOT_SIZE: MetricSpec = MetricSpec { + name: "keeper_snapshot_size", + help: "Most recently completed snapshot archive size in bytes.", +}; +/// 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.", +}; +/// Slow multicast consumer disconnection counter. +const SLOW_CONSUMER_DISCONNECTS: MetricSpec = MetricSpec { + name: "keeper_subscription_slow_consumer_disconnects", + help: "Multicast receivers disconnected because their queue was full.", +}; + +/// 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 in the default Prometheus registry. +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)) +} + +/// Records an account cache insertion that increases current occupancy. +pub(crate) fn account_cache_insert() { + metric::with_metrics(&METRICS, |m| m.account_cache_entries.inc()); +} + +/// 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(metric::gauge_value(count)) + }); +} + +/// Records the most recently completed snapshot archive size in bytes. +pub(crate) fn snapshot_size(bytes: u64) { + metric::with_metrics(&METRICS, |m| { + m.snapshot_size.set(metric::gauge_value(bytes)) + }); +} + +/// Records one account cache eviction. +pub(crate) fn account_cache_eviction() { + metric::with_metrics(&METRICS, |m| m.account_cache_evictions.inc()); +} + +/// Records one account resolution race condition. +pub(crate) fn account_resolution_race() { + metric::with_metrics(&METRICS, |m| m.account_resolution_race.inc()); +} + +/// Records one receiver disconnected because its queue was full. +pub(crate) fn slow_consumer_disconnect(subscription: Subscription) { + metric::with_metrics(&METRICS, |m| { + m.slow_consumer_disconnects.with_label_values(&[subscription.label()]).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, + /// Block hash cache entry gauge. + block_hash_cache_entries: IntGauge, + /// Most recently completed snapshot archive size in bytes. + snapshot_size: IntGauge, + /// Account cache eviction counter. + account_cache_evictions: IntCounter, + /// Account resolution race conditions counter. + account_resolution_race: IntCounter, + /// Slow consumer disconnections labeled by subscription stream. + slow_consumer_disconnects: IntCounterVec, +} +impl Default for Metrics { + /// Builds collectors and registers them in the default Prometheus registry. + fn default() -> Self { + Self { + operations: OperationCounters::new(OPERATION_TIME), + account_cache_entries: metric::gauge(ACCOUNT_CACHE_ENTRIES, 0), + block_hash_cache_entries: metric::gauge(BLOCK_HASH_CACHE_ENTRIES, 0), + snapshot_size: metric::gauge(SNAPSHOT_SIZE, 0), + account_cache_evictions: metric::counter(ACCOUNT_CACHE_EVICTIONS, 0), + account_resolution_race: metric::counter(ACCOUNT_RESOLUTION_RACES, 0), + slow_consumer_disconnects: metric::counter_vec( + SLOW_CONSUMER_DISCONNECTS, + &["subscription"], + ), + } + } +} diff --git a/keeper/src/subscriptions.rs b/keeper/src/subscriptions.rs new file mode 100644 index 00000000..1fe11373 --- /dev/null +++ b/keeper/src/subscriptions.rs @@ -0,0 +1,317 @@ +//! Live read-side notification channels. + +use std::{ + hash::Hash, + path::PathBuf, + sync::{Arc, OnceLock}, + time::Duration, +}; + +use accountsdb::AccountEntry; +use ahash::RandomState; +use ledger::{request::TransactionStatus, schema::Block}; +use nucleus::{ + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + tls::EncodedMessage, +}; +use scc::HashMap; +use smallvec::SmallVec; +use solana_account::AccountSharedData; +use solana_pubkey::Pubkey; +use solana_signature::Signature; +use solana_transaction_error::TransactionResult; +use tokio::{ + sync::mpsc::{self, error::TrySendError}, + time::{MissedTickBehavior, interval}, +}; + +use crate::{ + FullTransaction, + error::{KeeperError, Result}, + metrics::{self, Operation}, +}; + +type MpscSenders = SmallVec<[mpsc::Sender; 1]>; +type OneshotSenders = SmallVec<[oneshot::Sender; 1]>; + +/// Stable metric identity for a subscription stream. +#[derive(Clone, Copy)] +pub(crate) enum Subscription { + Accounts, + Programs, + Logs, + Blocks, + Transactions, + Snapshots, + Services, + Evictions, +} + +impl Subscription { + pub(crate) const fn label(self) -> &'static str { + match self { + Self::Accounts => "accounts", + Self::Programs => "programs", + Self::Logs => "logs", + Self::Blocks => "blocks", + Self::Transactions => "transactions", + Self::Snapshots => "snapshots", + Self::Services => "services", + Self::Evictions => "evictions", + } + } +} + +/// Live notification channels owned by keeper. +pub(crate) struct Subscriptions { + /// Account updates keyed by account pubkey. + pub(crate) accounts: Multicast, + /// Program account updates keyed by owner pubkey. + pub(crate) programs: Multicast, + /// Signature status updates keyed by transaction signature. + pub(crate) signatures: MulticastOneshot, + /// Log notifications keyed by mentioned program or account pubkey. + pub(crate) logs: Multicast>, + /// Newly committed blocks. + pub(crate) blocks: Multicast<(), Block>, + /// All committed transactions for the sole stream consumer. + pub(crate) transactions: Unicast, + /// Accountsdb snapshot archive completions. + pub(crate) snapshots: Multicast<(), PathBuf>, + /// Encoded service messages for the sole stream consumer. + pub(crate) services: Unicast, +} + +impl Subscriptions { + /// Builds subscription channels and starts cleanup for idle keyed entries. + pub(crate) fn new(shutdown: &mut ShutdownManager) -> Arc { + let subscriptions = Arc::new(Self { + accounts: Multicast::new(8, Subscription::Accounts), + programs: Multicast::new(16, Subscription::Programs), + signatures: Default::default(), + logs: Multicast::new(8, Subscription::Logs), + blocks: Multicast::new(32, Subscription::Blocks), + transactions: Unicast::new(1024, Subscription::Transactions), + snapshots: Multicast::new(4, Subscription::Snapshots), + services: Unicast::new(64, Subscription::Services), + }); + let shutdown = shutdown.handle(Service::SubscriptionsCleanup); + tokio::spawn(cleanup(subscriptions.clone(), shutdown)); + subscriptions + } + + async fn cleanup(&self) { + self.accounts.cleanup().await; + self.programs.cleanup().await; + self.signatures.cleanup().await; + self.logs.cleanup().await; + self.blocks.cleanup().await; + self.snapshots.cleanup().await; + } +} + +/// Composite log notification sent to log subscribers. +#[derive(Clone)] +pub struct TransactionLogs { + /// First transaction signature. + pub signature: Signature, + /// Runtime transaction result (carries the error on failure). + pub result: TransactionResult<()>, + /// Log lines emitted during execution. + pub logs: Arc>, +} + +/// One process-lifetime bounded receiver. +pub(crate) struct Unicast { + sender: OnceLock>, + capacity: usize, + subscription: Subscription, +} + +impl Unicast { + pub(crate) const fn new(capacity: usize, subscription: Subscription) -> Self { + Self { + sender: OnceLock::new(), + capacity, + subscription, + } + } + + /// Creates the process-lifetime receiver, rejecting every later subscriber. + pub(crate) fn subscribe(&self) -> Result> { + let (tx, rx) = mpsc::channel(self.capacity); + self.sender + .set(tx) + .map_err(|_| KeeperError::SubscriptionRegistered(self.subscription.label()))?; + Ok(rx) + } + + /// Sends asynchronously, waiting until the receiver has capacity. + pub(crate) async fn send(&self, value: V) { + let Some(sender) = self.sender.get() else { + return; + }; + let _ = sender.send(value).await; + } + + /// Sends from a synchronous worker, waiting until the receiver has capacity. + pub(crate) fn blocking_send(&self, value: V) { + let Some(sender) = self.sender.get() else { + return; + }; + let _ = sender.blocking_send(value); + } +} + +/// Persistent per-key fanout over one bounded queue per receiver. +pub(crate) struct Multicast { + senders: HashMap, RandomState>, + capacity: usize, + subscription: Subscription, +} + +impl Multicast +where + K: Eq + Hash, +{ + pub(crate) fn new(capacity: usize, subscription: Subscription) -> Self { + Self { + senders: Default::default(), + capacity, + subscription, + } + } + + /// Adds a receiver for `key` with its own bounded queue. + pub(crate) async fn subscribe(&self, key: K) -> mpsc::Receiver { + let (tx, rx) = mpsc::channel(self.capacity); + self.senders.entry_async(key).await.or_default().push(tx); + rx + } + + /// Adds a receiver synchronously when the public accessor cannot await. + pub(crate) fn subscribe_sync(&self, key: K) -> mpsc::Receiver { + let (tx, rx) = mpsc::channel(self.capacity); + self.senders.entry_sync(key).or_default().push(tx); + rx + } + + /// Returns whether `key` has any live receivers. + pub(crate) fn contains(&self, key: &K) -> bool { + let mut contains = false; + self.senders.remove_if_sync(key, |senders| { + senders.retain(|sender| !sender.is_closed()); + contains = !senders.is_empty(); + !contains + }); + contains + } + + /// Drops closed receivers and keys that no longer have receivers. + async fn cleanup(&self) { + self.senders + .retain_async(|_, senders| { + senders.retain(|sender| !sender.is_closed()); + !senders.is_empty() + }) + .await; + } +} + +impl Multicast +where + K: Eq + Hash, + V: Clone, +{ + /// Fans out without blocking, disconnecting receivers whose queues are full. + pub(crate) fn send(&self, key: &K, value: &V) { + self.senders.remove_if_sync(key, |senders| { + senders.retain(|sender| match sender.try_send(value.clone()) { + Ok(()) => true, + Err(TrySendError::Full(_)) => { + metrics::slow_consumer_disconnect(self.subscription); + false + } + Err(TrySendError::Closed(_)) => false, + }); + senders.is_empty() + }); + } +} + +/// Terminal per-key fanout over one oneshot channel per receiver. +pub(crate) struct MulticastOneshot(HashMap, RandomState>); + +impl Default for MulticastOneshot { + fn default() -> Self { + Self(Default::default()) + } +} + +impl MulticastOneshot +where + K: Eq + Hash, +{ + /// Adds a receiver for the terminal value associated with `key`. + pub(crate) async fn subscribe(&self, key: K) -> oneshot::Receiver { + let (tx, rx) = oneshot::channel(); + self.0.entry_async(key).await.or_default().push(tx); + rx + } + + /// Drops closed receivers and keys that no longer have receivers. + async fn cleanup(&self) { + self.0 + // Keep closed positions while any receiver is live so `send_last` + // cannot mistake an older subscription for the newest one. + .retain_async(|_, senders| senders.iter().any(|sender| !sender.is_closed())) + .await; + } +} + +impl MulticastOneshot +where + K: Eq + Hash, + V: Clone, +{ + /// Sends the terminal value only to the most recently added receiver. + pub(crate) fn send_last(&self, key: &K, value: &V) { + let Some(mut senders) = self.0.get_sync(key) else { + return; + }; + let sender = senders.pop(); + if senders.is_empty() { + let _ = senders.remove_entry(); + } + if let Some(sender) = sender { + let _ = sender.send(value.clone()); + } + } + + /// Removes `key` and sends its terminal value to every current receiver. + pub(crate) fn send(&self, key: &K, value: &V) { + let Some((_, senders)) = self.0.remove_sync(key) else { + return; + }; + for sender in senders { + let _ = sender.send(value.clone()); + } + } +} + +/// Drops abandoned multicast senders after their receivers are 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.cleanup().await; + } + } + } + shutdown.terminate(ShutdownReason::Signalled); +} diff --git a/keeper/src/testkit.rs b/keeper/src/testkit.rs new file mode 100644 index 00000000..0595ed25 --- /dev/null +++ b/keeper/src/testkit.rs @@ -0,0 +1,273 @@ +//! 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, clippy::unwrap_used)] + +use std::{ + collections::HashMap, + fs::{self, File}, + num::NonZeroU64, + path::{Path, PathBuf}, + sync::Arc, + time::Duration, +}; + +use accountsdb::{AccountsDB, STORAGE_FILE}; +use derive_more::Deref; +use nucleus::{ + config::{AccountsDBParams, BlockstoreParams, LedgerParams}, + ledger::ACCOUNTSDB_SNAPSHOT_FILE, + shutdown::ShutdownManager, + testkit::signed_view as compose_view, +}; +use solana_account::{AccountBuilder, AccountMode, 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, WireVersion, block, init_tracing, patterned_bytes, + sign_versioned_instructions, tempdir, transaction, v42_padded_value, v42_sum, +}; +use tokio::time; + +use crate::{Keeper, ResolvedTransaction, TransactionView, builder::KeeperBuilder}; + +/// The v42 calculator ELF, built and located by `keeper/build.rs`. +pub const V42_PROGRAM_ELF: &[u8] = include_bytes!(env!("V42_CALCULATOR_PROGRAM_SO")); +/// Slots sealed into each superblock by the standard test engine. +pub const SUPERBLOCK: NonZeroU64 = NonZeroU64::new(4).unwrap(); + +/// 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 100 ms blocktime. +/// +/// `builtins` and `accounts` default to empty and `programs` holds only v42; +/// individual tests fill the rest as needed. [`TestKeeper::new`] is the seeded +/// path, adding a funded payer on top. +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(), + lru_capacity: 256, + }, + ledger: LedgerParams { + directory: dirs.ledger.path().to_owned(), + size_limit: u64::MAX, + }, + blockstore: BlockstoreParams { + blocktime: Duration::from_millis(100), + superblock: 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 { + /// Directories backing this keeper, returned by [`Self::close`] so a test can + /// reopen over the same on-disk state. + pub dirs: Dirs, + /// Lifecycle manager owning the keeper's background services; exposed so + /// tests can register their own services against the same shutdown. + pub 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 { + Self::with(Dirs::default()).await + } + + /// [`Self::new`] over `dirs`, which may already hold state from an earlier + /// keeper closed over the same directories. + pub async fn with(dirs: Dirs) -> Self { + let mut builder = keeper_builder(&dirs); + let payer = Keypair::new(); + builder.accounts.insert( + payer.pubkey(), + AccountBuilder::default().lamports(1_000_000).build(), + ); + Self::from_builder(dirs, builder).await + } + + /// Builds a keeper from a caller-configured builder. + /// + /// `dirs` must own the directories referenced by `builder` and outlive the + /// resulting keeper. Unlike [`Self::with`], nothing is seeded beyond what the + /// builder already carries. + pub async fn from_builder(dirs: Dirs, builder: KeeperBuilder) -> Self { + let mut shutdown = ShutdownManager::default(); + let keeper = Arc::new(builder.build(&mut shutdown).await.unwrap()); + Self { dirs, shutdown, keeper } + } + + /// Flushes durable state, stops every background service, and returns the + /// directories for reopen. + /// + /// The flush republishes a valid accountsdb checksum, so a test that wants a + /// corrupt store must call [`corrupt`] on the returned directories *after* + /// this, never before. + pub async fn close(self) -> Dirs { + let Self { mut shutdown, keeper, dirs } = self; + keeper.sync(true).unwrap(); + shutdown.terminate().await; + dirs + } +} + +/// 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).unwrap(); + let resolved = + ResolvedTransaction::try_new(view, Some(Default::default()), &Default::default()).unwrap(); + (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, view) = compose_view(&payer, [ix], Hash::default()); + ResolvedTransaction::try_new(view, Some(Default::default()), &Default::default()).unwrap() +} + +/// Configures a v42 account carrying an 8-byte little-endian `i64` and twice +/// its rent-exempt minimum, leaving one reserve available for transfer tests. +pub fn v42_builder(value: i64, mode: AccountMode) -> AccountBuilder { + AccountBuilder::default() + .lamports(Rent::default().minimum_balance(8) * 2) + .owner(V42_ID) + .mode(mode) + .data(value.to_le_bytes().to_vec()) +} + +/// Stores a funded v42 `i64` account in `mode` and returns its pubkey. +pub fn store_v42(keeper: &Keeper, value: i64, mode: AccountMode) -> Pubkey { + let key = Pubkey::new_unique(); + keeper.accounts().store(&[(key, v42_builder(value, mode).build())]).unwrap(); + key +} + +/// Reads the little-endian `i64` payload of a v42 account, or `None` if absent. +pub fn load_v42_data(keeper: &Keeper, key: Pubkey) -> Option { + keeper.accounts().loader().read(&key, decode_v42).unwrap() +} + +/// Reads the lamport balance of a stored v42 account, or `None` if absent. +pub fn load_v42_lamports(keeper: &Keeper, key: Pubkey) -> Option { + keeper.accounts().loader().read(&key, ReadableAccount::lamports).unwrap() +} + +/// Signs `instruction` into the sanitized transaction view consumed by services. +pub fn signed_view( + keeper: &Keeper, + payer: Option<&Keypair>, + instruction: Instruction, +) -> (Signature, TransactionView) { + let payer = payer.unwrap_or(keeper.signer()); + compose_view(payer, [instruction], keeper.blockhash()) +} + +/// Returns the archived accountsdb snapshot path under any retained superblock, +/// or `None` when no superblock directory holds one yet. +pub fn archived_snapshot(keeper: &Keeper) -> Option { + fs::read_dir(&keeper.ledger.directory) + .unwrap() + .filter_map(Result::ok) + .map(|e| e.path().join(ACCOUNTSDB_SNAPSHOT_FILE)) + .find(|p| p.exists()) +} + +/// Waits until a subscribed detached snapshot archiver reports completion. +pub async fn await_archive(keeper: &Keeper) -> PathBuf { + let mut rx = keeper.accounts().subscribe_snapshots(); + time::timeout(Duration::from_secs(8), rx.recv()) + .await + .expect("snapshot archives in time") + .unwrap() +} + +/// Overwrites one `u64` metadata word in the closed persisted store. +/// +/// `DatabaseMeta` starts with version at offset 0, checksum at 8, slot at 16, +/// superblock at 24, and committed transaction count at 32. +/// +/// This must be the last write to the store: `flush(true)` recomputes and +/// republishes the checksum, so this must run *after* the keeper is closed, never +/// against a live one. +pub fn corrupt(root: &Path, offset: u64, value: u64) { + use std::io::{Seek, SeekFrom, Write}; + let path = AccountsDB::directory(root).join(STORAGE_FILE); + let mut file = File::options().write(true).open(&path).unwrap(); + file.seek(SeekFrom::Start(offset)).unwrap(); + file.write_all(&value.to_ne_bytes()).unwrap(); + file.flush().unwrap(); +} + +/// Decodes the little-endian `i64` payload stored in a v42 account. +pub fn decode_v42(account: &impl ReadableAccount) -> i64 { + i64::from_le_bytes(account.data()[..8].try_into().expect("v42 account holds an i64")) +} diff --git a/keeper/src/tests/caches.rs b/keeper/src/tests/caches.rs new file mode 100644 index 00000000..99fd3557 --- /dev/null +++ b/keeper/src/tests/caches.rs @@ -0,0 +1,120 @@ +//! Read-side cache primitives keeper owns: the slot-based `ExpiringCache` and the +//! `AccountCache` missing-load coordination. + +use std::sync::Arc; + +use solana_account::{AccountBuilder, AccountMode}; +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() { + use AccountMode::*; + let modes = [ReadOnly, Placeholder, Delegated, Ephemeral, Transient, System]; + for mode in modes { + for commit in [true, false] { + let cache = Arc::new(AccountCache::new(256)); + 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.complete(mode).await; + } else { + drop(load) + } + + assert_eq!( + waiter.await.unwrap(), + (pk, commit), + "waiter returns the load outcome" + ); + let tracked = matches!(mode, ReadOnly | Placeholder | System); + assert_eq!(cache.lru.get_sync(&pk).is_some(), tracked && 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 00000000..11a3dfae --- /dev/null +++ b/keeper/src/tests/mod.rs @@ -0,0 +1,11 @@ +//! 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 crate::testkit::{TestKeeper, signed_tx}; diff --git a/keeper/src/tests/recovery.rs b/keeper/src/tests/recovery.rs new file mode 100644 index 00000000..ee18a988 --- /dev/null +++ b/keeper/src/tests/recovery.rs @@ -0,0 +1,132 @@ +//! Startup seeding, corruption recovery + +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 super::TestKeeper; +use crate::testkit::{Dirs, archived_snapshot, await_archive, corrupt, keeper_builder}; + +// 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 keeper = TestKeeper::from_builder(dirs, builder).await; + let rent = Rent::default(); + let accounts = keeper.accounts(); + let loader = accounts.loader(); + + // 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::syscall_parameter_address_restrictions::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 = loader.load(&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 = loader.load(&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 = loader + .load(&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 = loader.load(&id).unwrap().expect("sysvar seeded"); + assert_eq!(acc.owner(), &sysvar::ID); + } + drop(loader); + keeper.close().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. +// +// The marker takes a distinct value in each state the reopen could land on, so +// the assertion separates all three: 1 is the older snapshot, 2 the newest, and +// 3 lives only in persisted state (stored after the last archive, so no snapshot +// holds it). Recovery must yield 2 — reading 3 back would mean the corruption +// went undetected and nothing was restored at all. +#[tokio::test] +async fn recovers_the_newest_snapshot() { + let marker = Pubkey::new_unique(); + let dirs = Dirs::default(); + let builder = keeper_builder(&dirs); + let keeper = TestKeeper::from_builder(dirs, builder.clone()).await; + + // First snapshot captures marker == 1. + store_marker(&keeper, marker, 1); + keeper.finalize_superblock().expect("first finalize"); + await_archive(&keeper).await; + assert!( + archived_snapshot(&keeper).is_some(), + "snapshot archived under superblock" + ); + // Second snapshot, in a later superblock, captures marker == 2. + store_marker(&keeper, marker, 2); + keeper.finalize_superblock().expect("second finalize"); + await_archive(&keeper).await; + // Past every archive: this value is what an un-restored store would keep. + store_marker(&keeper, marker, 3); + let dirs = keeper.close().await; + + // Corruption must follow the close, whose flush would otherwise republish a + // valid checksum over the poisoned word. + corrupt(dirs.accounts.path(), 8, 0xABAB_ABAB_ABAB_ABAB); + + let keeper = TestKeeper::from_builder(dirs, builder).await; + keeper.accounts().validate().expect("restored store validates"); + let restored = keeper.accounts().loader().load(&marker).unwrap().expect("marker restored"); + assert_eq!(restored.lamports(), 2, "newest snapshot wins"); + // The corrupt tree saved for inspection is removed on successful recovery. + assert!(!keeper.dirs.accounts.path().join("CURRENT.bkp").exists()); + + keeper.close().await; +} + +/// Stores the recovery marker account at `lamports`, the value each snapshot +/// captures and recovery must bring back. +fn store_marker(keeper: &TestKeeper, marker: Pubkey, lamports: u64) { + let account = AccountBuilder::default().lamports(lamports).mode(AccountMode::Delegated); + keeper.accounts().store(&[(marker, account.build())]).unwrap(); +} diff --git a/keeper/src/tests/subscriptions.rs b/keeper/src/tests/subscriptions.rs new file mode 100644 index 00000000..ee063695 --- /dev/null +++ b/keeper/src/tests/subscriptions.rs @@ -0,0 +1,121 @@ +//! Subscription fanout primitives, transaction-append dedup + +use std::sync::Arc; + +use super::{TestKeeper, signed_tx}; +use crate::{ + ResolvedTransaction, + subscriptions::{Multicast, MulticastOneshot, Subscription, Unicast}, +}; +use nucleus::testkit::{V42_ID, signed_view}; +use solana_hash::Hash; +use solana_instruction::Instruction; +use solana_keypair::Keypair; +use solana_transaction_error::TransactionError; + +/// Proves unicast exclusivity, persistent fanout, terminal fanout, and slow-receiver removal. +#[tokio::test] +async fn subscribers_send_semantics() { + let unicast = Arc::new(Unicast::new(1, Subscription::Transactions)); + let mut unicast_rx = unicast.subscribe().unwrap(); + assert!(unicast.subscribe().is_err()); + unicast.send(1).await; + let sender = unicast.clone(); + let send = tokio::spawn(async move { sender.send(2).await }); + tokio::task::yield_now().await; + assert!(!send.is_finished(), "async unicast send waits for capacity"); + assert_eq!(unicast_rx.recv().await, Some(1)); + send.await.unwrap(); + assert_eq!(unicast_rx.recv().await, Some(2)); + + unicast.send(3).await; + let sender = unicast.clone(); + let send = std::thread::spawn(move || sender.blocking_send(4)); + assert_eq!(unicast_rx.recv().await, Some(3)); + send.join().unwrap(); + assert_eq!(unicast_rx.recv().await, Some(4)); + drop(unicast_rx); + assert!(unicast.subscribe().is_err()); + + let multicast = Multicast::new(1, Subscription::Accounts); + multicast.send(&1, &9); + let mut first = multicast.subscribe(1).await; + let mut second = multicast.subscribe(1).await; + multicast.send(&1, &10); + assert_eq!(first.recv().await, Some(10)); + assert_eq!(second.recv().await, Some(10)); + multicast.send(&1, &11); + multicast.send(&1, &12); + assert_eq!(first.recv().await, Some(11)); + assert_eq!(first.recv().await, None); + assert_eq!(second.recv().await, Some(11)); + assert_eq!(second.recv().await, None); + + let oneshot = MulticastOneshot::default(); + let first = oneshot.subscribe(1).await; + let closed = oneshot.subscribe(1).await; + drop(closed); + oneshot.send_last(&1, &18); + assert!(matches!( + first.try_recv(), + Err(oneshot::TryRecvError::Empty) + )); + let second = oneshot.subscribe(1).await; + oneshot.send_last(&1, &19); + assert_eq!(second.await.unwrap(), 19); + oneshot.send(&1, &20); + assert_eq!(first.await.unwrap(), 20); + let third = oneshot.subscribe(1).await; + oneshot.send(&1, &21); + assert_eq!(third.await.unwrap(), 21); +} + +// Appending reserves the signature while rejection wakes only its own latest +// waiter. Invalid blockhash is retained as a terminal cached status. +#[tokio::test] +async fn append_dedup_and_status_sentinel() { + let keeper = TestKeeper::new().await; + let (signature, txn) = signed_tx(); + let slot = keeper.blocks().current_slot(); + let original = keeper.transactions().subscribe_signature(signature).await; + + // First append writes to the ledger; the duplicate is dropped. + assert!( + keeper.transactions().append(&txn).await.unwrap(), + "first append is accepted" + ); + let duplicate = keeper.transactions().subscribe_signature(signature).await; + assert!( + !keeper.transactions().append(&txn).await.unwrap(), + "duplicate is deduplicated" + ); + let status = duplicate.await.unwrap(); + assert_eq!(status.result, Err(TransactionError::AlreadyProcessed)); + assert_eq!(status.slot, slot); + assert!(matches!( + original.try_recv(), + Err(oneshot::TryRecvError::Empty) + )); + + // The sentinel makes status() return None from the cache. + assert!(keeper.transactions().status(signature).await.unwrap().is_none()); + + let payer = Keypair::new(); + let (signature, view) = signed_view( + &payer, + [Instruction::new_with_bytes(V42_ID, &[], vec![])], + Hash::new_from_array([1; 32]), + ); + let txn = + ResolvedTransaction::try_new(view, Some(Default::default()), &Default::default()).unwrap(); + let rejected = keeper.transactions().subscribe_signature(signature).await; + assert!(!keeper.transactions().append(&txn).await.unwrap()); + let status = rejected.await.unwrap(); + assert_eq!(status.result, Err(TransactionError::BlockhashNotFound)); + assert_eq!(status.slot, slot); + let cached = keeper.transactions().status(signature).await.unwrap().unwrap(); + assert_eq!(cached.result, Err(TransactionError::BlockhashNotFound)); + assert_eq!(cached.slot, slot); + + keeper.close().await; +} diff --git a/keeper/src/util.rs b/keeper/src/util.rs new file mode 100644 index 00000000..d73906ec --- /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/programs/v42-calculator-program/Cargo.toml b/programs/v42-calculator-program/Cargo.toml new file mode 100644 index 00000000..3f979414 --- /dev/null +++ b/programs/v42-calculator-program/Cargo.toml @@ -0,0 +1,32 @@ +[package] +name = "v42-calculator-program" + +authors.workspace = true +edition = "2021" +homepage.workspace = true +license.workspace = true +repository.workspace = true +version.workspace = true + +[dependencies] +v42-calculator-interface = { workspace = true } + +solana-account-info = { workspace = true } +solana-cpi = { workspace = true } +solana-instruction = { workspace = true, features = ["syscalls"] } +solana-msg = { workspace = true } +solana-program-entrypoint = { workspace = true } +solana-program-error = { workspace = true } +solana-pubkey = { workspace = true } +solana-sysvar = { workspace = true, features = ["bytemuck"] } + +# Compiled to BPF with `cargo build-sbf`; the `rlib` lets it also build on the +# host so the workspace's `cargo build`/`clippy` stay green. +[lib] +crate-type = ["cdylib", "rlib"] + +[lints.rust] +# The `entrypoint!` macro emits `target_os = "solana"` and `custom-heap`/ +# `custom-panic` feature gates that only exist under the SBF toolchain; silence +# the noise they produce on host builds. +unexpected_cfgs = { level = "allow" } diff --git a/programs/v42-calculator-program/README.md b/programs/v42-calculator-program/README.md new file mode 100644 index 00000000..2b3ff24c --- /dev/null +++ b/programs/v42-calculator-program/README.md @@ -0,0 +1,36 @@ +# `v42-calculator-program` + +This SBF test program evaluates the postfix wire format from +`v42-calculator-interface`. It exercises account-data reads, the `Clock` sysvar, +checked arithmetic, recursive CPI, return data, writable account output, and +direct lamport transfers. + +The crate uses Rust 2021 and upstream minimal program crates so it remains +compatible with `cargo build-sbf`. Keeper's build script produces +`target/deploy/v42_calculator_program.so` for its testkit. + +## Execution model + +The crate entrypoint dispatches to the calculator and transfer domains. The +calculator owns expression evaluation and value encoding, transfer owns signed +balance and value movement, and error owns the stable calculator error codes. + +The evaluator processes literals, account `i64` values, clock timestamps, +arithmetic opcodes, and length-prefixed nested calls on a fixed-capacity stack. +Malformed input, stack errors, arithmetic errors, missing accounts, and short +buffers return stable `ProgramError::Custom` codes from `CalcError`. + +A top-level invocation writes the final little-endian `i64` to account zero. A +nested invocation publishes the value through return data after all child calls +finish, because a subsequent CPI would clear earlier return data. + +The program emits `v42:` trace messages for entry, CPI, clock access, results, +and failures. Result messages distinguish `account0` from `return_data` routing. + +A `TRANSFER` instruction instead reads an exact little-endian `i64` delta from +its data and applies it to two distinct writable accounts owned by this program. +Account 0 receives the negated delta and account 1 receives the delta, updating +both lamports and the little-endian `i64` calculator value in their first eight +data bytes. Both updates use wrapping two's-complement arithmetic, so a negative +delta reverses the transfer direction. Neither account needs to sign. Transfer +does not run the evaluator or publish return data. diff --git a/programs/v42-calculator-program/src/calculator.rs b/programs/v42-calculator-program/src/calculator.rs new file mode 100644 index 00000000..972dfc25 --- /dev/null +++ b/programs/v42-calculator-program/src/calculator.rs @@ -0,0 +1,197 @@ +//! Postfix calculator evaluation and result routing. + +use solana_account_info::AccountInfo; +use solana_cpi::{get_return_data, invoke, set_return_data}; +use solana_instruction::{AccountMeta, Instruction, TRANSACTION_LEVEL_STACK_HEIGHT}; +use solana_msg::msg; +use solana_program_error::{ProgramError, ProgramResult}; +use solana_sysvar::clock::Clock; +use solana_sysvar::Sysvar; +use v42_calculator_interface::{opcodes::*, ID}; + +use crate::error::CalcError; + +/// Evaluates the RPN program in `data` and routes the result: a **top-level** +/// invocation writes it (LE `i64`) into the output account; a **CPI** +/// invocation returns it through return data. The two are told apart by the +/// stack height. +/// +/// A nested call must publish its result *after* its own child `CALL`s complete, +/// because every CPI clears the return-data register — which is exactly what +/// happens here, since the result is emitted only once evaluation is done. +pub(crate) fn process(accounts: &[AccountInfo], data: &[u8], height: usize) -> ProgramResult { + let result = eval(accounts, data)?; + if height == TRANSACTION_LEVEL_STACK_HEIGHT { + msg!("v42: result={} -> account0", result); + let output = accounts.first().ok_or(CalcError::MissingOutput)?; + write_account_value(&mut output.try_borrow_mut_data()?, result) + } else { + msg!("v42: result={} -> return_data", result); + set_return_data(&result.to_le_bytes()); + Ok(()) + } +} + +/// Runs the token stream against an operand stack and returns the single value +/// left at the end. +fn eval(accounts: &[AccountInfo], data: &[u8]) -> Result { + let mut cursor = Cursor(data); + let mut stack = Stack::new(); + while !cursor.is_empty() { + match cursor.u8()? { + PUSH_LIT => stack.push(cursor.i64()?)?, + PUSH_ACC => stack.push(account_i64(accounts, cursor.u8()?)?)?, + PUSH_CLOCK => stack.push(clock_ts()?)?, + op @ (ADD | SUB | MUL | DIV) => { + let b = stack.pop()?; + let a = stack.pop()?; + stack.push(arithmetic(op, a, b)?)?; + } + CALL => { + let len = cursor.u16()? as usize; + let nested = cursor.take(len)?; + stack.push(call(accounts, nested)?)?; + } + _ => return Err(CalcError::BadOpcode.into()), + } + } + stack.into_result().map_err(Into::into) +} + +fn arithmetic(op: u8, a: i64, b: i64) -> Result { + match op { + ADD => a.checked_add(b).ok_or(CalcError::Arithmetic), + SUB => a.checked_sub(b).ok_or(CalcError::Arithmetic), + MUL => a.checked_mul(b).ok_or(CalcError::Arithmetic), + DIV if b == 0 => Err(CalcError::DivByZero), + DIV => a.checked_div(b).ok_or(CalcError::Arithmetic), + _ => Err(CalcError::BadOpcode), + } +} + +/// Reads the calculator value from the first eight account data bytes. +pub(crate) fn read_account_value(bytes: &[u8]) -> Result { + read_head_i64(bytes, CalcError::ShortAccount).map_err(Into::into) +} + +/// Writes the calculator value to the first eight account data bytes. +pub(crate) fn write_account_value(bytes: &mut [u8], value: i64) -> ProgramResult { + bytes + .get_mut(..8) + .ok_or(CalcError::ShortAccount)? + .copy_from_slice(&value.to_le_bytes()); + Ok(()) +} + +/// Decodes the `i64` LE stored in the first 8 bytes of `bytes`, reporting `short` +/// if there aren't that many. +fn read_head_i64(bytes: &[u8], short: CalcError) -> Result { + let head = bytes.get(..8).ok_or(short)?; + Ok(i64::from_le_bytes(head.try_into().expect("length checked"))) +} + +/// Reads the `i64` LE operand from the first 8 data bytes of an instruction +/// account (borrowed, not copied). +fn account_i64(accounts: &[AccountInfo], index: u8) -> Result { + let account = accounts.get(index as usize).ok_or(CalcError::BadAccountIndex)?; + read_account_value(&account.try_borrow_data()?) +} + +fn clock_ts() -> Result { + let ts = Clock::get()?.unix_timestamp; + msg!("v42: clock.ts={}", ts); + Ok(ts) +} + +/// Evaluates `nested` by invoking this same program, forwarding every account +/// unchanged so `PUSH_ACC` indices are identical at every recursion depth, and +/// returns the callee's return-data `i64`. +fn call(accounts: &[AccountInfo], nested: &[u8]) -> Result { + let metas = accounts + .iter() + .map(|a| AccountMeta { + pubkey: *a.key, + is_signer: a.is_signer, + is_writable: a.is_writable, + }) + .collect(); + let instruction = Instruction { + program_id: ID, + accounts: metas, + data: nested.to_vec(), + }; + msg!("v42: cpi len={}", nested.len()); + invoke(&instruction, accounts)?; + + let (_, bytes) = get_return_data().ok_or(CalcError::MissingReturnData)?; + Ok(read_head_i64(&bytes, CalcError::ShortReturnData)?) +} + +/// A forward cursor over the instruction byte stream. Every read is bounds- +/// checked and advances the cursor; operands are read in place, never copied. +struct Cursor<'a>(&'a [u8]); + +impl<'a> Cursor<'a> { + fn is_empty(&self) -> bool { + self.0.is_empty() + } + + fn u8(&mut self) -> Result { + let (&byte, rest) = self.0.split_first().ok_or(CalcError::Truncated)?; + self.0 = rest; + Ok(byte) + } + + fn u16(&mut self) -> Result { + Ok(u16::from_le_bytes(self.array()?)) + } + + fn i64(&mut self) -> Result { + Ok(i64::from_le_bytes(self.array()?)) + } + + fn array(&mut self) -> Result<[u8; N], CalcError> { + Ok(self.take(N)?.try_into().expect("length checked")) + } + + fn take(&mut self, len: usize) -> Result<&'a [u8], CalcError> { + if self.0.len() < len { + return Err(CalcError::Truncated); + } + let (head, rest) = self.0.split_at(len); + self.0 = rest; + Ok(head) + } +} + +/// Fixed-capacity operand stack — deep enough for any expression a test would +/// build, and allocation-free. +struct Stack { + slots: [i64; 64], + len: usize, +} + +impl Stack { + fn new() -> Self { + Self { slots: [0; 64], len: 0 } + } + + fn push(&mut self, value: i64) -> Result<(), CalcError> { + *self.slots.get_mut(self.len).ok_or(CalcError::StackOverflow)? = value; + self.len += 1; + Ok(()) + } + + fn pop(&mut self) -> Result { + self.len = self.len.checked_sub(1).ok_or(CalcError::StackUnderflow)?; + Ok(self.slots[self.len]) + } + + /// The single value a well-formed program leaves behind. + fn into_result(self) -> Result { + match self.len { + 1 => Ok(self.slots[0]), + _ => Err(CalcError::UnbalancedProgram), + } + } +} diff --git a/programs/v42-calculator-program/src/error.rs b/programs/v42-calculator-program/src/error.rs new file mode 100644 index 00000000..7faf6e53 --- /dev/null +++ b/programs/v42-calculator-program/src/error.rs @@ -0,0 +1,29 @@ +//! Stable calculator errors exposed through `ProgramError::Custom`. + +use solana_msg::msg; +use solana_program_error::ProgramError; + +/// Failure modes with stable discriminants for callers and tests. +#[derive(Debug)] +#[repr(u32)] +pub(crate) enum CalcError { + Truncated = 1, + BadOpcode, + StackOverflow, + StackUnderflow, + UnbalancedProgram, + Arithmetic, + DivByZero, + BadAccountIndex, + ShortAccount, + MissingOutput, + MissingReturnData = 12, + ShortReturnData, +} + +impl From for ProgramError { + fn from(error: CalcError) -> Self { + msg!("v42: err {:?}", error); + ProgramError::Custom(error as u32) + } +} diff --git a/programs/v42-calculator-program/src/lib.rs b/programs/v42-calculator-program/src/lib.rs new file mode 100644 index 00000000..38b6501d --- /dev/null +++ b/programs/v42-calculator-program/src/lib.rs @@ -0,0 +1,25 @@ +#![doc = include_str!("../README.md")] + +mod calculator; +mod error; +mod transfer; + +use solana_account_info::AccountInfo; +use solana_instruction::syscalls::get_stack_height; +use solana_msg::msg; +use solana_program_entrypoint::entrypoint; +use solana_program_error::ProgramResult; +use solana_pubkey::Pubkey; +use v42_calculator_interface::TRANSFER; + +entrypoint!(process_instruction); + +fn process_instruction(_: &Pubkey, accounts: &[AccountInfo], data: &[u8]) -> ProgramResult { + let height = get_stack_height(); + msg!("v42: enter height={} len={}", height, data.len()); + if data.first() == Some(&TRANSFER) { + transfer::process(accounts, data) + } else { + calculator::process(accounts, data, height) + } +} diff --git a/programs/v42-calculator-program/src/transfer.rs b/programs/v42-calculator-program/src/transfer.rs new file mode 100644 index 00000000..7bd18a03 --- /dev/null +++ b/programs/v42-calculator-program/src/transfer.rs @@ -0,0 +1,41 @@ +//! Signed value transfers between calculator accounts. + +use solana_account_info::AccountInfo; +use solana_msg::msg; +use solana_program_error::{ProgramError, ProgramResult}; + +use crate::calculator::{read_account_value, write_account_value}; + +/// Applies the encoded signed delta to the lamports and calculator values of +/// two distinct accounts. Both accounts must be owned by this program and +/// writable. +pub(crate) fn process(accounts: &[AccountInfo], data: &[u8]) -> ProgramResult { + let [from, to, ..] = accounts else { + return Err(ProgramError::NotEnoughAccountKeys); + }; + if from.key == to.key { + return Err(ProgramError::InvalidArgument); + } + let delta = data + .get(1..) + .map(TryInto::<[u8; 8]>::try_into) + .ok_or(ProgramError::InvalidInstructionData)? + .map(i64::from_le_bytes) + .map_err(|_| ProgramError::InvalidInstructionData)?; + + msg!("v42: transfer delta={}", delta); + apply_delta(from, delta.wrapping_neg())?; + apply_delta(to, delta) +} + +/// Adds the same two's-complement delta to an account's balance and stored +/// calculator value. +fn apply_delta(account: &AccountInfo, delta: i64) -> ProgramResult { + let mut lamports = account.try_borrow_mut_lamports()?; + **lamports = (**lamports).wrapping_add(delta as u64); + drop(lamports); + + let mut data = account.try_borrow_mut_data()?; + let result = read_account_value(&data)?.wrapping_add(delta); + write_account_value(&mut data, result) +}