diff --git a/Cargo.toml b/Cargo.toml index 84197b2..4981135 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -4,6 +4,7 @@ members = [ "keeper", "ledger", "nucleus", + "processor", "programs/magic-root-interface", "programs/magic-root-program", "programs/v42-calculator-interface", @@ -31,6 +32,9 @@ ledger = { path = "ledger", package = "magicblock-ledger" } magic-root-interface = { path = "programs/magic-root-interface" } magic-root-program = { path = "programs/magic-root-program" } nucleus = { path = "nucleus", package = "magicblock-engine-nucleus" } +processor = { path = "processor", package = "magicblock-processor" } +v42-calculator-interface = { path = "programs/v42-calculator-interface", default-features = false } + solana-account = { path = "solana/account" } solana-program-runtime = { path = "solana/program-runtime" } solana-svm = { path = "solana/svm" } diff --git a/processor/Cargo.toml b/processor/Cargo.toml new file mode 100644 index 0000000..5925b8a --- /dev/null +++ b/processor/Cargo.toml @@ -0,0 +1,56 @@ +[package] +name = "magicblock-processor" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "processor" + +[dependencies] +accountsdb = { workspace = true } +keeper = { workspace = true } +nucleus = { workspace = true, features = ["ledger-schema", "metrics", "runtime", "shutdown", "tls"] } + +ahash = { workspace = true } +blake3 = { workspace = true } +derive_more = { workspace = true, features = ["from"] } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] } +tracing = { workspace = true } + +agave-feature-set = { workspace = true } +agave-precompiles = { workspace = true, features = ["agave-unstable-api"] } +agave-syscalls = { workspace = true, features = ["agave-unstable-api"] } +agave-transaction-view = { workspace = true } +solana-account = { workspace = true } +solana-compute-budget-instruction = { workspace = true, features = ["agave-unstable-api"] } +solana-hash = { workspace = true } +solana-precompile-error = { workspace = true } +solana-program-runtime = { workspace = true } +solana-pubkey = { workspace = true } +solana-signature = { workspace = true } +solana-svm = { workspace = true } +solana-svm-transaction = { workspace = true } +solana-sysvar = { workspace = true } +solana-transaction-error = { workspace = true } + +[dev-dependencies] +keeper = { workspace = true, features = ["testkit"] } +v42-calculator-interface = { workspace = true, features = ["builder"] } + +solana-instruction = { workspace = true } +solana-keypair = { workspace = true } +solana-sdk-ids = { workspace = true } +solana-signer = { workspace = true } +solana-transaction = { workspace = true, features = ["wincode"] } +oneshot = { workspace = true } +wincode = { workspace = true } + +[lints] +workspace = true diff --git a/processor/README.md b/processor/README.md new file mode 100644 index 0000000..9020301 --- /dev/null +++ b/processor/README.md @@ -0,0 +1,29 @@ +# `magicblock-processor` + +The processor is where transactions actually get run. Its job is to execute as +many of them as possible in parallel while still producing the result they'd have +if run one at a time — and the thing standing in the way of free parallelism is +*account conflicts*. Two transactions that touch the same account (with at least +one writing it) can't safely run at once; two that touch disjoint accounts can. + +So the sequencer's core task is conflict detection. It tracks each `Pubkey` with a +write bit plus a per-executor occupancy bitset (sized to `MAX_EXECUTORS`). +Transactions whose account sets don't collide are fanned out across the pool of +SVM executors and run concurrently; the rest are serialized behind the work +they conflict with. Reads through to accounts go via `keeper`/`accountsdb`, and +writes go back the same way. + +## Quiescence + +Some operations need a consistent, frozen view of state — taking a snapshot at a +superblock seal, or replaying the ledger on restart. In-flight transactions would +make that view inconsistent, so the processor has a **quiescence barrier** that +drains outstanding work and holds new work off until the barrier is released. +That's the hook the keeper's finalize and the engine's replay rely on. + +## Simulation + +The simulation path runs a transaction against current state on *owned copies* of +the accounts and returns the execution record without committing anything. Because +it never touches the real accounts, it can run without disturbing the live +scheduling path. diff --git a/processor/src/callback.rs b/processor/src/callback.rs new file mode 100644 index 0000000..af1be32 --- /dev/null +++ b/processor/src/callback.rs @@ -0,0 +1,54 @@ +use agave_feature_set::FeatureSet; +use nucleus::Slot; +use solana_account::AccountSharedData; +use solana_precompile_error::PrecompileError; +use solana_pubkey::Pubkey; +use solana_svm::transaction_processing_callback::{ + InvokeContextCallback, TransactionProcessingCallback, +}; +use tracing::error; + +use accountsdb::AccountLoader; + +/// Bridges the SVM to keeper-backed account loads. +/// +/// When `LOAD_OWNED` is set, loaded accounts are copied to owned data; this is +/// used by simulation, which must always operate on owned account copies. +pub(crate) struct SVMCallback<'a, const LOAD_OWNED: bool> { + /// Reads accounts from the engine's accounts store. + pub(crate) loader: AccountLoader<'a>, + /// Active feature set governing precompiles and runtime behavior. + pub(crate) featureset: &'a FeatureSet, +} + +impl<'a, const LO: bool> InvokeContextCallback for SVMCallback<'a, LO> { + fn is_precompile(&self, program_id: &Pubkey) -> bool { + agave_precompiles::is_precompile(program_id, |id| self.featureset.is_active(id)) + } + + fn process_precompile( + &self, + program_id: &Pubkey, + data: &[u8], + instruction_datas: Vec<&[u8]>, + ) -> Result<(), PrecompileError> { + agave_precompiles::get_precompile(program_id, |id| self.featureset.is_active(id)) + .ok_or(PrecompileError::InvalidPublicKey) + .and_then(|p| p.verify(data, &instruction_datas, self.featureset)) + } +} + +impl<'a, const LOAD_OWNED: bool> TransactionProcessingCallback for SVMCallback<'a, LOAD_OWNED> { + fn get_account_shared_data(&self, pubkey: &Pubkey) -> Option<(AccountSharedData, Slot)> { + self.loader + .load(pubkey) + .inspect_err(|error| error!(?error, "accountsdb load error")) + .unwrap_or_default() + .map(|mut acc| { + if LOAD_OWNED { + acc = acc.owned().into(); + } + (acc, 0) + }) + } +} diff --git a/processor/src/error.rs b/processor/src/error.rs new file mode 100644 index 0000000..a251fbd --- /dev/null +++ b/processor/src/error.rs @@ -0,0 +1,27 @@ +//! Error type for the transaction processor. + +use std::io; + +use derive_more::From; +use keeper::error::KeeperError; +use nucleus::shutdown::Service; + +/// Convenience result alias for processor operations. +pub type Result = std::result::Result; + +/// Failures raised while scheduling or executing transactions. +#[derive(Debug, thiserror::Error, From)] +pub enum ProcessorError { + /// An I/O error, e.g. while spawning a worker thread or building a runtime. + #[error("i/o error: {0}")] + Io(#[source] io::Error), + /// A failure surfaced by the underlying keeper-backed state. + #[error("state error: {0}")] + State(#[source] KeeperError), + /// A background service is no longer reachable, so work could not be handed off. + #[error("service became unavailable: {0:?}")] + ServiceUnavailable(Service), + /// An internal invariant was violated; carries a human-readable description. + #[error("internal error: {0}")] + Internal(String), +} diff --git a/processor/src/executor.rs b/processor/src/executor.rs new file mode 100644 index 0000000..f0b16cf --- /dev/null +++ b/processor/src/executor.rs @@ -0,0 +1,167 @@ +//! Transaction execution coordination. + +use std::{ + collections::VecDeque, + sync::{ + Arc, + mpsc::{self, Receiver, SyncSender}, + }, + thread::{self, JoinHandle}, +}; + +use ahash::HashMap; +use keeper::{ExecutionRecord, FullTransaction, Keeper}; +use nucleus::{ + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + tls::AUTHORITY, +}; +use solana_program_runtime::loaded_programs::ProgramCache; +use solana_pubkey::Pubkey; +use tokio::sync::mpsc::Sender; +use tracing::{error, info}; + +use crate::{ + ExecutorMessage, ExecutorReady, ResolvedTransaction, Result, callback::SVMCallback, + svm::SvmContext, +}; + +/// Index identifying an executor within the pool. +pub(crate) type ExecutorId = u32; + +/// Worker that drives the SVM for batches of conflict-free transactions. +pub(crate) struct TransactionExecutor { + /// This executor's index within the pool. + id: ExecutorId, + /// Inbound channel of execution batches and block boundaries. + rx: Receiver, + /// Channel used to report back to the sequencer once a batch completes. + ready: Sender, + /// SVM batch processor and per-block environment driving execution. + svm: SvmContext, + /// Durable engine state used for account loads and commits. + state: Arc, + /// Handle used to report cooperative shutdown for this worker. + shutdown: ShutdownHandle, + /// Whether the executor runs in ledger-replay mode: when set, only state + /// transitions are committed directly instead of recording full execution. + replay: bool, +} + +/// Sequencer-side handle to a spawned executor: its dispatch channel, pending +/// batch, held locks, and join handle. +pub(crate) struct ExecutorHandle { + /// Index of the executor this handle controls. + pub(crate) id: ExecutorId, + /// Transactions accumulated for the next dispatch. + pub(crate) batch: Vec, + /// Accounts held on this executor's behalf, each mapped to the number of + /// in-flight transactions referencing it (released when the count hits zero). + pub(crate) locks: HashMap, + /// Transactions queued behind this executor on lock contention. + pub(crate) blocked: VecDeque, + /// Channel for dispatching batches and block boundaries to the worker. + pub(crate) tx: SyncSender, + /// Join handle for the worker thread, taken during shutdown. + pub(crate) task: Option>, +} + +impl TransactionExecutor { + /// Builds the SVM environment, spawns the worker thread, and returns its + /// sequencer-side handle. + pub(crate) fn spawn( + id: ExecutorId, + state: Arc, + cache: Arc, + shutdown: &mut ShutdownManager, + ready: Sender, + replay: bool, + ) -> Result { + let svm = SvmContext::new(&state, cache)?; + let shutdown = shutdown.handle(Service::TransactionExecutor(id)); + let (tx, rx) = mpsc::sync_channel(3); + let executor = Self { + id, + rx, + svm, + ready, + state, + shutdown, + replay, + }; + let task = thread::Builder::new() + .name(format!("transaction-executor-{id}")) + .spawn(move || executor.run())?; + Ok(ExecutorHandle { + id, + task: Some(task), + tx, + batch: Vec::new(), + locks: Default::default(), + blocked: VecDeque::new(), + }) + } + + /// Worker loop: executes batches and applies block transitions until the + /// channel closes or a batch fails, then reports the termination reason. + fn run(mut self) { + // MagicRoot authorizes callers against this thread-local; publish the + // engine authority before executing any transaction on this thread. + AUTHORITY.set(self.state.authority()); + let mut error = None; + while let Ok(msg) = self.rx.recv() { + match msg { + ExecutorMessage::Transactions(mut batch) => { + let result = self.process(&mut batch); + if let Err(e) = result { + error.replace(e); + break; + } + let signal = ExecutorReady { id: self.id, batch }; + if self.ready.blocking_send(signal).is_err() { + info!(id = self.id, "ready channel closed, executor exiting"); + break; + } + } + ExecutorMessage::Block(block) => self.svm.transition(block), + }; + } + let reason = if let Some(error) = error { + error!( + ?error, + id = self.id, + "executor encountered catastrophic failure, terminating" + ); + ShutdownReason::Error(Box::new(error)) + } else { + ShutdownReason::Signalled + }; + self.shutdown.terminate(reason); + } + + /// Loads and executes each transaction in the batch through the SVM, + /// committing either raw state transitions (replay) or full execution. + fn process(&mut self, transactions: &mut Vec) -> Result<()> { + let accessor = self.state.accounts(); + let callback = SVMCallback:: { + loader: accessor.loader(), + featureset: self.state.features(), + }; + for txn in transactions.drain(..) { + let output = self.svm.execute(&callback, &txn, self.state.features()); + if self.replay { + self.state.transactions().commit_state_transitions(&output.processing_result)?; + } else { + let txn = FullTransaction { + transaction: txn.into_view(), + execution: ExecutionRecord { + result: output.processing_result, + balances: output.balance_collector, + slot: self.svm.slot(), + }, + }; + self.state.transactions().commit_execution(txn)?; + } + } + Ok(()) + } +} diff --git a/processor/src/lib.rs b/processor/src/lib.rs new file mode 100644 index 0000000..d367ccc --- /dev/null +++ b/processor/src/lib.rs @@ -0,0 +1,36 @@ +#![doc = include_str!("../README.md")] + +use keeper::ResolvedTransaction; +use nucleus::ledger::Block; + +use crate::executor::ExecutorId; + +mod callback; +mod error; +mod executor; +mod metrics; +pub mod sequencer; +pub mod simulator; +mod svm; + +#[cfg(test)] +mod tests; + +pub use error::{ProcessorError, Result}; +pub use nucleus::runtime::{SequencerMessage, Simulation, SimulatorMessage}; + +/// Batch of work delivered from the sequencer to a transaction executor. +pub enum ExecutorMessage { + /// A set of conflict-free transactions to execute together. + Transactions(Vec), + /// A block boundary advancing the executor's processing environment. + Block(Block), +} + +/// Notification that an executor finished its batch and is idle again. +struct ExecutorReady { + /// Identifier of the executor that became ready. + id: ExecutorId, + /// The drained batch handed back for reuse + batch: Vec, +} diff --git a/processor/src/metrics.rs b/processor/src/metrics.rs new file mode 100644 index 0000000..ad91d0b --- /dev/null +++ b/processor/src/metrics.rs @@ -0,0 +1,115 @@ +//! Prometheus metrics for processor. + +use std::sync::OnceLock; + +use nucleus::metrics::{self as metric, OperationTimer}; +use nucleus::metrics::{IntCounter, IntGauge, MetricOperation, MetricSpec, OperationCounters}; + +/// Process-wide processor metrics registered in the default Prometheus registry. +static METRICS: OnceLock = OnceLock::new(); + +/// Operation latency histogram recorded in microseconds. +const OPERATION_TIME: MetricSpec = MetricSpec { + name: "processor_operation_duration_micros", + help: "Processor operation duration distribution in microseconds.", +}; +/// Executors currently running transaction batches. +const BUSY_EXECUTORS: MetricSpec = MetricSpec { + name: "processor_busy_executors", + help: "Current processor executors running transaction batches.", +}; +/// Transactions queued behind account-lock conflicts. +const BLOCKED_TRANSACTIONS: MetricSpec = MetricSpec { + name: "processor_blocked_transactions", + help: "Current processor transactions queued behind account-lock conflicts.", +}; +/// Account lock conflict counter. +const LOCK_CONFLICTS: MetricSpec = MetricSpec { + name: "processor_lock_conflicts", + help: "Processor account-lock conflicts.", +}; + +/// Processor operation used as a low-cardinality operation label. +#[derive(Clone, Copy)] +pub(crate) enum Operation { + /// Block finalization path. + FinalizeBlock, + /// Quiescence barrier drain path. + BarrierDrain, +} + +impl MetricOperation for Operation { + /// Returns the Prometheus label value for this operation. + fn label(self) -> &'static str { + match self { + Operation::FinalizeBlock => "finalize_block", + Operation::BarrierDrain => "barrier_drain", + } + } +} + +/// Registers processor metrics once. +pub(crate) fn init() { + METRICS.get_or_init(Default::default); +} + +/// Starts an operation timer that records latency when the returned guard drops. +pub(crate) fn time(op: Operation) -> OperationTimer<'static> { + op.time(METRICS.get().map(|m| &m.operations)) +} + +/// Refreshes the busy executor gauge. +pub(crate) fn busy_executors(count: usize) { + metric::with_metrics(&METRICS, |m| m.busy_executors.set(count as i64)); +} + +/// Records one transaction queued behind an executor. +pub(crate) fn blocked_transaction() { + metric::with_metrics(&METRICS, |m| m.blocked_transactions.inc()); +} + +/// Records one queued transaction leaving an executor backlog. +pub(crate) fn unblocked_transaction() { + metric::with_metrics(&METRICS, |m| m.blocked_transactions.dec()); +} + +/// Records one account-lock conflict. +pub(crate) fn lock_conflict() { + metric::with_metrics(&METRICS, |m| m.lock_conflicts.inc()); +} + +/// Owns all Prometheus collectors registered by processor. +struct Metrics { + /// Runtime operation duration and completion counters. + operations: OperationCounters, + /// Executors currently running transaction batches. + busy_executors: IntGauge, + /// Transactions queued behind account-lock conflicts. + blocked_transactions: IntGauge, + /// Account-lock conflict counter. + lock_conflicts: IntCounter, +} + +impl Default for Metrics { + /// Builds collectors and registers them in the default Prometheus registry. + fn default() -> Self { + let metrics = Self { + operations: OperationCounters::new(OPERATION_TIME), + busy_executors: metric::gauge(BUSY_EXECUTORS, 0), + blocked_transactions: metric::gauge(BLOCKED_TRANSACTIONS, 0), + lock_conflicts: metric::counter(LOCK_CONFLICTS, 0), + }; + metrics.register(); + metrics + } +} + +impl Metrics { + /// Registers all collectors in the default Prometheus registry. + fn register(&self) { + self.operations.register("processor"); + metric::register("processor", self.busy_executors.clone()); + metric::register("processor", self.blocked_transactions.clone()); + metric::register("processor", self.lock_conflicts.clone()); + } +} diff --git a/processor/src/sequencer/locks.rs b/processor/src/sequencer/locks.rs new file mode 100644 index 0000000..8159165 --- /dev/null +++ b/processor/src/sequencer/locks.rs @@ -0,0 +1,83 @@ +//! Account lock tracking for scheduled transactions. + +use crate::executor::ExecutorId; + +pub(super) const MAX_EXECUTORS: u32 = u64::BITS - 1; +const WRITE_BIT: u64 = 1 << MAX_EXECUTORS; + +/// Tracks active account holders as executor bits plus a top write-mode bit, +/// with an optional contender granted priority on the account. +/// +/// [`WRITE_BIT`] is lock-mode metadata for the active holder set, not an +/// executor. `contender` is priority metadata and is not an active holder. +/// +/// Locks are reentrant per executor: an executor that already holds the account +/// can re-acquire it — including upgrading its read hold to a write — because a +/// conflict is only raised against a *different* executor. +#[derive(Default)] +pub(super) struct AccountLock { + /// Bitset of holding executors; [`WRITE_BIT`] set marks an exclusive lock. + lock: u64, + /// Executor that lost a conflict here and is owed the account next. + contender: Option, +} + +impl AccountLock { + /// Acquires an exclusive (write) lock for `executor`. + /// + /// Fails with the blocking executor's id if a contender other than + /// `executor` is queued, or if another executor already holds the account. + pub(super) fn write(&mut self, executor: ExecutorId) -> Result<(), ExecutorId> { + if let Some(contender) = self.contender + && contender != executor + { + return Err(contender); + } + self.contender.take(); + let holders = self.lock & !WRITE_BIT; + let bit = 1 << executor; + let others = holders & !bit; + if others != 0 { + return Err(others.trailing_zeros()); + } + self.lock = WRITE_BIT | bit; + + Ok(()) + } + + /// Acquires a shared (read) lock for `executor`. + /// + /// Fails with the blocking executor's id if a contender other than + /// `executor` is queued, or if another executor holds it for writing. + pub(super) fn read(&mut self, executor: ExecutorId) -> Result<(), ExecutorId> { + if let Some(contender) = self.contender + && contender != executor + { + return Err(contender); + } + self.contender.take(); + let holders = self.lock & !WRITE_BIT; + let bit = 1 << executor; + if self.lock & WRITE_BIT != 0 && holders & bit == 0 { + return Err(holders.trailing_zeros()); + } + self.lock |= bit; + + Ok(()) + } + + /// Releases `executor`'s hold on the account, clearing the write bit. + pub(super) fn unlock(&mut self, executor: ExecutorId) { + self.lock &= !(1 << executor | WRITE_BIT); + } + + /// Records `executor` as the contender owed the account next. + pub(super) fn contend(&mut self, executor: ExecutorId) { + self.contender.replace(executor); + } + + /// Returns whether any executor still actively holds the account. + pub(super) fn locked(&self) -> bool { + self.lock & !WRITE_BIT != 0 + } +} diff --git a/processor/src/sequencer/mod.rs b/processor/src/sequencer/mod.rs new file mode 100644 index 0000000..ba8334e --- /dev/null +++ b/processor/src/sequencer/mod.rs @@ -0,0 +1,307 @@ +//! Transaction sequencer: resolves account-lock conflicts and fans +//! non-conflicting transactions out to a pool of executors. + +use std::{sync::Arc, thread}; + +use ahash::HashMap; +use blake3::Hasher; +use keeper::{Keeper, ResolvedTransaction, TransactionView}; +use nucleus::{ + Slot, + ledger::Block, + runtime::{BarrierGuard, SequencerHandle}, + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, +}; +use solana_hash::Hash; +use solana_program_runtime::loaded_programs::ProgramCache; +use solana_pubkey::Pubkey; +use solana_svm_transaction::svm_message::SVMMessage; +use tokio::{runtime::Builder, sync::mpsc}; +use tracing::{debug, error, info, warn}; + +mod locks; +mod pool; + +#[cfg(test)] +mod tests; + +use self::{ + locks::{AccountLock, MAX_EXECUTORS}, + pool::Executors, +}; +use crate::{ + ExecutorMessage, ExecutorReady, Result, SequencerMessage, + executor::{ExecutorHandle, ExecutorId, TransactionExecutor}, + metrics::{self, Operation}, + simulator::TransactionSimulator, +}; + +/// Upper bound on transactions stalled on lock contention; once reached the +/// sequencer stops accepting new inbound work until some unblock. +const MAX_BLOCKED_TRANSACTIONS: usize = 64; + +/// Schedules inbound transactions onto executors, ordering them by per-account +/// lock conflicts and finalizing block boundaries. +pub struct Sequencer { + /// Durable engine state used for appends and account/block lookups. + state: Arc, + /// Per-account locks held by in-flight transactions, used to detect conflicts. + locks: HashMap, + /// Inbound stream of transactions and block boundaries. + rx: mpsc::Receiver, + /// Rolling hash chaining transactions into the current block's hash. + hasher: Hasher, + /// The executor pool and its availability bookkeeping. + executors: Executors, + /// Slot currently being sequenced. + slot: Slot, + /// Handle used to observe and report cooperative shutdown. + shutdown: ShutdownHandle, + /// Whether the sequencer runs in ledger-replay mode; propagated to every + /// spawned executor (see the executor's `replay` field). + replay: bool, +} + +impl Sequencer { + /// Spawns the executor pool and assembles a sequencer over the given state. + pub fn new( + executors: usize, + state: Arc, + cache: Arc, + shutdown: &mut ShutdownManager, + replay: bool, + ) -> Result<(Self, SequencerHandle)> { + let count = executors.min(MAX_EXECUTORS as usize); + metrics::init(); + let (ready_tx, ready_rx) = mpsc::channel(count); + let mut executors = Vec::with_capacity(count); + for id in 0..count as u32 { + let handle = TransactionExecutor::spawn( + id, + state.clone(), + cache.clone(), + shutdown, + ready_tx.clone(), + replay, + )?; + executors.push(handle); + } + + let hash = state.blocks().latest().hash; + let mut hasher = Hasher::new(); + hasher.update(hash.as_ref()); + let (execution, rx) = mpsc::channel(1024); + let simulation = TransactionSimulator::spawn(state.clone(), cache, shutdown)?; + let shutdown = shutdown.handle(Service::Sequencer); + + let sequencer = Self { + slot: state.blocks().current_slot(), + state, + locks: Default::default(), + executors: Executors::new(executors, ready_rx), + rx, + hasher, + shutdown, + replay, + }; + let handle = SequencerHandle { execution, simulation }; + info!(executors = count, replay, "sequencer started"); + Ok((sequencer, handle)) + } + + /// Moves the sequencer onto its own thread with a current-thread runtime. + pub fn spawn(self) -> Result<()> { + let runtime = Builder::new_current_thread().build()?; + thread::Builder::new() + .name("transaction-sequencer".into()) + .spawn(move || runtime.block_on(self.run()))?; + Ok(()) + } + + /// Main loop: drains executor-ready signals, inbound messages, and the + /// shutdown signal until cancellation, then joins the executor threads. + async fn run(mut self) { + loop { + let result = tokio::select! { + biased; + Some(msg) = self.executors.ready.recv() => { + self.handle_ready(msg) + } + _ = self.shutdown.signalled() => { + break; + } + Some(msg) = self.rx.recv(), if self.executors.ready() => { + self.handle_message(msg).await + } + }; + if let Err(error) = result { + error!(?error, "sequencer encountered a fatal error, terminating"); + break; + } + } + info!("sequencer shutdown is requested, draining the in-flight work"); + let _ = self.drain().await; + for mut e in self.executors.handles.drain(..) { + let handle = e.task.take(); + drop(e); + if let Some(handle) = handle { + let _ = handle.join(); + } + } + self.shutdown.terminate(ShutdownReason::Signalled); + } + + /// Dispatches an inbound message to the transaction or block handler. + async fn handle_message(&mut self, msg: SequencerMessage) -> Result<()> { + match msg { + SequencerMessage::Transaction(txn) => self.schedule(txn).await, + SequencerMessage::Block(block) => self.finalize(block).await, + SequencerMessage::Barrier(guard) => self.barrier(guard).await, + } + } + + /// Resolves account-lock conflicts for a transaction and either dispatches + /// it to a free executor or queues it behind the executor that blocks it. + async fn schedule(&mut self, txn: TransactionView) -> Result<()> { + let Ok(txn) = + ResolvedTransaction::try_new(txn, Some(Default::default()), &Default::default()) + else { + return Ok(()); + }; + if !self.replay && !self.state.transactions().append(&txn).await? { + return Ok(()); + }; + let Some(executor) = self.executors.available() else { + self.executors.enqueue(txn, 0); + return Ok(()); + }; + let id = executor.id; + if let Err(blocker) = self.acquire_locks(executor, &txn) { + metrics::lock_conflict(); + self.executors.enqueue(txn, blocker); + return Ok(()); + } + executor.batch.push(txn); + self.executors.dispatch(id) + } + + /// Reclaims an executor that finished its batch: releases the account locks + /// it held, then re-tries the transactions queued behind it. Any that can + /// now acquire their locks form a fresh batch dispatched back to it; the + /// rest are re-queued behind whichever executor still blocks them. + fn handle_ready(&mut self, msg: ExecutorReady) -> Result<()> { + let id = msg.id; + let Some(executor) = self.executors.release(id) else { + warn!(id, "ready signal for unknown executor; ignoring"); + return Ok(()); + }; + executor.batch = msg.batch; + for acc in executor.locks.drain() { + let Some(lock) = self.locks.get_mut(&acc.0) else { + continue; + }; + lock.unlock(id); + if !lock.locked() { + self.locks.remove(&acc.0); + } + } + while let Some(txn) = executor.blocked.pop_front() { + metrics::unblocked_transaction(); + if let Err(blocker) = self.acquire_locks(executor, &txn) { + metrics::lock_conflict(); + self.executors.enqueue(txn, blocker); + continue; + } + executor.batch.push(txn); + } + self.executors.dispatch(id) + } + + /// Raises a quiescence barrier: drains all in-flight work so the sequencer + /// and its executors are idle, acknowledges the controller, then waits to be + /// released before resuming. Used to take a consistent state snapshot at + /// superblock boundaries. + async fn barrier(&mut self, guard: BarrierGuard) -> Result<()> { + info!("sequencer is halting operation"); + self.drain().await?; + let _ = guard.acknowledged.send(()); + let _ = guard.released.await; + info!("sequencer is resuming operation"); + Ok(()) + } + + /// Awaits executor-ready signals, reclaiming each finished executor + /// until the whole pool is idle. Used to ensure that the in-flight + /// work is complete before finalizing a block and during shutdown. + async fn drain(&mut self) -> Result<()> { + let _timer = metrics::time(Operation::BarrierDrain); + while !self.executors.idle() { + let Some(signal) = self.executors.ready.recv().await else { + debug!("executor pool closed during drain; abandoning in-flight work"); + return Ok(()); + }; + self.handle_ready(signal)?; + } + Ok(()) + } + + /// Acquires all of `txn`'s account locks for `executor`, recording each on + /// the executor's hold map. On the first conflict it rolls back the locks + /// taken so far (marking the blocker as their contender) and returns the + /// blocking executor's id; on success every key is locked. + fn acquire_locks( + &mut self, + executor: &mut ExecutorHandle, + txn: &ResolvedTransaction, + ) -> std::result::Result<(), ExecutorId> { + let id = executor.id; + let mut locked = 0; + let mut result = Ok(()); + for (i, &acc) in txn.static_account_keys().iter().enumerate() { + let lock = self.locks.entry(acc).or_default(); + result = if txn.is_writable(i) { lock.write(id) } else { lock.read(id) }; + if result.is_err() { + break; + } + *executor.locks.entry(acc).or_default() += 1; + locked += 1; + } + let Err(blocker) = result else { + return Ok(()); + }; + for acc in txn.static_account_keys().iter().take(locked) { + let Some(count) = executor.locks.get_mut(acc) else { + continue; + }; + *count -= 1; + let Some(lock) = self.locks.get_mut(acc) else { + continue; + }; + lock.contend(blocker); + if *count == 0 { + lock.unlock(id); + } + } + Err(blocker) + } + + /// Finalizes the current block: chains its hash, appends it, and notifies + /// every executor of the new block boundary. + async fn finalize(&mut self, mut block: Block) -> Result<()> { + let _timer = metrics::time(Operation::FinalizeBlock); + block.hash = Hash::from(*self.hasher.finalize().as_bytes()); + self.state.blocks().append(block)?; + // Block boundaries synchronize executors: + // 1. sysvar writes bypass account locks and must be ordered deterministically + // 2. replaying should always result in transactions executing in the same block + self.drain().await?; + for e in &self.executors.handles { + e.tx.send(ExecutorMessage::Block(block)) + .map_err(|_| Service::TransactionExecutor(e.id))?; + } + self.state.accounts().update_sysvars(block)?; + self.hasher.reset().update(block.hash.as_ref()); + self.slot = self.state.blocks().current_slot(); + Ok(()) + } +} diff --git a/processor/src/sequencer/pool.rs b/processor/src/sequencer/pool.rs new file mode 100644 index 0000000..6bba9e6 --- /dev/null +++ b/processor/src/sequencer/pool.rs @@ -0,0 +1,142 @@ +//! Executor-pool bookkeeping for the sequencer. + +use std::mem; + +use keeper::ResolvedTransaction; +use nucleus::shutdown::Service; +use tokio::sync::mpsc::Receiver; +use tracing::warn; + +use super::MAX_BLOCKED_TRANSACTIONS; +use crate::{ + ExecutorMessage, ExecutorReady, Result, + executor::{ExecutorHandle, ExecutorId}, + metrics, +}; + +/// The pool of executors and the state needed to dispatch work to them. +pub(super) struct Executors { + /// One handle per executor worker, indexed by [`ExecutorId`]. + pub(super) handles: Vec, + /// Channel on which executors signal they have finished a batch. + pub(super) ready: Receiver, + /// Bitset of executors currently free to accept a batch. + available: AvailableExecutors, +} + +/// Bitset of free executors, one bit per [`ExecutorId`]. +pub(super) struct AvailableExecutors { + /// Set bits are executor IDs currently free to accept work. + bitflags: u64, + /// Number of executor slots represented by `bitflags`. + total: u32, +} + +impl Executors { + /// Builds the pool from spawned executor handles and their readiness channel. + pub(super) fn new(handles: Vec, ready: Receiver) -> Self { + let available = AvailableExecutors::new(handles.len() as u32); + Self { handles, ready, available } + } + + /// Whether the pool can accept more work: at least one executor is free and + /// the total blocked backlog is below `MAX_BLOCKED_TRANSACTIONS`. + pub(super) fn ready(&self) -> bool { + let blocked = || self.handles.iter().map(|h| h.blocked.len()).sum::(); + !self.available.empty() && blocked() < MAX_BLOCKED_TRANSACTIONS + } + + /// Returns whether every executor is currently free. + pub(super) fn idle(&self) -> bool { + self.available.idle() + } + + /// Queues a transaction behind the executor that currently blocks it, to be + /// retried once that executor releases its conflicting locks. + pub(super) fn enqueue(&mut self, txn: ResolvedTransaction, executor: ExecutorId) { + self.handles[executor as usize].blocked.push_back(txn); + metrics::blocked_transaction(); + } + + /// Returns a handle to a currently free executor, or `None` if all are busy. + /// The executor is not yet marked busy; the caller does that once it commits + /// a batch to it. + pub(super) fn available<'b>(&mut self) -> Option<&'b mut ExecutorHandle> { + self.available.get().and_then(|idx| self.get(idx)) + } + + /// Returns the handle for executor `idx`, if such an executor exists. + fn get<'b>(&mut self, idx: ExecutorId) -> Option<&'b mut ExecutorHandle> { + let handle = self.handles.get_mut(idx as usize)?; + // SAFETY: launders the borrow lifetime only; it never aliases. Handles + // are reachable solely through `Executors`, and every caller uses the + // returned `&mut` within one stack frame, so no two coexist and none + // outlives `self`. + Some(unsafe { mem::transmute::<&mut _, &mut _>(handle) }) + } + + /// Marks executor `idx` as available again and returns its handle. + pub(super) fn release<'b>(&mut self, idx: ExecutorId) -> Option<&'b mut ExecutorHandle> { + self.available.insert(idx); + metrics::busy_executors(self.available.busy()); + self.get(idx) + } + + /// Sends executor `idx`'s accumulated batch to its worker and marks it busy. + /// A no-op if the executor is unknown or its batch is empty. + pub(super) fn dispatch(&mut self, idx: ExecutorId) -> Result<()> { + let Some(executor) = self.get(idx) else { + warn!(idx, "dispatch to unknown executor; ignoring"); + return Ok(()); + }; + if executor.batch.is_empty() { + return Ok(()); + } + let msg = ExecutorMessage::Transactions(mem::take(&mut executor.batch)); + executor.tx.send(msg).map_err(|_| Service::TransactionExecutor(idx))?; + self.available.remove(idx); + metrics::busy_executors(self.available.busy()); + Ok(()) + } +} + +impl AvailableExecutors { + /// Starts with every executor marked available. + pub(super) fn new(executors: u32) -> Self { + Self { + bitflags: (1u64 << executors) - 1, + total: executors, + } + } + + /// Returns the id of an available executor, or `None` if all are busy. + pub(super) fn get(&self) -> Option { + let position = self.bitflags.trailing_zeros(); + (position != u64::BITS).then_some(position) + } + + /// Returns whether no executor is currently available. + pub(super) fn empty(&self) -> bool { + self.bitflags == 0 + } + + /// Returns whether every executor is currently free. + pub(super) fn idle(&self) -> bool { + self.bitflags.count_ones() == self.total + } + + /// Returns how many executors are currently busy. + pub(super) fn busy(&self) -> usize { + (self.total - self.bitflags.count_ones()) as usize + } + + /// Marks an executor as busy. + pub(super) fn remove(&mut self, executor: ExecutorId) { + self.bitflags &= !(1 << executor) + } + + /// Marks an executor as available again. + pub(super) fn insert(&mut self, executor: ExecutorId) { + self.bitflags |= 1 << executor + } +} diff --git a/processor/src/sequencer/tests.rs b/processor/src/sequencer/tests.rs new file mode 100644 index 0000000..e6e533f --- /dev/null +++ b/processor/src/sequencer/tests.rs @@ -0,0 +1,268 @@ +//! Sequencer unit tests for account locks and executor availability. +//! +//! These tests stay below the public processor surface so they can exercise the +//! scheduling invariants directly: lock fairness, partial-acquire behavior, and +//! the bookkeeping that decides when executor work can be drained. + +use std::{ + collections::VecDeque, + sync::{Arc, mpsc}, +}; + +use blake3::Hasher; +use keeper::{ + Keeper, + testkit::{TestKeeper, resolved}, +}; +use nucleus::shutdown::{Service, ShutdownManager}; +use solana_pubkey::Pubkey; +use tokio::sync::mpsc as tokio_mpsc; + +use super::{ + Sequencer, + locks::AccountLock, + pool::{AvailableExecutors, Executors}, +}; +use crate::{ + ExecutorMessage, ExecutorReady, + executor::{ExecutorHandle, ExecutorId}, +}; + +/// Constructs a bare sequencer around `state` without spawning executors. +/// +/// Tests fill only the fields needed to call scheduling helpers directly; the +/// message channels are intentionally inert. +fn sequencer(state: Arc, shutdown: &mut ShutdownManager) -> Sequencer { + let (_tx, rx) = tokio_mpsc::channel(1); + let (_ready_tx, ready_rx) = tokio_mpsc::channel(1); + let hash = state.blocks().latest().hash; + let mut hasher = Hasher::new(); + hasher.update(hash.as_ref()); + Sequencer { + slot: state.blocks().current_slot(), + state, + locks: Default::default(), + rx, + hasher, + executors: Executors::new(Vec::new(), ready_rx), + shutdown: shutdown.handle(Service::Sequencer), + replay: false, + } +} + +/// Builds an idle executor handle with no worker task behind its channel. +fn executor(id: ExecutorId) -> ExecutorHandle { + let (tx, _rx) = mpsc::sync_channel(1); + ExecutorHandle { + id, + batch: Vec::new(), + locks: Default::default(), + blocked: VecDeque::new(), + tx, + task: None, + } +} + +/// Builds an idle executor handle together with the receiver end of its dispatch +/// channel, so a test can observe the batch `dispatch` actually sends. +fn executor_with_rx(id: ExecutorId) -> (ExecutorHandle, mpsc::Receiver) { + let (tx, rx) = mpsc::sync_channel(1); + let handle = ExecutorHandle { + id, + batch: Vec::new(), + locks: Default::default(), + blocked: VecDeque::new(), + tx, + task: None, + }; + (handle, rx) +} + +// Readers share a lock until a writer contends, then the writer waits for every +// current reader and is granted before later readers. +#[test] +fn read_locks_share_until_a_writer_arrives() { + let mut lock = AccountLock::default(); + + assert_eq!(lock.read(0), Ok(())); + assert_eq!(lock.read(1), Ok(())); + assert_eq!(lock.write(2), Err(0)); + assert!(lock.locked()); + + lock.unlock(0); + assert_eq!(lock.write(2), Err(1)); + lock.unlock(1); + assert_eq!(lock.write(2), Ok(())); +} + +// One executor may reacquire a lock it already owns and may upgrade its own read +// lock to a write lock without blocking itself. +#[test] +fn same_executor_can_reenter_and_upgrade() { + let mut lock = AccountLock::default(); + + assert_eq!(lock.read(4), Ok(())); + assert_eq!(lock.write(4), Ok(())); + assert_eq!(lock.read(4), Ok(())); + + lock.unlock(4); + assert!(!lock.locked()); + assert_eq!(lock.read(5), Ok(())); +} + +// A contending executor keeps priority after it is queued, preventing unrelated +// readers from slipping ahead while the current writer drains. +#[test] +fn contender_gets_priority_until_granted() { + let mut lock = AccountLock::default(); + + assert_eq!(lock.write(0), Ok(())); + lock.contend(1); + assert_eq!(lock.read(2), Err(1)); + + lock.unlock(0); + assert_eq!(lock.read(1), Ok(())); + assert_eq!(lock.read(2), Ok(())); +} + +// If acquiring a multi-account transaction fails partway through, already-held +// locks keep the blocked executor marked as the contender until its turn. +#[tokio::test(flavor = "current_thread")] +async fn acquire_locks_preserves_contender_priority_after_partial_conflict() { + let mut tk = TestKeeper::new().await; + let mut sequencer = sequencer(tk.state(), tk.shutdown()); + let a = Pubkey::new_unique(); + let b = Pubkey::new_unique(); + let mut blocker = executor(0); + let mut blocked = executor(1); + + sequencer + .acquire_locks(&mut blocker, &resolved(&[(b, true)])) + .expect("blocker locks b"); + let err = sequencer + .acquire_locks(&mut blocked, &resolved(&[(a, true), (b, true)])) + .expect_err("b blocks the second transaction"); + + assert_eq!(err, 0); + assert_eq!(blocked.locks.get(&a), Some(&0)); + assert!(sequencer.locks.get_mut(&a).expect("a lock remains").read(1).is_err()); + assert_eq!( + sequencer.locks.get_mut(&a).expect("a lock remains").read(0), + Ok(()) + ); + assert!(blocker.locks.contains_key(&b)); +} + +// Executor availability reports the first idle executor, the busy count, and the +// all-idle/all-busy states used by sequencer drain logic. +#[test] +fn available_executors_track_busy_and_idle_state() { + let mut available = AvailableExecutors::new(3); + + assert_eq!(available.get(), Some(0)); + assert!(available.idle()); + available.remove(0); + available.remove(2); + + assert_eq!(available.get(), Some(1)); + assert_eq!(available.busy(), 2); + assert!(!available.empty()); + assert!(!available.idle()); + + available.remove(1); + assert!(available.empty()); + assert_eq!(available.get(), None); + + available.insert(2); + assert_eq!(available.get(), Some(2)); + assert_eq!(available.busy(), 2); +} + +// When a freed executor retries a transaction queued behind it, and that +// transaction re-acquires its now-free lock but then conflicts with a lock still +// held by another executor, it rolls back and is re-queued behind the new blocker. +#[tokio::test(flavor = "current_thread")] +async fn handle_ready_requeues_behind_the_new_blocker() { + let mut tk = TestKeeper::new().await; + let mut sequencer = sequencer(tk.state(), tk.shutdown()); + let x = Pubkey::new_unique(); + let y = Pubkey::new_unique(); + + let mut e0 = executor(0); + let mut e1 = executor(1); + sequencer.acquire_locks(&mut e0, &resolved(&[(x, true)])).expect("e0 locks x"); + sequencer.acquire_locks(&mut e1, &resolved(&[(y, true)])).expect("e1 locks y"); + // A transaction needing both x and y is parked behind executor 0, the x holder. + e0.blocked.push_back(resolved(&[(x, true), (y, true)])); + + let (_ready_tx, ready_rx) = tokio_mpsc::channel(1); + sequencer.executors = Executors::new(vec![e0, e1], ready_rx); + + // Executor 0 finishes: x is released, but the retried transaction now conflicts + // with y (still held by executor 1) and is re-parked behind executor 1. + sequencer + .handle_ready(ExecutorReady { id: 0, batch: Vec::new() }) + .expect("ready handled"); + + assert_eq!( + sequencer.executors.handles[1].blocked.len(), + 1, + "requeued behind y's holder" + ); + assert!( + sequencer.executors.handles[0].blocked.is_empty(), + "no longer queued behind x" + ); + assert!( + sequencer.executors.handles[0].batch.is_empty(), + "nothing dispatched to executor 0" + ); + // x was rolled back (no holder) but keeps executor 1 as its priority contender. + let x_lock = sequencer.locks.get_mut(&x).expect("x lock retained"); + assert!(!x_lock.locked()); + assert_eq!( + x_lock.read(2), + Err(1), + "blocker keeps contender priority on x" + ); + // y is still write-held by executor 1. + assert_eq!( + sequencer.locks.get_mut(&y).expect("y lock retained").read(2), + Err(1) + ); +} + +// A freed executor re-dispatches a transaction queued behind it once it can fully +// re-acquire its locks, handing it back to the executor in a fresh batch. +#[tokio::test(flavor = "current_thread")] +async fn handle_ready_redispatches_unblocked_transaction() { + let mut tk = TestKeeper::new().await; + let mut sequencer = sequencer(tk.state(), tk.shutdown()); + let x = Pubkey::new_unique(); + + let (mut e0, rx) = executor_with_rx(0); + sequencer.acquire_locks(&mut e0, &resolved(&[(x, true)])).expect("e0 locks x"); + e0.blocked.push_back(resolved(&[(x, true)])); + + let (_ready_tx, ready_rx) = tokio_mpsc::channel(1); + sequencer.executors = Executors::new(vec![e0], ready_rx); + + // Executor 0 finishes: x is released, the queued transaction re-acquires it and + // is dispatched straight back to executor 0. + sequencer + .handle_ready(ExecutorReady { id: 0, batch: Vec::new() }) + .expect("ready handled"); + + let ExecutorMessage::Transactions(batch) = rx.try_recv().expect("batch dispatched") else { + panic!("dispatched a transaction batch"); + }; + assert_eq!(batch.len(), 1); + assert!(batch[0].static_account_keys().contains(&x)); + // x is held again for the redispatched transaction, and its queue is empty. + assert_eq!( + sequencer.locks.get_mut(&x).expect("x lock").read(2), + Err(0), + "x re-held by executor 0" + ); + assert!(sequencer.executors.handles[0].blocked.is_empty()); +} diff --git a/processor/src/simulator.rs b/processor/src/simulator.rs new file mode 100644 index 0000000..1a155d7 --- /dev/null +++ b/processor/src/simulator.rs @@ -0,0 +1,112 @@ +//! Transaction simulation: executes transactions against current state on +//! owned account copies, without committing any changes. + +use std::{sync::Arc, thread}; + +use keeper::{ExecutionRecord, Keeper, ResolvedTransaction}; +use nucleus::{ + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + tls::AUTHORITY, +}; +use solana_program_runtime::loaded_programs::ProgramCache; +use solana_transaction_error::TransactionError; +use tokio::{ + runtime::Builder, + sync::mpsc::{self, Receiver, Sender}, +}; +use tracing::debug; + +use crate::{Result, Simulation, SimulatorMessage, callback::SVMCallback, svm::SvmContext}; + +/// Worker that simulates transactions against current state without committing. +pub struct TransactionSimulator { + /// Inbound channel of simulation requests and block boundaries. + rx: Receiver, + /// SVM batch processor and per-block environment driving simulation. + svm: SvmContext, + /// Durable engine state used for account loads. + state: Arc, + /// Handle used to report cooperative shutdown for this worker. + shutdown: ShutdownHandle, +} + +impl TransactionSimulator { + /// Builds the SVM environment and spawns the simulator on its own thread. + pub fn spawn( + state: Arc, + cache: Arc, + shutdown: &mut ShutdownManager, + ) -> Result> { + let svm = SvmContext::new(&state, cache)?; + let shutdown = shutdown.handle(Service::TransactionSimulator); + let (tx, rx) = mpsc::channel(16); + let executor = Self { rx, svm, state, shutdown }; + let runtime = Builder::new_current_thread().build()?; + thread::Builder::new() + .name("transaction-simulator".into()) + .spawn(move || runtime.block_on(executor.run()))?; + Ok(tx) + } + + /// Worker loop: simulates requests and applies block transitions until the + /// channel closes, then reports cooperative shutdown. + async fn run(mut self) { + // Mirror the executor: simulated MagicRoot calls authorize against the + // same authority published on this simulator thread. + AUTHORITY.set(self.state.authority()); + loop { + tokio::select! { + biased; + _ = self.shutdown.signalled() => { + break; + } + Some(msg) = self.rx.recv() => { + self.handle_message(msg); + } + } + } + self.shutdown.terminate(ShutdownReason::Signalled); + } + + fn handle_message(&mut self, msg: SimulatorMessage) { + match msg { + SimulatorMessage::Transaction(simulation) => { + self.process(simulation); + } + SimulatorMessage::Block(block) => self.svm.transition(block), + SimulatorMessage::Barrier(guard) => { + let _ = guard.acknowledged.send(()); + let _ = guard.released.recv(); + } + }; + } + + /// Resolves and executes a single transaction on owned account copies, + /// producing a result without persisting any state. + fn process(&mut self, simulation: Simulation) { + let result = ResolvedTransaction::try_new( + simulation.transaction, + Some(Default::default()), + &Default::default(), + ); + let Ok(txn) = result else { + // transaction resolution can only fail if ALTs are used + debug!("simulation rejected: invalid address lookup table"); + let error = Err(TransactionError::InvalidAddressLookupTableData); + let _ = simulation.response.send(error); + return; + }; + let accessor = self.state.accounts(); + let callback = SVMCallback:: { + loader: accessor.loader(), + featureset: self.state.features(), + }; + let output = self.svm.execute(&callback, &txn, self.state.features()); + let execution = ExecutionRecord { + result: output.processing_result, + balances: output.balance_collector, + slot: self.svm.slot(), + }; + let _ = simulation.response.send(Ok(execution)); + } +} diff --git a/processor/src/svm.rs b/processor/src/svm.rs new file mode 100644 index 0000000..978a795 --- /dev/null +++ b/processor/src/svm.rs @@ -0,0 +1,130 @@ +//! Shared SVM execution context for the transaction executor and simulator. + +use std::sync::Arc; + +use agave_feature_set::FeatureSet; +use keeper::{Keeper, ResolvedTransaction}; +use nucleus::{Slot, ledger::Block}; +use solana_compute_budget_instruction::instructions_processor::process_compute_budget_instructions; +use solana_program_runtime::loaded_programs::{ProgramCache, ProgramRuntimeEnvironments}; +use solana_svm::{ + account_loader::CheckedTransactionDetails, + transaction_processing_callback::TransactionProcessingCallback, + transaction_processor::{ + ExecutionRecordingConfig, LoadAndExecuteSanitizedTransactionOutput, + TransactionBatchProcessor, TransactionProcessingConfig, TransactionProcessingEnvironment, + }, +}; +use solana_sysvar::clock::Clock; +use solana_transaction_error::TransactionResult; + +use crate::{Result, callback::SVMCallback}; + +/// SVM batch processor together with its per-block environment and recording +/// config, shared verbatim by the executor and simulator workers. +pub(crate) struct SvmContext { + /// SVM batch processor that loads and executes transactions. + processor: TransactionBatchProcessor, + /// Per-block processing environment (blockhash, features, rent, ...). + env: TransactionProcessingEnvironment, + /// Recording and logging configuration for execution. + config: TransactionProcessingConfig, +} + +impl SvmContext { + /// Builds the SVM runtime environment and batch processor from current state. + pub(crate) fn new(state: &Keeper, cache: Arc) -> Result { + let budget = Default::default(); + let runtime_env = agave_syscalls::create_program_runtime_environment( + &state.features().runtime_features(), + &budget, + true, + false, + ) + .map_err(|e| e.to_string())?; + let envs = ProgramRuntimeEnvironments::new(runtime_env); + let block = state.blocks().latest(); + let env = TransactionProcessingEnvironment { + blockhash: block.hash, + feature_set: state.features().runtime_features(), + program_runtime_environments_for_execution: envs, + rent: state.rent().clone(), + blockhash_lamports_per_signature: 0, + epoch_total_stake: 0, + }; + let mut processor = TransactionBatchProcessor::new(block.slot + 1, cache); + let accessor = state.accounts(); + let callback = SVMCallback:: { + loader: accessor.loader(), + featureset: state.features(), + }; + processor.fill_missing_sysvar_cache_entries(&callback); + let config = TransactionProcessingConfig { + log_messages_bytes_limit: None, + recording_config: ExecutionRecordingConfig::new_single_setting(true), + }; + Ok(Self { processor, env, config }) + } + + /// Loads and executes a single transaction through the SVM. + pub(crate) fn execute( + &self, + callback: &impl TransactionProcessingCallback, + txn: &ResolvedTransaction, + features: &FeatureSet, + ) -> LoadAndExecuteSanitizedTransactionOutput { + let details = match self.parse_details(txn, features) { + Ok(d) => d, + Err(e) => { + return LoadAndExecuteSanitizedTransactionOutput { + processing_result: Err(e), + balance_collector: None, + }; + } + }; + self.processor.load_and_execute_sanitized_transaction( + callback, + txn, + details, + &self.env, + &self.config, + ) + } + + /// Advances the context to a new block: bumps the slot and blockhash and + /// refreshes the cached clock sysvar. + pub(crate) fn transition(&mut self, block: Block) { + let slot = block.slot + 1; + let hash = block.hash; + self.processor.slot = slot; + self.env.blockhash = hash; + let clock = Clock { + slot, + unix_timestamp: block.time, + ..Default::default() + }; + self.processor.sysvar_cache_mut().set_clock(&clock); + } + + /// Slot the context is currently executing against. + pub(crate) fn slot(&self) -> Slot { + self.processor.slot + } + + /// Derives the compute budget and limits from the transaction's compute-budget + /// instructions. Fees are forced to zero and depth-8 CPIs disabled on the ER. + pub(crate) fn parse_details( + &self, + txn: &ResolvedTransaction, + features: &FeatureSet, + ) -> TransactionResult { + let ixs = txn.program_instructions_iter(); + let limits = process_compute_budget_instructions(ixs, features)?; + let limits = limits.get_compute_budget_and_limits( + limits.loaded_accounts_bytes, + Default::default(), // Fee is always zero on ER + false, // Depth-8 CPIs are disabled on solana + ); + Ok(CheckedTransactionDetails::new(None, limits)) + } +} diff --git a/processor/src/tests.rs b/processor/src/tests.rs new file mode 100644 index 0000000..b66b478 --- /dev/null +++ b/processor/src/tests.rs @@ -0,0 +1,402 @@ +//! End-to-end processor checks driven by the loadable v42 calculator program. + +use std::{sync::Arc, time::Duration}; + +use crate::{SequencerMessage, SimulatorMessage, sequencer::Sequencer}; +use keeper::{ + TransactionView, + testkit::{TestKeeper, V42_ID, load_v42, signed_view, store_v42}, +}; +use nucleus::{ + Slot, + ledger::Block, + runtime::{Simulation, barrier}, +}; +use solana_account::{AccountMode, ReadableAccount}; +use solana_hash::Hash; +use solana_instruction::Instruction; +use solana_keypair::Keypair; +use solana_program_runtime::loaded_programs::ProgramCache; +use solana_pubkey::Pubkey; +use solana_sdk_ids::loader_v4; +use solana_signature::Signature; +use solana_svm::transaction_processing_result::{ + TransactionProcessingResult, TransactionProcessingResultExtensions, +}; +use solana_transaction_error::TransactionResult; +use tokio::time::timeout; +use v42_calculator_interface::builder::Expr as E; + +/// End-to-end fixture with a real keeper, sequencer, simulator, and v42 program. +/// +/// The tempdirs must live as long as the keeper because accountsdb and ledger +/// keep files open below them. Each harness owns one funded payer so generated +/// transactions can be signed against the keeper's current blockhash. +struct Harness { + /// Seeded keeper (v42 + funded payer) owning the engine state along with the + /// directories and shutdown lifecycle its background services run on. + keeper: TestKeeper, + /// Channels used by tests to drive execution and simulation paths. + handle: nucleus::runtime::SequencerHandle, +} + +impl Harness { + /// Builds a seeded keeper and wires a sequencer/simulator onto its lifecycle. + /// + /// `replay` selects whether the sequencer records transaction status while + /// still committing account state, matching the processor replay mode. + async fn new(replay: bool) -> Self { + let (harness, sequencer) = Self::unspawned(replay).await; + sequencer.spawn().expect("sequencer spawns"); + harness + } + + /// Builds a seeded keeper and sequencer without starting the sequencer loop. + /// + /// This lets tests fill the execution channel before the sequencer can + /// consume from it, forcing contention resolution to happen from a backlog. + async fn unspawned(replay: bool) -> (Self, Sequencer) { + let mut keeper = TestKeeper::new().await; + let (sequencer, handle) = Sequencer::new( + 2, + keeper.state(), + Arc::new(ProgramCache::default()), + keeper.shutdown(), + replay, + ) + .expect("sequencer builds"); + (Self { keeper, handle }, sequencer) + } + + /// Stores one delegated v42-owned u64 account and returns its pubkey. + fn account(&self, value: u64) -> Pubkey { + store_v42(&self.keeper, value, AccountMode::Delegated) + } + + /// Signs `instruction` with the harness payer and returns its first signature + /// plus the sanitized transaction view consumed by processor services. + fn tx(&self, instruction: Instruction) -> (Signature, TransactionView) { + signed_view( + self.keeper.payer(), + instruction, + self.keeper.blocks().latest().hash, + ) + } + + /// Signs `instruction` with a fresh funded payer. + /// + /// A shared payer would become a writable account in every transaction and + /// hide the account-intersection patterns these tests are trying to stress. + fn tx_with_fresh_payer(&self, instruction: Instruction) -> (Signature, TransactionView) { + let payer = Keypair::new(); + signed_view(&payer, instruction, self.keeper.blocks().latest().hash) + } + + /// Queues a transaction on the execution path without waiting for commit. + async fn execute(&self, tx: TransactionView) { + self.handle + .execution + .send(SequencerMessage::Transaction(tx)) + .await + .expect("transaction sends"); + } + + /// Runs a transaction through simulation and returns the raw processing result. + /// + /// Simulation shares the keeper state but must not persist account writes or + /// transaction status. + async fn simulate(&self, tx: TransactionView) -> TransactionProcessingResult { + let (response, rx) = oneshot::channel(); + self.handle + .simulation + .send(SimulatorMessage::Transaction(Simulation { + transaction: tx, + response, + })) + .await + .expect("simulation sends"); + rx.await.expect("simulation responds").expect("simulation resolves").result + } + + /// Sends the same block transition to execution and simulation services. + /// + /// Both paths maintain sysvar caches, so block metadata must be delivered to + /// both before comparing simulated and committed behavior. + async fn block(&self, block: Block) { + self.handle + .execution + .send(SequencerMessage::Block(block)) + .await + .expect("block sends to execution"); + self.handle + .simulation + .send(SimulatorMessage::Block(block)) + .await + .expect("block sends to simulation"); + } + + /// Waits for all previously submitted execution work to finish, failing + /// quickly if lock contention stops making progress. + async fn barrier(&self) { + let (controller, guard) = barrier(); + self.handle + .execution + .send(SequencerMessage::Barrier(guard)) + .await + .expect("barrier sends"); + timeout(Duration::from_secs(8), controller.acknowledged) + .await + .expect("barrier timed out") + .expect("barrier acknowledged"); + controller.released.send(()).expect("barrier releases"); + } + + /// Reads the little-endian u64 payload stored in a v42 output account. + fn u64_at(&self, key: Pubkey) -> u64 { + load_v42(&self.keeper, key).expect("account exists") + } + + /// Returns the committed transaction result recorded for `signature`. + async fn status(&self, signature: Signature) -> (TransactionResult<()>, Slot) { + let status = self + .keeper + .transactions() + .status(signature) + .await + .expect("status loads") + .expect("status exists"); + (status.result, status.slot) + } + + /// Asserts replay mode left no transaction status entry for `signature`. + async fn assert_no_status(&self, signature: Signature) { + assert!( + self.keeper + .transactions() + .status(signature) + .await + .expect("status read") + .is_none() + ); + } + + /// Drops service handles and waits for background tasks to stop. + async fn close(self) { + let Self { keeper, handle } = self; + keeper.flush().expect("disk flush should succeed"); + drop(handle); + keeper.close().await; + } +} + +/// Asserts that SVM processing accepted the transaction. +fn assert_success(result: &TransactionProcessingResult) { + assert!(result.flattened_result().is_ok(), "{result:?}"); +} + +/// Wraps an expression in nested self-CPI calls. +fn with_cpi_depth(mut expr: E, depth: usize) -> E { + for _ in 0..depth { + expr = expr.cpi(); + } + expr +} + +/// Picks a read-only operand different from `output`. +fn operand(accounts: &[Pubkey], output: Pubkey, index: usize) -> Pubkey { + let mut key = accounts[index % accounts.len()]; + if key == output { + key = accounts[(index + 1) % accounts.len()]; + } + key +} + +// Simulation and execution both accept the same v42 transaction, but only +// execution commits account state and records transaction status. +#[tokio::test(flavor = "current_thread")] +async fn execution_commits_and_simulation_does_not() { + let harness = Harness::new(false).await; + let output = harness.account(0); + let input = harness.account(40); + let ix = (E::acc(1) + E::lit(2)).compose(output, &[input]); + let (_, sim_tx) = harness.tx(ix.clone()); + + assert_success(&harness.simulate(sim_tx).await); + assert_eq!(harness.u64_at(output), 0); + + let (signature, tx) = harness.tx(ix); + harness.execute(tx).await; + harness.barrier().await; + + assert_eq!(harness.u64_at(output), 42); + harness.status(signature).await.0.expect("successful execution"); + harness.close().await; +} + +// Program seeding installs the v42 ELF under loader-v4, and recursive CPI +// preserves return-data flow through a committed execution. +#[tokio::test(flavor = "current_thread")] +async fn seeded_program_and_recursive_cpi_return_data_work() { + let harness = Harness::new(false).await; + let account = harness + .keeper + .accounts() + .get(&V42_ID) + .expect("program loads") + .expect("program exists"); + + assert!(account.executable()); + assert_eq!(*account.owner(), loader_v4::ID); + assert_eq!(account.data().get(..4), Some(&[0x7f, b'E', b'L', b'F'][..])); + + let output = harness.account(0); + let expr = E::lit(42) + (E::lit(31) * E::lit(4)).cpi() - E::lit(56); + let (_, tx) = harness.tx(expr.compose(output, &[])); + + harness.execute(tx).await; + harness.barrier().await; + + assert_eq!(harness.u64_at(output), 110); + harness.close().await; +} + +// A block transition updates the Clock sysvar for both simulation and execution, +// while committed execution lands in the next slot. +#[tokio::test(flavor = "current_thread")] +async fn block_transition_updates_execution_and_simulation_sysvars() { + let harness = Harness::new(false).await; + let block = Block { + slot: 7, + hash: Hash::new_from_array([7; 32]), + time: 1234, + }; + harness.block(block).await; + + let output = harness.account(0); + let ix = E::clock().compose(output, &[]); + let (_, sim_tx) = harness.tx(ix.clone()); + assert_success(&harness.simulate(sim_tx).await); + + let (signature, tx) = harness.tx(ix); + harness.execute(tx).await; + harness.barrier().await; + + assert_eq!(harness.u64_at(output), 1234); + assert_eq!(harness.status(signature).await.1, 8); + harness.close().await; +} + +// Replay mode still applies account writes, but it must not publish transaction +// status because replayed entries were already recorded by the original run. +#[tokio::test(flavor = "current_thread")] +async fn replay_mode_commits_state_without_recording_status() { + let harness = Harness::new(true).await; + let output = harness.account(0); + let (signature, tx) = harness.tx(E::lit(9).compose(output, &[])); + + harness.execute(tx).await; + harness.barrier().await; + + assert_eq!(harness.u64_at(output), 9); + harness.assert_no_status(signature).await; + harness.close().await; +} + +// A transaction that runs but fails still commits a status receipt carrying the +// error, and leaves its output account untouched — the failure is recorded, not +// silently dropped like an unresolvable transaction. +#[tokio::test(flavor = "current_thread")] +async fn failed_execution_records_an_error_status() { + let harness = Harness::new(false).await; + let output = harness.account(5); + // 1 - 2 underflows the program's checked_sub, so it returns an error before + // ever writing the output account. + let (signature, tx) = harness.tx((E::lit(1) - E::lit(2)).compose(output, &[])); + + harness.execute(tx).await; + harness.barrier().await; + + // The failed run is committed as a status with an error result, unlike the + // success cases the other tests assert. + let status = harness + .keeper + .transactions() + .status(signature) + .await + .expect("status loads") + .expect("failed transaction still records a status"); + assert!( + status.result.is_err(), + "recorded result reflects the failure" + ); + // The account keeps its pre-execution value; nothing was committed. + assert_eq!( + harness.u64_at(output), + 5, + "failed execution commits no writes" + ); + harness.close().await; +} + +// A backlog of writes to one account must keep making progress even when every +// transaction initially contends for the same lock. +#[tokio::test(flavor = "current_thread")] +async fn prefilled_same_writable_account_backlog_drains() { + const TRANSACTIONS: usize = 128; + + let (harness, sequencer) = Harness::unspawned(false).await; + let output = harness.account(0); + let mut signatures = Vec::with_capacity(TRANSACTIONS); + + for i in 0..TRANSACTIONS { + let expr = with_cpi_depth(E::lit((i + 1) as u64), i % 5); + let (signature, tx) = harness.tx_with_fresh_payer(expr.compose(output, &[])); + signatures.push(signature); + harness.execute(tx).await; + } + + sequencer.spawn().expect("sequencer spawns"); + harness.barrier().await; + + assert_eq!(harness.u64_at(output), TRANSACTIONS as u64); + for signature in signatures { + harness.status(signature).await.0.expect("successful execution"); + } + harness.close().await; +} + +// Mixed read-only and writable intersections should eventually resolve even +// when the sequencer starts with more conflicted work than it can keep unblocked +// at once. +#[tokio::test(flavor = "current_thread")] +async fn prefilled_mixed_read_write_contention_stress_drains() { + const ACCOUNTS: usize = 16; + const TRANSACTIONS: usize = 512; + + let (harness, sequencer) = Harness::unspawned(false).await; + let accounts: Vec<_> = (0..ACCOUNTS).map(|i| harness.account(i as u64 + 1)).collect(); + let mut signatures = Vec::with_capacity(TRANSACTIONS); + + for i in 0..TRANSACTIONS { + let output = match i % 4 { + 0 => accounts[0], + 1 => accounts[i % ACCOUNTS], + 2 => accounts[(i + 3) % ACCOUNTS], + _ => accounts[(i * 7 + 5) % ACCOUNTS], + }; + let left = operand(&accounts, output, i + 1); + let right = operand(&accounts, output, i * 3 + 2); + let expr = with_cpi_depth(E::acc(1) + E::lit((i % 5) as u64), i % 5); + let (signature, tx) = harness.tx_with_fresh_payer(expr.compose(output, &[left, right])); + signatures.push(signature); + harness.execute(tx).await; + } + + sequencer.spawn().expect("sequencer spawns"); + harness.barrier().await; + + for signature in signatures { + harness.status(signature).await.0.expect("successful execution"); + } + harness.close().await; +}