diff --git a/Cargo.toml b/Cargo.toml index d6af573..044252f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "accountsdb", + "engine", "keeper", "ledger", "nucleus", @@ -28,6 +29,7 @@ version = "0.1.0" [workspace.dependencies] accountsdb = { path = "accountsdb", package = "magicblock-accountsdb" } +engine = { path = "engine", package = "magicblock-engine" } keeper = { path = "keeper", package = "magicblock-keeper" } ledger = { path = "ledger", package = "magicblock-ledger" } magic-root-interface = { path = "programs/magic-root-interface" } diff --git a/clippy.toml b/clippy.toml index db02f67..19eebeb 100644 --- a/clippy.toml +++ b/clippy.toml @@ -19,7 +19,6 @@ max-struct-bools = 2 check-private-items = true disallowed-methods = [ - { path = "std::option::Option::unwrap", reason = "Use expect with context or handle the None case explicitly" }, - { path = "std::result::Result::unwrap", reason = "Use expect with context or propagate the error explicitly" }, - { path = "std::thread::sleep", reason = "Avoid timing-based flakiness in workspace code; isolate retries/backoff behind abstractions" } + { path = "std::thread::sleep", reason = "Avoid timing-based flakiness in workspace code; use event driven logic" }, + { path = "tokio::time::sleep", reason = "Avoid timing-based flakiness in workspace code; use event driven logic" } ] diff --git a/engine/Cargo.toml b/engine/Cargo.toml new file mode 100644 index 0000000..05060e1 --- /dev/null +++ b/engine/Cargo.toml @@ -0,0 +1,54 @@ +[package] +name = "magicblock-engine" + +authors.workspace = true +edition.workspace = true +homepage.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true + +[lib] +name = "engine" + +[features] +testkit = ["keeper/testkit", "nucleus/testkit", "tokio/time"] + +[dependencies] +keeper = { workspace = true } +ledger = { workspace = true } +magic-root-interface = { workspace = true } +magic-root-program = { workspace = true } +nucleus = { workspace = true, features = ["shutdown"] } +processor = { workspace = true } + +derive_more = { workspace = true } +num_cpus = { workspace = true } +oneshot = { workspace = true } +thiserror = { workspace = true } +tokio = { workspace = true, features = ["sync"] } +tracing = { workspace = true } +wincode = { workspace = true } + +agave-transaction-view = { workspace = true } +solana-account = { workspace = true } +solana-instruction = { workspace = true } +solana-keypair = { workspace = true } +solana-program-runtime = { workspace = true } +solana-pubkey = { workspace = true } +solana-transaction = { workspace = true, features = ["wincode"] } + +[dev-dependencies] +keeper = { workspace = true, features = ["testkit"] } +magicblock-engine = { path = ".", features = ["testkit"] } +nucleus = { workspace = true, features = ["testkit"] } +v42-calculator-interface = { workspace = true, features = ["builder"] } + +solana-instruction-error = { workspace = true } +solana-signer = { workspace = true } +solana-sysvar = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "sync", "time"] } + +[lints] +workspace = true diff --git a/engine/README.md b/engine/README.md new file mode 100644 index 0000000..a18b8a1 --- /dev/null +++ b/engine/README.md @@ -0,0 +1,38 @@ +# `magicblock-engine` + +This crate exposes `Engine`, the consumer-facing handle over keeper state, +transaction sequencing, simulation, block pacing, recovery, and MagicRoot +account operations. It registers MagicRoot as a native builtin before keeper +opens startup state. + +`Engine::signer` is always the local keypair. `Engine::authority` returns the +configured remote authority for a replica, or the local identity when no +override is configured. Replication uses that distinction to sign locally while +authenticating its immediate upstream. + +## Startup and recovery + +Keeper restores an accountsdb snapshot when the active store is corrupt or its +sealed superblock trails the retained ledger. The latter also covers snapshots +staged by a replication follower. + +If accountsdb then trails the ledger tip, `Engine::new` replays retained entries +through a temporary sequencer. Replay quiesces at superblock seals and compares +the reconstructed checksum with the recorded seal. A mismatch returns +`ReplayError::StateMismatch`. Current state opens without replay. + +Internal pacing appends one reset marker at the upcoming slot and clears +chain-mirrored volatile accounts before the pacemaker task starts. Internal +system accounts remain available. Replicas use external pacing and retain +restored volatile state. + +## Shutdown + +Shutdown behavior follows the pacing source. Internal pacing publishes a final +block and flushes durable state. External pacing flushes the durable cursor +before writing `CURRENT/volatile.db`, allowing the next open and replication +handshake to resume from matching state. + +The embedding service retains the `ShutdownManager` passed to `Engine::new` and +calls `terminate` after stopping external ingress. The manager stops the +replication client, pacemaker, sequencer, and backing services in order. diff --git a/engine/src/accessor.rs b/engine/src/accessor.rs new file mode 100644 index 0000000..ec011a1 --- /dev/null +++ b/engine/src/accessor.rs @@ -0,0 +1,112 @@ +//! Account- and transaction-scoped operation facades. + +use std::{sync::atomic::Ordering, time::Duration}; + +use keeper::{ExecutionRecord, TransactionView}; +use magic_root_interface::MagicRootInstruction; +use processor::{SequencerMessage, Simulation, SimulatorMessage}; +use solana_account::{AccountFieldPatch, OwnedAccount}; +use solana_instruction::Instruction; +use solana_pubkey::Pubkey; +use solana_transaction::TransactionResult; +use tokio::time; + +use crate::{Engine, IntoTransactionView, error::Result}; + +/// Upper bound on awaiting a submitted transaction's committed result. +const EXECUTION_TIMEOUT: Duration = Duration::from_secs(8); + +/// Account-scoped operations bound to a single `pubkey`. +pub struct AccountAccessor<'a> { + pub(crate) pubkey: Pubkey, + pub(crate) engine: &'a Engine, +} + +/// Transaction-submission operations bound to an engine instance. +pub struct TransactionAccessor<'a> { + pub(crate) engine: &'a Engine, + pub(crate) transaction: TransactionView, +} + +impl AccountAccessor<'_> { + /// Creates the account by patching in every field and finalizing it, + /// optionally running follow-up `actions` once it is finalized. + pub async fn create( + &self, + account: OwnedAccount, + actions: Option>, + ) -> Result<()> { + let mut instructions = MagicRootInstruction::compose_account(self.pubkey, account)?; + if let Some(actions) = actions { + instructions.push(MagicRootInstruction::PostFinalize(actions).compose(self.pubkey)?); + } + self.execute(instructions).await + } + + /// Updates the account by patching in every field of `account` + pub async fn update(&self, account: OwnedAccount) -> Result<()> { + let instructions = MagicRootInstruction::compose_account(self.pubkey, account)?; + self.execute(instructions).await + } + + /// Applies the given field patches to the account. + pub async fn patch(&self, patches: Vec) -> Result<()> { + let mut instructions = Vec::with_capacity(patches.len()); + for patch in patches { + let ix = MagicRootInstruction::Patch(patch).compose(self.pubkey)?; + instructions.push(ix); + } + self.execute(instructions).await + } + + /// Closes the account. + pub async fn delete(&self) -> Result<()> { + let instructions = vec![MagicRootInstruction::Delete.compose(self.pubkey)?]; + self.execute(instructions).await + } + + /// Composes the instructions into a signed engine transaction, executes it, + /// and flattens the committed transaction result into the engine error type. + async fn execute(&self, instructions: Vec) -> Result<()> { + let txn = instructions.compose(self.engine)?; + self.engine.transaction(txn)?.execute().await?.map_err(Into::into) + } +} + +impl TransactionAccessor<'_> { + /// Submits `transaction` for execution and awaits its committed result. + pub async fn execute(self) -> Result> { + if self.engine.terminating.load(Ordering::Acquire) { + Err("engine is shutting down".to_string())?; + } + let signature = self.transaction.signatures()[0]; + let msg = SequencerMessage::Transaction(self.transaction); + let mut rx = self.engine.transactions().subscribe_signature(signature).await; + self.engine.sequencer.send(msg).await?; + let status = time::timeout(EXECUTION_TIMEOUT, rx.recv()) + .await + .map_err(|e| e.to_string())? + .map_err(|e| e.to_string())?; + Ok(status.result) + } + + /// Submits `transaction` for execution without awaiting its result. + pub async fn schedule(self) -> Result<()> { + if self.engine.terminating.load(Ordering::Acquire) { + Err("engine is shutting down".to_string())?; + } + let msg = SequencerMessage::Transaction(self.transaction); + self.engine.sequencer.send(msg).await.map_err(Into::into) + } + + /// Simulates `transaction` against current state without committing it. + pub async fn simulate(self) -> Result> { + let (response, rx) = oneshot::channel(); + let msg = SimulatorMessage::Transaction(Simulation { + transaction: self.transaction, + response, + }); + self.engine.sequencer.simulation.send(msg).await?; + rx.await.map_err(Into::into) + } +} diff --git a/engine/src/error.rs b/engine/src/error.rs new file mode 100644 index 0000000..90b3a56 --- /dev/null +++ b/engine/src/error.rs @@ -0,0 +1,81 @@ +//! Engine error types. + +use agave_transaction_view::result::TransactionViewError; +use derive_more::From; +use keeper::error::KeeperError; +use ledger::{LedgerError, LedgerRequestError}; +use nucleus::shutdown::Service; +use processor::ProcessorError; +use solana_transaction::{InstructionError, SignerError, TransactionError}; +use tokio::sync::mpsc::error::SendError; + +/// Result type used by engine APIs. +pub type Result = std::result::Result; + +/// Failures surfaced by the top-level engine. +#[derive(From, thiserror::Error, Debug)] +pub enum EngineError { + /// A durable-state (keeper) operation failed. + #[error("state error: {0}")] + State(#[source] KeeperError), + /// Scheduling or executing a transaction failed. + #[error("processor error: {0}")] + Processor(#[source] ProcessorError), + /// Replaying the ledger into volatile state on startup failed. + #[error("replay error: {0}")] + Replay(#[source] ReplayError), + /// A background service is no longer reachable. + #[error("service became unavailable: {0:?}")] + ServiceUnavailable(Service), + /// Signing a transaction with the engine authority failed. + #[error("signature error: {0}")] + Signature(#[source] SignerError), + /// Serializing or deserializing a transaction failed. + #[error("serialization error: {0}")] + Serde(#[source] wincode::Error), + /// Sanitizing a serialized transaction into a transaction view failed. + #[error("transaction sanitization: {0:?}")] + Sanitization(TransactionViewError), + /// A submitted transaction carried an invalid signature. + #[error("transaction signature verification failed")] + SignatureVerification, + /// A submitted transaction was committed with an execution failure. + #[error("transaction execution failed: {0}")] + TransactionExecution(#[source] TransactionError), + /// An unexpected internal failure carrying a contextual message. + #[error("internal error: {0}")] + Internal(String), +} + +impl From> for EngineError { + fn from(_: SendError) -> Self { + Self::ServiceUnavailable(Service::Sequencer) + } +} +impl From for EngineError { + fn from(error: InstructionError) -> Self { + Self::TransactionExecution(TransactionError::InstructionError(0, error)) + } +} +impl From for EngineError { + fn from(_: oneshot::RecvError) -> Self { + Self::ServiceUnavailable(Service::Sequencer) + } +} + +/// Failures raised while replaying retained ledger entries on startup. +#[derive(From, thiserror::Error, Debug)] +pub enum ReplayError { + /// A retained transaction could not be sanitized into a transaction view. + #[error("transaction sanitization: {0:?}")] + Sanitization(TransactionViewError), + /// The replayed account state checksum diverged from the sealed superblock. + #[error("replayed state checksum mismatch")] + StateMismatch, + /// Waiting for the ledger reader's replay response failed. + #[error("ledger replay request failed: {0}")] + Request(LedgerRequestError), + /// Reading or decoding retained ledger entries failed. + #[error("ledger replay failed: {0}")] + Ledger(LedgerError), +} diff --git a/engine/src/lib.rs b/engine/src/lib.rs new file mode 100644 index 0000000..a7b3e35 --- /dev/null +++ b/engine/src/lib.rs @@ -0,0 +1,191 @@ +#![doc = include_str!("../README.md")] + +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Instant, +}; + +use derive_more::Deref; +use keeper::{Keeper, builder::KeeperBuilder, error::KeeperError}; +use ledger::schema::OwnedBlockestoreEntry; +use magic_root_program::entrypoint::MagicRootEntrypoint; +use nucleus::{ + runtime::{self, BarrierHandle, SequencerHandle}, + shutdown::{Service, ShutdownManager, ShutdownReason}, +}; +use processor::{SequencerMessage, sequencer::Sequencer}; +use solana_program_runtime::{ + loaded_programs::{ProgramCache, ProgramCacheEntry}, + solana_sbpf::program::BuiltinFunctionDefinition, +}; +use solana_pubkey::Pubkey; +use tracing::{error, info}; + +mod accessor; +mod error; +pub mod pacemaker; +mod transaction; + +#[cfg(feature = "testkit")] +pub mod testkit; + +pub use accessor::{AccountAccessor, TransactionAccessor}; +pub use error::{EngineError, ReplayError, Result}; +pub use transaction::IntoTransactionView; + +use crate::pacemaker::{ExternalPacer, PaceMaker}; + +/// Top-level engine handle: owns the durable state and the sequencer submission +/// channels. +#[derive(Deref, Clone)] +pub struct Engine { + /// Durable engine state (accountsdb + ledger), shared across components. + #[deref] + state: Arc, + /// Submission handle into the sequencer's execution and simulation channels. + sequencer: SequencerHandle, + /// Rejects new transactions once coordinated shutdown begins. + terminating: Arc, +} + +impl Engine { + /// Builds and starts the engine. + /// + /// Opens durable state through the keeper builder (coming up on persisted + /// state), replays retained ledger entries to rebuild volatile state only when + /// recovering from a rewound accountsdb, starts the live sequencer, and spawns + /// the pacemaker using the builder's blockstore timing. + pub async fn new( + mut builder: KeeperBuilder, + pacer: Option, + shutdown: &mut ShutdownManager, + ) -> Result { + let cache = Arc::new(ProgramCache::default()); + let cpus = (num_cpus::get().saturating_sub(2)).max(2); + builder.builtins.insert( + magic_root_interface::ID, + (MagicRootEntrypoint::vm, MagicRootEntrypoint::codegen), + ); + for (&id, builtin) in &builder.builtins { + let entry = ProgramCacheEntry::new_builtin(*builtin); + cache.assign_program(id, entry.into()); + } + let blockstore = builder.blockstore; + let state = Arc::new(builder.build(shutdown).await?); + Self::try_replay(&state, &cache, cpus).await?; + let (service, sequencer) = Sequencer::new(cpus / 2, state.clone(), cache, shutdown, false)?; + service.spawn()?; + let terminating = Arc::new(AtomicBool::new(false)); + let engine = Self { state, sequencer, terminating }; + PaceMaker::spawn(engine.clone(), pacer, blockstore, shutdown)?; + info!(authority = %engine.authority(), cpus, "engine started"); + Ok(engine) + } + + /// Quiesces execution and closes durable state. + /// + /// `dump` serializes chain-mirrored state for an externally paced replica to + /// restore on its next open. Internally paced leaders clear that state once + /// during startup instead and only flush durable state here. + pub async fn shutdown(&self, dump: bool) -> Result<()> { + info!(dump, "shutting down the engine"); + self.terminating.store(true, Ordering::Release); + let _guard = self.barrier().await?; + if dump { + self.accounts().dump(None).map_err(KeeperError::from)?; + } + self.flush().map_err(Into::into) + } + + /// Returns an accessor for mutating the account at `pubkey`. + pub fn account(&self, pubkey: Pubkey) -> AccountAccessor<'_> { + AccountAccessor { engine: self, pubkey } + } + + /// Returns an accessor for signing and submitting transactions. + pub fn transaction(&self, transaction: T) -> Result> + where + T: IntoTransactionView, + { + let transaction = transaction.compose(self)?; + Ok(TransactionAccessor { engine: self, transaction }) + } + + /// Drains in-flight execution and keeps the sequencer paused until the handle is dropped. + pub async fn barrier(&self) -> Result { + let (controller, guard) = runtime::barrier(); + self.sequencer.send(SequencerMessage::Barrier(guard)).await?; + controller.acknowledged.await?; + Ok(controller.released) + } + + /// Applies one retained ledger entry through the engine's ordered paths. + /// + /// Seal and reset entries quiesce execution before touching shared state; + /// a reconstructed seal whose checksum differs returns + /// [`ReplayError::StateMismatch`]. + pub async fn replay(&self, entry: OwnedBlockestoreEntry) -> Result<()> { + match entry { + OwnedBlockestoreEntry::Transaction(txn) => self.transaction(txn)?.schedule().await?, + OwnedBlockestoreEntry::Block(block) => { + self.sequencer.send(SequencerMessage::Block(block)).await?; + } + OwnedBlockestoreEntry::Superblock(expected) => { + let _guard = self.barrier().await?; + self.accounts().set_superblock(expected.id); + self.flush()?; + let observed = self.superblocks().sealed(); + if observed != expected { + error!(?observed, ?expected, "state mismatch; aborting replay"); + Err(ReplayError::StateMismatch)?; + } + } + OwnedBlockestoreEntry::Reset(slot) => { + let _guard = self.barrier().await?; + self.reset(slot)?; + } + }; + Ok(()) + } + + /// Rebuilds state through a temporary replay sequencer when accountsdb trails + /// the retained ledger, then stops every temporary service before returning. + async fn try_replay(state: &Arc, cache: &Arc, cpus: usize) -> Result<()> { + let timer = Instant::now(); + let Some(mut replayer) = state.replay().await? else { + return Ok(()); + }; + let mut shutdown = ShutdownManager::default(); + let mut sh = shutdown.handle(Service::LedgerReplayer); + let (service, sequencer) = + Sequencer::new(cpus, state.clone(), cache.clone(), &mut shutdown, true)?; + service.spawn()?; + let engine = Self { + state: state.clone(), + sequencer, + terminating: Default::default(), + }; + while let Some(entry) = replayer.rx.recv().await { + engine.replay(entry).await?; + } + replayer + .response + .recv_timeout() + .await + .map_err(ReplayError::from)? + .map_err(ReplayError::from)?; + + drop(engine.barrier().await?); + engine.flush()?; + + let slot = state.blocks().latest().slot; + info!(slot, duration = ?timer.elapsed(), "ledger replay complete"); + drop(engine); + sh.terminate(ShutdownReason::Signalled); + shutdown.terminate().await; + Ok(()) + } +} diff --git a/engine/src/pacemaker.rs b/engine/src/pacemaker.rs new file mode 100644 index 0000000..575e3c7 --- /dev/null +++ b/engine/src/pacemaker.rs @@ -0,0 +1,197 @@ +//! Block-boundary pacing. + +use std::{num::NonZeroU64, time::Duration}; + +use derive_more::Deref; +use keeper::builder::BlockstoreParams; +use ledger::schema::Block; +use nucleus::{ + Slot, + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, + unix_time, +}; +use processor::{SequencerMessage, SimulatorMessage}; +use tokio::{ + sync::mpsc::Receiver, + time::{self, Interval, MissedTickBehavior}, +}; +use tracing::error; + +use crate::{Engine, Result}; + +/// Channel used by external block producers. +pub type ExternalPacer = Receiver; + +/// Emits block boundaries into engine execution paths. +#[derive(Deref)] +pub struct PaceMaker { + /// Engine handle used to submit each boundary. + #[deref] + engine: Engine, + /// Source for the next block boundary. + pacer: Pacer, + /// Number of slots sealed into each superblock. + superblock: NonZeroU64, +} + +/// Source of block boundaries. +pub enum Pacer { + /// Interval-driven slot production. + Internal(BlockTicker), + /// Externally supplied block boundaries. + External(ExternalPacer), +} + +/// State for interval-driven slot production. +pub struct BlockTicker { + /// Next slot to emit. + slot: Slot, + /// Block production interval. + ticker: Interval, +} + +impl BlockTicker { + /// Builds an interval ticker starting at the engine's current slot. + pub(crate) fn new(engine: &Engine, blocktime: Duration) -> Self { + let slot = engine.blocks().current_slot(); + let mut ticker = time::interval(blocktime); + ticker.set_missed_tick_behavior(MissedTickBehavior::Skip); + ticker.reset(); + BlockTicker { slot, ticker } + } + + /// Returns the next block boundary and advances the slot cursor. + pub(crate) fn block(&mut self) -> Block { + let block = Block { + slot: self.slot, + hash: Default::default(), // will be set from the sequencer + time: unix_time().as_secs() as i64, + }; + self.slot += 1; + block + } +} + +/// Block boundary submitted by an external producer. +pub struct ExternalBlock { + /// Boundary to enqueue. + pub block: Block, + /// Notified after the pacemaker handles the boundary locally. + /// + /// On ordinary slots this means the boundary was queued and the keeper slot + /// was advanced. On superblock slots it also includes the synchronous seal. + pub submitted: oneshot::Sender<()>, +} + +impl ExternalBlock { + /// Pairs a boundary with the receiver signalled once the pacemaker has locally + /// handled it, letting the submitter await ordered application. + pub fn new(block: Block) -> (Self, oneshot::Receiver<()>) { + let (submitted, guard) = oneshot::channel(); + let block = Self { block, submitted }; + (block, guard) + } +} + +impl PaceMaker { + /// Registers and starts the pacemaker task. + /// + /// Uses an external block source when supplied. Otherwise it records one + /// reset at the keeper's current slot, clears chain-mirrored volatile state, + /// and starts emitting slots on the configured block interval. + pub fn spawn( + engine: Engine, + pacer: Option, + blockstore: BlockstoreParams, + shutdown: &mut ShutdownManager, + ) -> Result<()> { + let pacer = match pacer { + Some(rx) => Pacer::External(rx), + None => { + let ticker = BlockTicker::new(&engine, blockstore.blocktime); + engine.reset(ticker.slot)?; + Pacer::Internal(ticker) + } + }; + let shutdown = shutdown.handle(Service::PaceMaker); + let superblock = blockstore.superblock; + let pacemaker = Self { engine, pacer, superblock }; + tokio::spawn(pacemaker.run(shutdown)); + Ok(()) + } + + /// Paces block boundaries until shutdown or the block source is exhausted. + /// + /// Shutdown follows the pacing mode. Internal pacing publishes one last + /// block and flushes durable state. External pacing also checkpoints + /// volatile state alongside its durable cursor for the next upstream + /// handshake. + async fn run(mut self, mut shutdown: ShutdownHandle) { + let mut res = loop { + tokio::select! { + biased; + _ = shutdown.signalled() => { + break Ok(()); + } + success = self.pace() => { + if !*success.as_ref().unwrap_or(&false) { + break success.map(|_| ()); + } + } + } + }; + res = if let Pacer::Internal(ref mut t) = self.pacer { + let b = t.block(); + res.and(self.handle(b).await).and(self.shutdown(false).await) + } else { + res.and(self.shutdown(true).await) + }; + if let Err(error) = res { + error!(?error, "pace maker terminated with critical failure"); + shutdown.terminate(ShutdownReason::Error(error.into())); + } else { + shutdown.terminate(ShutdownReason::Signalled); + } + } + + /// Emits one block boundary and seals a superblock when the emitted slot is + /// on the configured interval. + async fn pace(&mut self) -> Result { + let (block, submission) = match &mut self.pacer { + Pacer::Internal(t) => { + t.ticker.tick().await; + (t.block(), None) + } + Pacer::External(rx) => { + let Some(msg) = rx.recv().await else { + return Ok(false); + }; + (msg.block, Some(msg.submitted)) + } + }; + let result = self.handle(block).await.map(|()| true); + if let Some(submission) = submission + && result.is_ok() + { + let _ = submission.send(()); + } + result + } + + /// Advances the execution and simulation environments to `block`, sealing a + /// superblock when the slot lands on the configured interval. + /// + /// The seal is taken behind a barrier and runs synchronously: it exports an + /// accountsdb snapshot, which is only coherent while no store operation can + /// race it. Holding the boundary here is what buys that exclusivity, at the + /// cost of stalling block production until the seal completes. + async fn handle(&self, block: Block) -> Result<()> { + self.sequencer.send(SequencerMessage::Block(block)).await?; + self.sequencer.simulation.send(SimulatorMessage::Block(block)).await?; + if block.slot.is_multiple_of(self.superblock.get()) { + let _guard = self.barrier().await?; + self.finalize_superblock()?; + } + Ok(()) + } +} diff --git a/engine/src/testkit.rs b/engine/src/testkit.rs new file mode 100644 index 0000000..b957e06 --- /dev/null +++ b/engine/src/testkit.rs @@ -0,0 +1,188 @@ +//! Shared black-box harness for engine-backed integration suites. +//! +//! Builds a real [`Engine`] over [`keeper::testkit`] directories with internal or +//! externally controlled pacing. Compiled only under the `testkit` feature. +#![allow(clippy::expect_used, clippy::unwrap_used)] + +use std::{path::PathBuf, sync::Arc, time::Duration}; + +use derive_more::Deref; +use keeper::{ + ExecutionRecord, + builder::{Authority, KeeperBuilder}, + testkit::{Dirs, SUPERBLOCK, block, keeper_builder}, +}; +use nucleus::{Slot, ledger::BlockstorePosition, shutdown::ShutdownManager}; +use solana_account::AccountSharedData; +use solana_instruction::Instruction; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_transaction::TransactionResult; +use tokio::sync::mpsc; + +use crate::{Engine, pacemaker::ExternalBlock}; + +const TIMEOUT: Duration = Duration::from_secs(4); + +/// Block pacing for a [`TestEngine`]. +pub enum Pacing { + /// The test supplies blocks through [`TestEngine::pacer`]. + External, + /// The engine runs its own pacemaker. + Internal, +} + +/// A running engine plus its deterministic pacing and lifecycle handles. +#[derive(Deref)] +pub struct TestEngine { + #[deref] + engine: Engine, + shutdown: ShutdownManager, + authority: Authority, + dirs: Dirs, + pacer: Option>, + slot: Slot, +} + +impl TestEngine { + /// Starts the standard test engine on fresh directories. + pub async fn new() -> Self { + Self::with(Dirs::default(), Arc::new(Keypair::new())).await + } + + /// Starts the standard test engine over `dirs` with `authority`. + pub async fn with(dirs: Dirs, authority: impl Into) -> Self { + Self::try_with(dirs, authority).await.unwrap() + } + + /// Fallible [`Self::with`], used when startup failure is the assertion. + pub async fn try_with(dirs: Dirs, authority: impl Into) -> crate::Result { + let mut builder = keeper_builder(&dirs); + builder.authority = authority.into(); + Self::try_from_builder(dirs, builder, Pacing::External).await + } + + /// Starts an engine from a caller-configured keeper builder. + /// + /// `dirs` must own the directories referenced by `builder` and outlive the + /// resulting engine. + pub async fn from_builder(dirs: Dirs, builder: KeeperBuilder, pacing: Pacing) -> Self { + Self::try_from_builder(dirs, builder, pacing).await.unwrap() + } + + /// Fallible [`Self::from_builder`]. + pub async fn try_from_builder( + dirs: Dirs, + builder: KeeperBuilder, + pacing: Pacing, + ) -> crate::Result { + let authority = builder.authority.clone(); + let (pacer, rx) = match pacing { + Pacing::External => { + let (tx, rx) = mpsc::channel(64); + (Some(tx), Some(rx)) + } + Pacing::Internal => (None, None), + }; + let mut shutdown = ShutdownManager::default(); + let engine = Engine::new(builder, rx, &mut shutdown).await?; + let slot = engine.blocks().current_slot(); + Ok(Self { + engine, + shutdown, + authority, + dirs, + pacer, + slot, + }) + } + + /// Cloneable external pacemaker sender for services under test. + /// + /// # Panics + /// + /// Panics if the engine is internally paced. + pub fn pacer(&self) -> mpsc::Sender { + self.pacer.clone().expect("engine is externally paced") + } + + /// Mutable lifecycle manager used to register or await test services. + pub fn shutdown(&mut self) -> &mut ShutdownManager { + &mut self.shutdown + } + + /// Drains engine work, flushes queued ledger appends, and returns the durable cursor. + pub async fn sync(&self) -> BlockstorePosition { + drop(self.barrier().await.unwrap()); + self.superblocks().sync().unwrap(); + self.superblocks().position() + } + + /// Full committed account, or `None` when absent/closed. + pub fn account(&self, key: Pubkey) -> Option { + self.engine.accounts().get(&key).unwrap() + } + + /// Executes instructions and returns the committed transaction result. + pub async fn execute(&self, ixs: &[Instruction]) -> TransactionResult<()> { + self.transaction(ixs).unwrap().execute().await.unwrap() + } + + /// Simulates instructions without committing them. + pub async fn simulate(&self, ixs: &[Instruction]) -> TransactionResult { + self.transaction(ixs).unwrap().simulate().await.unwrap() + } + + /// Schedules instructions without awaiting commit. + pub async fn schedule(&self, ixs: &[Instruction]) { + self.transaction(ixs).unwrap().schedule().await.unwrap(); + } + + /// Advances `n` block boundaries. + /// + /// # Panics + /// + /// Panics if the engine is internally paced. + pub async fn advance(&mut self, n: u64) { + for _ in 0..n { + let (block, submitted) = ExternalBlock::new(block(self.slot)); + self.pacer().send(block).await.unwrap(); + submitted.recv_timeout(TIMEOUT).expect("pacemaker accepts the block in time"); + self.slot += 1; + } + } + + /// Advances through the next superblock boundary. + pub async fn seal_superblock(&mut self) -> Slot { + let boundary = self.slot.next_multiple_of(SUPERBLOCK.into()); + while self.slot <= boundary { + self.advance(1).await; + } + boundary + } + + /// Seals the next superblock and waits for its snapshot archive. + pub async fn seal_and_archive(&mut self) -> PathBuf { + self.seal_superblock().await; + self.await_archive().await + } + + /// Directory paths used by this engine. + pub fn dirs(&self) -> &Dirs { + &self.dirs + } + + /// Stops every service and returns the directories and authority for reopen. + pub async fn close(self) -> (Dirs, Authority) { + let Self { + mut shutdown, + dirs, + authority, + engine, + .. + } = self; + drop(engine); + shutdown.terminate().await; + (dirs, authority) + } +} diff --git a/engine/src/transaction.rs b/engine/src/transaction.rs new file mode 100644 index 0000000..27ff2da --- /dev/null +++ b/engine/src/transaction.rs @@ -0,0 +1,66 @@ +//! Composing values into sanitized transaction views. + +use keeper::TransactionView; +use solana_instruction::Instruction; +use solana_transaction::{Message, Transaction}; + +use crate::{Engine, error::EngineError, error::Result}; + +/// Conversion of anything composable into an executable +/// transaction into a sanitized [`TransactionView`]. +pub trait IntoTransactionView { + /// Composes `self` into a sanitized [`TransactionView`], signing with + /// `engine`'s authority and latest blockhash where applicable. + fn compose(self, engine: &Engine) -> Result; +} + +impl IntoTransactionView for Message { + fn compose(self, engine: &Engine) -> Result { + let mut transaction = Transaction::new_unsigned(self); + transaction.try_sign(&[&engine.state.signer()], engine.blockhash())?; + let data = wincode::serialize(&transaction).map_err(wincode::Error::from)?; + data.compose(engine) + } +} + +impl IntoTransactionView for &[Instruction] { + fn compose(self, engine: &Engine) -> Result { + let msg = Message::new(self, Some(&engine.authority())); + msg.compose(engine) + } +} + +impl IntoTransactionView for Vec { + fn compose(self, engine: &Engine) -> Result { + TransactionView::try_new_sanitized(self.into(), true)?.compose(engine) + } +} + +impl IntoTransactionView for TransactionView { + fn compose(self, _: &Engine) -> Result { + sigverify(&self)?; + Ok(self) + } +} + +/// The engine's sole signature-verification point. +/// +/// Execution is trustless: every submission funnels through the [`TransactionView`] +/// `compose` and is verified here, including replay and replication of already-committed +/// transactions — no path reaches the sequencer unverified. Downstream code may therefore +/// assume the fee payer and every required signer actually signed +fn sigverify(view: &TransactionView) -> Result<()> { + // Sanitization guarantees `signatures().len() == num_required_signatures` and at + // least that many static keys, so the leading pairs are indexed without checks. + let (keys, sigs, msg) = ( + view.static_account_keys(), + view.signatures(), + view.message_data(), + ); + for i in 0..view.num_required_signatures() as usize { + if !sigs[i].verify(keys[i].as_ref(), msg) { + return Err(EngineError::SignatureVerification); + } + } + Ok(()) +} diff --git a/engine/tests/accounts.rs b/engine/tests/accounts.rs new file mode 100644 index 0000000..2ebe10b --- /dev/null +++ b/engine/tests/accounts.rs @@ -0,0 +1,147 @@ +//! Account CRUD through the MagicRoot builtin — the privileged mutation path +//! exposed by `AccountAccessor`. This path is untested below the engine: it needs +//! the always-on MagicRoot builtin plus the executor's per-thread authority +//! (MagicRoot authorizes the transaction's fee payer against it). Asserts the +//! create/patch/update/delete round-trip and the sponsor-balance invariant, and +//! that post-finalize actions actually run. +#![cfg(test)] + +use engine::{Engine, testkit::TestEngine}; +use keeper::testkit::v42_owned; +use solana_account::{ + AccountBuilder, AccountFieldPatch, AccountMode, OwnedAccount, ReadableAccount, +}; +use solana_pubkey::Pubkey; +use v42_calculator_interface::builder::Expr as E; + +/// Rent-exempt for the data sizes used below; the SVM rejects a created account +/// that falls under the rent floor. +const LAMPORTS: u64 = 2_000_000; + +/// Delegated account with `data`, funded at the shared rent-exempt balance. +fn delegated(owner: Pubkey, data: Vec) -> OwnedAccount { + AccountBuilder::default() + .lamports(LAMPORTS) + .owner(owner) + .mode(AccountMode::Delegated) + .data(data) + .build() +} + +// The full lifecycle. `create` materializes a fresh account by patching every +// field then finalizing it (finalize drains the created balance from the +// authority, conserving supply); `patch` mutates individual fields in place; +// `update` overwrites an existing account or materializes a fresh key; and +// `delete` closes it. `patch`/`update` on an already-funded account must not +// change lamports — finalize would re-charge the authority for the delta and +// unbalance the transaction — so mutations here keep the balance constant. +#[tokio::test(flavor = "multi_thread")] +async fn account_crud_lifecycle() { + let te = TestEngine::new().await; + let engine: &Engine = &te; + let key = Pubkey::new_unique(); + let owner = Pubkey::new_unique(); + + let created = delegated(owner, vec![1, 2, 3, 4, 5, 6, 7, 8]); + let authority_before = te.account(te.authority()).expect("sponsor exists").lamports(); + + engine.account(key).create(created, None).await.unwrap(); + + let acc = te.account(key).expect("created account exists"); + assert_eq!(acc.lamports(), LAMPORTS); + assert_eq!(acc.owner(), &owner); + assert_eq!(acc.data(), &[1, 2, 3, 4, 5, 6, 7, 8]); + assert!(acc.mutable(), "a delegated account is ER-controlled"); + + let authority_after = te.account(te.authority()).expect("sponsor exists").lamports(); + assert_eq!( + authority_before - authority_after, + LAMPORTS, + "finalize sponsors the created balance from the authority" + ); + + // patch the data, growing the buffer (exercises the borrowed-image + // copy-on-write promotion), leaving other fields untouched. + engine + .account(key) + .patch(vec![AccountFieldPatch::DataAt { + offset: 0, + data: vec![9; 16], + }]) + .await + .unwrap(); + let acc = te.account(key).expect("still exists"); + assert_eq!(acc.data(), &[9; 16]); + assert_eq!(acc.lamports(), LAMPORTS, "patch leaves lamports untouched"); + + // patch a different field (owner). + let new_owner = Pubkey::new_unique(); + engine + .account(key) + .patch(vec![AccountFieldPatch::Owner(new_owner)]) + .await + .unwrap(); + assert_eq!(te.account(key).expect("still exists").owner(), &new_owner); + + // update overwrites the existing account in place: same-length data (the + // patch sequence extends but never truncates) and identical lamports, so + // finalize's delta charge is zero and the transaction stays balanced. The + // owner reverting to `owner` proves the earlier patch was overwritten. + engine.account(key).update(delegated(owner, vec![5; 16])).await.unwrap(); + let acc = te.account(key).expect("still exists"); + assert_eq!(acc.data(), &[5; 16], "update replaced the data wholesale"); + assert_eq!(acc.owner(), &owner, "update replaced the patched owner"); + + // delete: the account is gone from storage. + engine.account(key).delete().await.unwrap(); + assert!(te.account(key).is_none(), "deleted account is removed"); + + // update also materializes a fresh account the same way create does, minus + // the post-finalize actions. + let key2 = Pubkey::new_unique(); + engine.account(key2).update(delegated(owner, vec![3; 8])).await.unwrap(); + assert_eq!(te.account(key2).expect("materialized").data(), &[3; 8]); + + te.close().await; +} + +// Post-finalize actions are invoked via CPI after the account is finalized, so a +// failing action aborts the whole creation (nothing commits), while a benign one +// lets it through. The contrast proves the actions actually execute rather than +// being silently dropped. +#[tokio::test(flavor = "multi_thread")] +async fn create_runs_post_finalize_actions() { + let te = TestEngine::new().await; + let engine: &Engine = &te; + + // A benign v42 action (at CPI depth it only emits return data) succeeds, so + // creation completes. + let ok_key = Pubkey::new_unique(); + let benign = E::lit(1).compose(ok_key, &[]); + engine + .account(ok_key) + .create(v42_owned(0, AccountMode::Delegated), Some(vec![benign])) + .await + .expect("create with a succeeding post-finalize action"); + assert!( + te.account(ok_key).is_some(), + "account created once its action ran" + ); + + // An underflowing v42 action errors; the failure propagates and rolls back + // the account creation in the same transaction. + let bad_key = Pubkey::new_unique(); + let failing = (E::lit(1) - E::lit(2)).compose(bad_key, &[]); + let acc = v42_owned(0, AccountMode::Delegated); + let result = engine.account(bad_key).create(acc, Some(vec![failing])).await; + assert!( + result.is_err(), + "failing post-finalize action surfaces an error" + ); + assert!( + te.account(bad_key).is_none(), + "nothing commits when the action fails" + ); + + te.close().await; +} diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs new file mode 100644 index 0000000..c6ac10d --- /dev/null +++ b/engine/tests/recovery.rs @@ -0,0 +1,136 @@ +//! Full-engine replay recovery — the engine's most distinctive orchestration. +//! After an accountsdb corruption the keeper restores an older archived snapshot, +//! leaving durable state behind the ledger tip; the engine then spins a temporary +//! replay sequencer to re-execute the retained ledger entries and rebuild the +//! missing state, checksum-verified at each sealed superblock. Nothing below the +//! engine wires this end to end. Covered here: the healthy restart that must not +//! recover, the replay that crosses a sealed checksum and succeeds, and the +//! replay that diverges from one and must refuse to start. +#![cfg(test)] + +use std::{path::PathBuf, time::Duration}; + +use engine::{EngineError, ReplayError, testkit::TestEngine}; +use keeper::testkit::corrupt; +use nucleus::ledger::ACCOUNTSDB_SNAPSHOT_FILE; +use solana_account::AccountMode; +use solana_pubkey::Pubkey; +use tokio::time; +use v42_calculator_interface::builder::Expr as E; + +/// Commits `K = value` through a full transaction and seals the following +/// superblock, returning its archived snapshot path. +async fn commit_and_seal(te: &mut TestEngine, key: Pubkey, value: u64) -> PathBuf { + te.execute(&[E::lit(value).compose(key, &[])]).await.unwrap(); + te.seal_and_archive().await +} + +// Replay must rebuild everything between the restored snapshot and the ledger +// tip: dropping superblock 2's archive forces the restore back onto snapshot 1, +// so re-executing B crosses superblock 2's sealed checksum (the verification +// arm's happy path) before C is rebuilt from the unsealed head. +#[tokio::test(flavor = "multi_thread")] +async fn replay_rebuilds_state_after_corruption() { + let mut te = TestEngine::new().await; + let key = te.store_v42(0, AccountMode::Delegated); + + // A: K = 10 sealed into superblock 1, whose snapshot the restore lands on. + let s1 = commit_and_seal(&mut te, key, 10).await; + assert!(s1.exists(), "archived accountsdb snapshot exists on disk"); + assert!( + s1.ends_with(ACCOUNTSDB_SNAPSHOT_FILE), + "archive is the compressed accountsdb tarball" + ); + // B: K = 20 sealed into superblock 2; C: K = 30 lives only in the ledger's + // unsealed head, past every archived snapshot. + let s2 = commit_and_seal(&mut te, key, 20).await; + te.execute(&[E::lit(30).compose(key, &[])]).await.expect("C commits"); + te.advance(2).await; + let (dirs, authority) = te.close().await; + + // Corrupt the closed accountsdb and drop the newest archive: the reopen falls + // back to snapshot 1 (K = 10) and replays everything after it. Corruption must + // follow the close, whose flush would otherwise republish a valid checksum. + corrupt(dirs.accounts.path()); + std::fs::remove_file(&s2).unwrap(); + + // Guarded by a timeout so a replay regression fails fast instead of hanging. + let te2 = time::timeout(Duration::from_secs(10), TestEngine::with(dirs, authority)) + .await + .expect("engine reopens and finishes replay"); + assert_eq!( + te2.load_v42_data(key), + Some(30), + "both post-snapshot mutations were rebuilt purely from ledger replay" + ); + // The temporary replay sequencer must hand off to a working live one. + te2.execute(&[E::lit(1).compose(key, &[])]) + .await + .expect("engine is live after replay"); + + te2.close().await; +} + +// A mutation that bypasses the ledger is sealed into superblock 2's checksum but +// can never be rebuilt by replay, so the reopen must refuse to come up with +// `StateMismatch` rather than run on quietly diverged state. +#[tokio::test(flavor = "multi_thread")] +async fn replay_aborts_on_checksum_mismatch() { + let mut te = TestEngine::new().await; + let key = te.store_v42(0, AccountMode::Delegated); + commit_and_seal(&mut te, key, 10).await; + // Direct store: lands in persisted state (and superblock 2's checksum) + // without a ledger entry. + te.store_v42(7, AccountMode::Delegated); + let s2 = commit_and_seal(&mut te, key, 20).await; + let (dirs, authority) = te.close().await; + + corrupt(dirs.accounts.path()); + std::fs::remove_file(&s2).unwrap(); + + let result = time::timeout( + Duration::from_secs(10), + TestEngine::try_with(dirs, authority), + ) + .await + .expect("replay aborts in time"); + let error = result.err().expect("diverged checksum refuses startup"); + assert!( + matches!(error, EngineError::Replay(ReplayError::StateMismatch)), + "unexpected startup error: {error:?}" + ); +} + +// A healthy restart opens persisted state as-is and restores the clean-shutdown +// volatile dump. The direct-stored delegated account exists in neither snapshots +// nor ledger, while the read-only account exists only in the volatile dump; the +// post-seal transaction write pins the persisted tip alongside both probes. +#[tokio::test(flavor = "multi_thread")] +async fn clean_restart_reopens_persisted_and_volatile_state() { + let mut te = TestEngine::new().await; + let key = te.store_v42(0, AccountMode::Delegated); + commit_and_seal(&mut te, key, 10).await; + te.execute(&[E::lit(20).compose(key, &[])]).await.unwrap(); + let direct = te.store_v42(7, AccountMode::Delegated); + let volatile = te.store_v42(8, AccountMode::ReadOnly); + let (dirs, authority) = te.close().await; + + let te2 = TestEngine::with(dirs, authority).await; + assert_eq!( + te2.load_v42_data(key), + Some(20), + "persisted tip state reopened as-is" + ); + assert_eq!( + te2.load_v42_data(direct), + Some(7), + "ledger-invisible account intact, so no snapshot was restored" + ); + assert_eq!( + te2.load_v42_data(volatile), + Some(8), + "clean shutdown restores volatile state" + ); + + te2.close().await; +} diff --git a/engine/tests/security.rs b/engine/tests/security.rs new file mode 100644 index 0000000..6e96afa --- /dev/null +++ b/engine/tests/security.rs @@ -0,0 +1,126 @@ +//! Account-mutability enforcement at the engine boundary. +//! +//! The SVM lets a program write any account it owns; the engine's post-execution +//! guard (`validate_access`) is what rejects writes to accounts that are not in a +//! mutable mode, unless the whole transaction is privileged (every instruction +//! targets MagicRoot). These black-box tests drive the full engine and assert both +//! that the rejection surfaces the right error and that the illegal write never +//! commits. A second enforcement path — MagicRoot's own `post_finalize` check — +//! is covered by `post_finalize_immutable_action_is_rejected`. +#![cfg(test)] + +use engine::{Engine, testkit::TestEngine}; +use keeper::testkit::v42_owned; +use magic_root_interface::MagicRootInstruction; +use solana_account::{AccountFieldPatch, AccountMode}; +use solana_instruction::Instruction; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_signer::Signer; +use solana_transaction::TransactionError; +use v42_calculator_interface::builder::Expr as E; + +// The SVM permits the v42 program to write accounts it owns, but the guard +// rejects the commit and discards the mutation whenever the account is immutable +// and the transaction is not privileged. Two branches: a writable operand yields +// InvalidWritableAccount, the fee payer itself yields InvalidAccountForFee. A +// delegated (mutable) account is the positive control. +#[tokio::test(flavor = "multi_thread")] +async fn immutable_writes_are_rejected_and_not_committed() { + let te = TestEngine::new().await; + + // A writable, non-payer immutable operand: `compose` marks it writable, so + // execution dirties it before the guard runs. + let operand = te.store_v42(5, AccountMode::ReadOnly); + assert_eq!( + te.execute(&[E::lit(9).compose(operand, &[])]).await, + Err(TransactionError::InvalidWritableAccount) + ); + assert_eq!( + te.load_v42_data(operand), + Some(5), + "writable-account write discarded" + ); + + // The immutable account is the fee payer itself. The harness `execute` always + // pays with the engine authority, so this branch needs a hand-signed + // transaction: message compilation merges the signer and the writable output + // into account 0. Fees are zero and this SVM does no fee-payer validation, so + // a program-owned payer loads as-is. + let payer = Keypair::new(); + te.accounts() + .store(&[(payer.pubkey(), v42_owned(5, AccountMode::ReadOnly).into())]) + .unwrap(); + let (_sig, view) = te.signed_view(Some(&payer), E::lit(9).compose(payer.pubkey(), &[])); + let result = te.transaction(view).unwrap().execute().await.unwrap(); + assert_eq!(result, Err(TransactionError::InvalidAccountForFee)); + assert_eq!( + te.load_v42_data(payer.pubkey()), + Some(5), + "fee-payer write discarded" + ); + + // Positive control: a delegated (mutable) account commits normally. + let mutable = te.store_v42(0, AccountMode::Delegated); + assert!(te.execute(&[E::lit(9).compose(mutable, &[])]).await.is_ok()); + assert_eq!(te.load_v42_data(mutable), Some(9), "mutable write commits"); + + te.close().await; +} + +// Post-finalize actions are invoked via CPI after an account is created, and +// MagicRoot's `post_finalize` refuses to run them against a writable account that +// is not mutable. Creating a ReadOnly account with an attached v42 write is +// therefore rejected, and the whole creation rolls back — a distinct enforcement +// path from `validate_access` (this fires inside the program, not after). +#[tokio::test(flavor = "multi_thread")] +async fn post_finalize_immutable_action_is_rejected() { + let te = TestEngine::new().await; + let engine: &Engine = &te; + + let key = Pubkey::new_unique(); + let action = Some(vec![E::lit(9).compose(key, &[])]); + let acc = v42_owned(0, AccountMode::ReadOnly); + let result = engine.account(key).create(acc, action).await; + assert!( + result.is_err(), + "post-finalize write to an immutable account is refused" + ); + assert!( + te.account(key).is_none(), + "the rejected creation commits nothing" + ); + + te.close().await; +} + +// Privilege cannot be laundered through account creation: a single transaction +// that mixes MagicRoot's create-a-ReadOnly-account instructions with a top-level +// (foreign) v42 write is not privileged — `is_privileged` requires *every* +// instruction to be MagicRoot — so the guard runs and the whole transaction, +// creation included, reverts. +#[tokio::test(flavor = "multi_thread")] +async fn mixed_foreign_write_on_created_readonly_is_rejected() { + let te = TestEngine::new().await; + + let key = Pubkey::new_unique(); + let mut ixs: Vec = + AccountFieldPatch::sequence(v42_owned(0, AccountMode::ReadOnly)) + .into_iter() + .map(|patch| MagicRootInstruction::Patch(patch).compose(key).unwrap()) + .collect(); + ixs.push(MagicRootInstruction::Finalize.compose(key).unwrap()); + // The foreign instruction that makes the whole transaction non-privileged. + ixs.push(E::lit(9).compose(key, &[])); + + assert_eq!( + te.execute(&ixs).await, + Err(TransactionError::InvalidWritableAccount) + ); + assert!( + te.account(key).is_none(), + "the mixed transaction reverts wholesale" + ); + + te.close().await; +} diff --git a/engine/tests/transactions.rs b/engine/tests/transactions.rs new file mode 100644 index 0000000..c66284f --- /dev/null +++ b/engine/tests/transactions.rs @@ -0,0 +1,163 @@ +//! Transaction submission at the engine boundary: the `execute`, `simulate`, and +//! `schedule` wrappers around the sequencer. The processor suite already proves +//! the SVM commits/simulates correctly; these assert the `TransactionAccessor` +//! ergonomics on top — subscribe-then-await commit, the separate simulation +//! channel that never commits, and fire-and-forget scheduling. +#![cfg(test)] + +use engine::testkit::TestEngine; +use keeper::testkit::decode_v42; +use solana_account::AccountMode; +use solana_instruction_error::InstructionError; +use solana_transaction::TransactionError; +use v42_calculator_interface::builder::Expr as E; + +// Simulation runs against live state through the dedicated simulation channel +// but must not commit; execution of the same expression does. +#[tokio::test(flavor = "multi_thread")] +async fn simulate_does_not_commit_execute_does() { + let te = TestEngine::new().await; + let output = te.store_v42(0, AccountMode::Delegated); + let input = te.store_v42(40, AccountMode::Ephemeral); + let ixs = [(E::acc(1) + E::lit(2)).compose(output, &[input])]; + + // The record's post-execution account copy proves simulation actually ran + // the program, not merely that the channel round-tripped. + let record = te.simulate(&ixs).await.expect("simulation resolves"); + let executed = record.result.expect("simulated transaction processes"); + assert!(executed.was_successful(), "simulated execution succeeds"); + let (_, simulated) = executed + .loaded_transaction + .accounts + .iter() + .find(|(key, _)| *key == output) + .expect("simulation loaded the output account"); + assert_eq!( + decode_v42(simulated), + 42, + "simulation computed the result on its own account copy" + ); + assert_eq!( + te.load_v42_data(output), + Some(0), + "simulation leaves state untouched" + ); + + assert!(te.execute(&ixs).await.is_ok(), "execution resolves"); + assert_eq!( + te.load_v42_data(output), + Some(42), + "execution commits the computed value" + ); + + te.close().await; +} + +// A real processed transaction retains its execution artifacts through keeper's +// projection and the ledger's compressed append→index→reader round-trip. +#[tokio::test(flavor = "multi_thread")] +async fn processed_transaction_details_roundtrip() { + let te = TestEngine::new().await; + let slot = te.blocks().current_slot(); + let mut expected = Vec::new(); + + for value in [7, 42, 99] { + let output = te.store_v42(0, AccountMode::Ephemeral); + let ix = E::lit(value).cpi().compose(output, &[]); + let (signature, transaction) = te.signed_view(None, ix.clone()); + let bytes = transaction.inner_data().as_ref().clone(); + + te.execute(&[ix]).await.expect("processed transaction succeeds"); + expected.push((signature, bytes, value)); + } + + te.sync().await; + + for (signature, bytes, value) in expected { + let response = te + .transactions() + .get(signature) + .await + .expect("ledger read succeeds") + .expect("processed transaction is retained"); + assert_eq!(response.transaction, bytes); + assert_eq!(response.execution.header.signature, signature); + assert_eq!(response.execution.header.slot, slot); + assert!(response.execution.header.result.is_ok()); + + let details = response.execution.details.expect("execution details retained"); + assert_eq!( + details.fee, 0, + "the engine does not charge transaction fees" + ); + assert!( + !details.balances.pre.is_empty(), + "native balances were recorded" + ); + assert_eq!( + details.balances.pre, details.balances.post, + "the calculator changes account data, not lamports" + ); + assert!(details.logs.iter().any(|line| line.contains("v42:"))); + assert!(details.compute_units > 0); + assert!( + details + .cpi + .as_ref() + .is_some_and(|groups| groups.iter().any(|group| !group.0.is_empty())), + "the nested expression retains its CPI trace" + ); + let returned = details.return_data.expect("CPI return data retained"); + assert_eq!(returned.program, v42_calculator_interface::ID.to_bytes()); + assert_eq!(returned.data.as_slice(), &value.to_le_bytes()); + } + + te.close().await; +} + +// A transaction that runs but errors resolves as a committed error result and +// leaves its output account untouched — the engine surfaces the failure through +// the outer Ok / inner Err split rather than dropping it. +#[tokio::test(flavor = "multi_thread")] +async fn failed_execution_surfaces_error_result() { + let te = TestEngine::new().await; + let output = te.store_v42(5, AccountMode::Ephemeral); + // 1 - 2 underflows the program's checked_sub before any write. + let ixs = [(E::lit(1) - E::lit(2)).compose(output, &[])]; + + let error = te.execute(&ixs).await.expect_err("underflow yields an error result"); + // CalcError::Arithmetic = 6; its discriminants are stable for tests. + assert_eq!( + error, + TransactionError::InstructionError(0, InstructionError::Custom(6)), + "the program's own failure is surfaced, not a substitute" + ); + assert_eq!( + te.load_v42_data(output), + Some(5), + "failed execution commits no writes" + ); + + te.close().await; +} + +// schedule returns before the transaction commits; the write still lands, and an +// account subscription (not a poll loop) observes it. +#[tokio::test(flavor = "multi_thread")] +async fn schedule_is_fire_and_forget() { + let te = TestEngine::new().await; + let output = te.store_v42(0, AccountMode::Ephemeral); + let mut updates = te.accounts().subscribe(output).await; + let ixs = [E::lit(7).compose(output, &[])]; + + te.schedule(&ixs).await; + + let account = updates.recv().await.expect("scheduled write reaches the subscriber"); + assert_eq!( + decode_v42(&account), + 7, + "scheduled transaction commits the write" + ); + + te.close().await; +}