From 87c2933270ec3778fb8772a7215275bc067804b1 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 23 Jun 2026 18:28:57 +0400 Subject: [PATCH] feat: adds top level engine crate for global orchestration --- Cargo.toml | 2 + clippy.toml | 5 +- engine/Cargo.toml | 50 +++++++++ engine/README.md | 47 ++++++++ engine/src/accessor.rs | 118 ++++++++++++++++++++ engine/src/error.rs | 73 +++++++++++++ engine/src/lib.rs | 161 ++++++++++++++++++++++++++++ engine/src/pacemaker.rs | 175 ++++++++++++++++++++++++++++++ engine/src/transaction.rs | 37 +++++++ engine/tests/accounts.rs | 157 +++++++++++++++++++++++++++ engine/tests/harness.rs | 202 +++++++++++++++++++++++++++++++++++ engine/tests/recovery.rs | 130 ++++++++++++++++++++++ engine/tests/security.rs | 138 ++++++++++++++++++++++++ engine/tests/transactions.rs | 101 ++++++++++++++++++ 14 files changed, 1393 insertions(+), 3 deletions(-) create mode 100644 engine/Cargo.toml create mode 100644 engine/README.md create mode 100644 engine/src/accessor.rs create mode 100644 engine/src/error.rs create mode 100644 engine/src/lib.rs create mode 100644 engine/src/pacemaker.rs create mode 100644 engine/src/transaction.rs create mode 100644 engine/tests/accounts.rs create mode 100644 engine/tests/harness.rs create mode 100644 engine/tests/recovery.rs create mode 100644 engine/tests/security.rs create mode 100644 engine/tests/transactions.rs diff --git a/Cargo.toml b/Cargo.toml index 4981135..42b6910 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "accountsdb", + "engine", "keeper", "ledger", "nucleus", @@ -27,6 +28,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..a19e40d --- /dev/null +++ b/engine/Cargo.toml @@ -0,0 +1,50 @@ +[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" + +[dependencies] +keeper = { workspace = true } +ledger = { workspace = true } +magic-root-interface = { workspace = true } +magic-root-program = { workspace = true } +nucleus = { workspace = true } +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"] } +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", "time", "sync"] } + +[lints] +workspace = true diff --git a/engine/README.md b/engine/README.md new file mode 100644 index 0000000..5c43c68 --- /dev/null +++ b/engine/README.md @@ -0,0 +1,47 @@ +# `magicblock-engine` + +This is the top of the stack — the crate a consumer actually instantiates. It +wires the two layers below it together: the durable state (`keeper`, which is +accountsdb + ledger) and the transaction sequencer (`processor`). An `Engine` +owns the signing authority, the shared `Keeper`, and a sequencer handle, and +nothing else in the workspace depends on it. Everything above this point is the +caller's program. + +## Startup and recovery + +A healthy restart comes up on **persisted state** — the authoritative copy is +already on disk, so the engine just opens it. It does *not* replay the ledger or +rebuild volatile state on a normal startup. + +At startup the engine always registers MagicRoot as a native built-in before +opening keeper state. That keeps the account mutation path available even when a +caller supplies no built-ins of its own. + +Replay is the **corruption-recovery** path. When accountsdb opens corrupt, the +keeper restores an older archived snapshot, which leaves accountsdb trailing the +ledger tip. Only then does the engine replay the retained ledger entries from the +restored slot up to the tip, re-executing them through a temporary replay +sequencer to rebuild volatile state and catch back up. (When accountsdb is already +current, `replay` is a no-op.) + +Replay is also where silent corruption gets caught. At each sealed superblock the +replay is quiesced and the reconstructed account checksum is compared against the +checksum that was sealed into that superblock. A mismatch is +`ReplayError::StateMismatch` — it means the durable state diverged from what was +recorded, and surfacing it loudly on restart is far better than running on top of +quietly wrong state. + +Volatile state can also be dropped at runtime. Because it's a mirror of +chain-owned accounts the engine can always re-fetch, a caller that can no +longer trust the mirror is in sync with chain (for example, the chain-sync +connection dropped) may reset it; persisted ER-exclusive state is left intact. + +## What it exposes + +- `account()` — CRUD over accounts, expressed as MagicRoot instruction sequences. + Account mutation is privileged, so it can't be an ordinary write; it goes + through the MagicRoot built-in. +- `transaction()` — execute, schedule, or simulate a transaction. +- `IntoTransactionView` — signs and sanitizes a `Message`, an `&[Instruction]`, + or an existing view, using the engine authority and the latest blockhash, so + callers can hand in whichever form is convenient. diff --git a/engine/src/accessor.rs b/engine/src/accessor.rs new file mode 100644 index 0000000..92fcdaf --- /dev/null +++ b/engine/src/accessor.rs @@ -0,0 +1,118 @@ +//! Account- and transaction-scoped operation facades. + +use std::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 = self.patch_instructions(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 = self.patch_instructions(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 + } + + /// Builds the instruction sequence that patches every + /// field of `account` and finalizes it. + fn patch_instructions(&self, account: OwnedAccount) -> Result> { + let patches = AccountFieldPatch::sequence(account); + let mut instructions = Vec::with_capacity(patches.len() + 1); + for patch in patches { + instructions.push(MagicRootInstruction::Patch(patch).compose(self.pubkey)?); + } + instructions.push(MagicRootInstruction::Finalize.compose(self.pubkey)?); + Ok(instructions) + } + + /// 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> { + 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<()> { + 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..346e977 --- /dev/null +++ b/engine/src/error.rs @@ -0,0 +1,73 @@ +//! 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::{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 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(_: 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..d6adc3c --- /dev/null +++ b/engine/src/lib.rs @@ -0,0 +1,161 @@ +#![doc = include_str!("../README.md")] + +use std::{sync::Arc, time::Instant}; + +use derive_more::Deref; +use keeper::{Keeper, TransactionView, builder::KeeperBuilder}; +use ledger::schema::OwnedBlockestoreEntry; +use magic_root_program::entrypoint::MagicRootEntrypoint; +use nucleus::{ + runtime::{self, 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 solana_transaction::Hash; +use tracing::{error, info}; + +mod accessor; +mod error; +pub mod pacemaker; +mod transaction; + +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. + pub(crate) sequencer: SequencerHandle, +} + +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::replay(&state, &cache, cpus).await?; + let (service, sequencer) = Sequencer::new(cpus / 2, state.clone(), cache, shutdown, false)?; + service.spawn()?; + let engine = Self { state, sequencer }; + PaceMaker::spawn(engine.clone(), pacer, blockstore, shutdown); + info!(authority = %engine.authority(), cpus, "engine started"); + Ok(engine) + } + + /// Replays retained ledger entries from the accountsdb slot up to the ledger + /// tip through a temporary replay sequencer, rebuilding volatile state. A + /// no-op when accountsdb is already current. At each sealed superblock the + /// replay sequencer is quiesced via a [`barrier`] so the reconstructed + /// account checksum can be compared against the sealed value, surfacing + /// [`ReplayError::StateMismatch`] on divergence. + async fn 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()?; + while let Some(entry) = replayer.rx.recv().await { + match entry { + OwnedBlockestoreEntry::Transaction(txn) => { + let txn = TransactionView::try_new_sanitized(Arc::new(txn), false) + .map_err(ReplayError::Sanitization)?; + sequencer.send(SequencerMessage::Transaction(txn)).await?; + } + OwnedBlockestoreEntry::Block(block) => { + sequencer.send(SequencerMessage::Block(block)).await?; + } + OwnedBlockestoreEntry::Superblock(superblock) => { + let _release = barrier(&sequencer).await?; + // The published checksum is only refreshed by a sync flush + // (live sealing gets one from the snapshot export), so + // recompute it here before comparing against the seal. + state.flush()?; + let observed = state.accounts().checksum(); + let expected = superblock.checksum; + if observed != expected { + error!(observed, expected, "state mismatch; aborting replay"); + Err(ReplayError::StateMismatch)?; + } + } + }; + } + replayer + .response + .recv_timeout() + .await + .map_err(ReplayError::from)? + .map_err(ReplayError::from)?; + + drop(barrier(&sequencer).await?); + state.flush()?; + + let slot = state.blocks().latest().slot; + info!(slot, duration = ?timer.elapsed(), "ledger replay complete"); + drop(sequencer); + sh.terminate(ShutdownReason::Signalled); + shutdown.terminate().await; + Ok(()) + } + + /// Returns the latest block hash + pub fn blockhash(&self) -> Hash { + self.state.blocks().latest().hash + } + + /// Returns an accessor for reading and 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 }) + } +} + +async fn barrier(sequencer: &SequencerHandle) -> Result> { + let (controller, guard) = runtime::barrier(); + sequencer.send(SequencerMessage::Barrier(guard)).await?; + controller.acknowledged.await?; + Ok(controller.released) +} diff --git a/engine/src/pacemaker.rs b/engine/src/pacemaker.rs new file mode 100644 index 0000000..248954a --- /dev/null +++ b/engine/src/pacemaker.rs @@ -0,0 +1,175 @@ +//! Block-boundary pacing. + +use std::{ + num::NonZeroU64, + sync::Arc, + time::{Duration, UNIX_EPOCH}, +}; + +use derive_more::Deref; +use keeper::{builder::BlockstoreParams, error::KeeperError}; +use ledger::schema::Block; +use nucleus::{ + Slot, + shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason}, +}; +use processor::{SequencerMessage, SimulatorMessage}; +use tokio::{ + sync::{Notify, mpsc::Receiver}, + time::{self, Interval, MissedTickBehavior}, +}; +use tracing::error; + +use crate::{Engine, Result, barrier}; + +/// 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: time(), + }; + 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: Arc, +} + +impl PaceMaker { + /// Registers and starts the pacemaker task. + /// + /// Uses an external block source when supplied; otherwise emits slots on + /// the configured block interval starting from the keeper's current slot. + pub fn spawn( + engine: Engine, + pacer: Option, + blockstore: BlockstoreParams, + shutdown: &mut ShutdownManager, + ) { + let pacer = match pacer { + Some(rx) => Pacer::External(rx), + None => Pacer::Internal(BlockTicker::new(&engine, blockstore.blocktime)), + }; + let shutdown = shutdown.handle(Service::PaceMaker); + let superblock = blockstore.superblock; + let pacemaker = Self { engine, pacer, superblock }; + tokio::spawn(pacemaker.run(shutdown)); + } + + /// Runs the pacemaker until shutdown or a pacing error. + async fn run(mut self, mut shutdown: ShutdownHandle) { + let result = loop { + tokio::select! { + biased; + _ = shutdown.signalled() => { + break Ok(()); + } + r = self.pace() => { + if !*r.as_ref().unwrap_or(&false) { + break r.map(|_| ()); + } + } + } + }; + if let Err(error) = result { + error!(?error, "pace maker terminated with critical failure"); + shutdown.terminate(ShutdownReason::Error(error.into())); + return; + } + if let Pacer::Internal(ref mut t) = self.pacer { + let block = t.block(); + let _ = self.handle(block).await; + } + 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)) + } + }; + self.handle(block).await?; + if let Some(submitted) = submission { + submitted.notify_waiters(); + } + Ok(true) + } + + async fn handle(&self, block: Block) -> Result<()> { + self.sequencer.send(SequencerMessage::Block(block)).await?; + self.sequencer.simulation.send(SimulatorMessage::Block(block)).await?; + self.accounts().set_slot(block.slot).map_err(KeeperError::from)?; + if block.slot.is_multiple_of(self.superblock.get()) { + let _release = barrier(&self.sequencer).await?; + self.finalize_superblock()?; + } + Ok(()) + } +} + +/// Returns the current Unix timestamp in whole seconds. +fn time() -> i64 { + // NOTE: system time query is an infallible operation + UNIX_EPOCH.elapsed().unwrap_or_default().as_secs() as i64 +} diff --git a/engine/src/transaction.rs b/engine/src/transaction.rs new file mode 100644 index 0000000..36e921e --- /dev/null +++ b/engine/src/transaction.rs @@ -0,0 +1,37 @@ +//! Composing values into sanitized transaction views. + +use keeper::TransactionView; +use solana_instruction::Instruction; +use solana_transaction::{Message, Transaction}; + +use crate::{Engine, 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)?; + TransactionView::try_new_sanitized(data.into(), true).map_err(Into::into) + } +} + +impl IntoTransactionView for &[Instruction] { + fn compose(self, engine: &Engine) -> Result { + let msg = Message::new(self, Some(&engine.authority())); + msg.compose(engine) + } +} + +impl IntoTransactionView for TransactionView { + fn compose(self, _: &Engine) -> Result { + Ok(self) + } +} diff --git a/engine/tests/accounts.rs b/engine/tests/accounts.rs new file mode 100644 index 0000000..b3aac3a --- /dev/null +++ b/engine/tests/accounts.rs @@ -0,0 +1,157 @@ +//! 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. + +use engine::Engine; +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; + +mod harness; +use harness::TestEngine; + +/// 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.expect("create succeeds"); + + 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 + .expect("patch data succeeds"); + 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 + .expect("patch owner succeeds"); + 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 + .expect("update overwrites in place"); + 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.expect("delete succeeds"); + 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 + .expect("update succeeds"); + 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/harness.rs b/engine/tests/harness.rs new file mode 100644 index 0000000..6218966 --- /dev/null +++ b/engine/tests/harness.rs @@ -0,0 +1,202 @@ +//! Shared black-box harness for the engine integration suite. +//! +//! Builds a real [`Engine`] the same way [`keeper::testkit::TestKeeper`] builds a +//! keeper — reusing [`keeper_builder`] + [`seed_v42`] plus a prefunded authority — +//! and drives it through the public engine surface only. Block boundaries are fed +//! through an [`ExternalPacer`] so tests advance slots (and cross superblock +//! seals) deterministically instead of waiting on the wall-clock ticker. + +#![cfg(test)] +// A shared integration-test module: each test binary uses a different subset of +// the harness surface, so unused items here are expected. +#![allow(dead_code, unreachable_pub)] + +use std::{num::NonZeroU64, path::PathBuf, sync::Arc, time::Duration}; + +use derive_more::Deref; +use engine::{Engine, pacemaker::ExternalBlock}; +use keeper::{ + ExecutionRecord, + testkit::{Dirs, block, keeper_builder, load_v42, store_v42}, +}; +use nucleus::{Slot, shutdown::ShutdownManager}; +use solana_account::{AccountMode, AccountSharedData}; +use solana_instruction::Instruction; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_transaction::TransactionResult; +use tokio::{ + sync::{Notify, mpsc}, + time, +}; + +/// Slots sealed into each superblock. Kept small so a test crosses a seal in a +/// handful of blocks rather than the testkit default of 16. +pub const SUPERBLOCK: u64 = 4; + +/// A running engine over throwaway directories, plus the lifecycle and pacing +/// handles a black-box test needs. Derefs to [`Engine`] for the public surface. +#[derive(Deref)] +pub struct TestEngine { + #[deref] + engine: Engine, + shutdown: ShutdownManager, + authority: Arc, + dirs: Dirs, + pacer: mpsc::Sender, + /// Next slot to feed the pacemaker. + slot: Slot, +} + +impl TestEngine { + /// Fresh engine on fresh directories with a fresh authority. + pub async fn new() -> Self { + Self::with(Dirs::default(), Arc::new(Keypair::new())).await + } + + /// Builds the engine over `dirs`, signing its own transactions with + /// `authority`. Reopening after a restart must reuse the same `authority`: + /// the authority's prefunded sponsor account is part of the checksummed + /// state, so a fresh one would diverge from the sealed superblock and abort + /// replay recovery. + pub async fn with(dirs: Dirs, authority: Arc) -> Self { + Self::try_with(dirs, authority).await.expect("engine starts") + } + + /// Fallible [`Self::with`], surfacing `Engine::new` failures so recovery + /// tests can assert a refused startup (e.g. a replay checksum mismatch). + pub async fn try_with(dirs: Dirs, authority: Arc) -> engine::Result { + let mut builder = keeper_builder(&dirs); + builder.authority = authority.clone(); + builder.blockstore.superblock = NonZeroU64::new(SUPERBLOCK).unwrap(); + let (pacer, rx) = mpsc::channel(64); + let mut shutdown = ShutdownManager::default(); + let engine = Engine::new(builder, Some(rx), &mut shutdown).await?; + let slot = engine.blocks().current_slot(); + Ok(Self { + engine, + shutdown, + authority, + dirs, + pacer, + slot, + }) + } + + /// Reads the little-endian `u64` payload of an account, or `None` if closed. + pub fn v42_u64(&self, key: Pubkey) -> Option { + load_v42(&self.engine, key) + } + + /// Stores a delegated, v42-owned `u64` account and returns its pubkey. + /// + /// Delegated accounts are ER-controlled, so the store routes them to the + /// persisted backend — the same shape the processor suite uses for operands. + pub fn seed_v42_account(&self, value: u64) -> Pubkey { + store_v42(&self.engine, value, AccountMode::Delegated) + } + + /// Full committed account, or `None` when absent/closed. + pub fn account(&self, key: Pubkey) -> Option { + self.engine.accounts().get(&key).expect("account loads") + } + + /// Executes instructions and returns the committed transaction result. + pub async fn execute(&self, ixs: &[Instruction]) -> TransactionResult<()> { + self.transaction(ixs) + .expect("transaction composes") + .execute() + .await + .expect("execution responds") + } + + /// Simulates instructions without committing them. + pub async fn simulate(&self, ixs: &[Instruction]) -> TransactionResult { + self.transaction(ixs) + .expect("transaction composes") + .simulate() + .await + .expect("simulation responds") + } + + /// Schedules instructions without awaiting commit. + pub async fn schedule(&self, ixs: &[Instruction]) { + self.transaction(ixs) + .expect("transaction composes") + .schedule() + .await + .expect("transaction scheduled"); + } + + /// Feeds one block boundary and waits until the pacemaker has handled it + /// locally (and sealed a superblock when the slot lands on one). + pub async fn advance_once(&mut self) { + let notify = Arc::new(Notify::new()); + let submitted = notify.clone(); + let notified = notify.notified(); + tokio::pin!(notified); + // Register before sending so the pacemaker's notify_waiters is not missed. + notified.as_mut().enable(); + self.pacer + .send(ExternalBlock { + block: block(self.slot), + submitted, + }) + .await + .expect("pacer accepts block"); + time::timeout(Duration::from_secs(8), notified) + .await + .expect("block processed in time"); + self.slot += 1; + } + + /// Advances `n` block boundaries. + pub async fn advance(&mut self, n: u64) { + for _ in 0..n { + self.advance_once().await; + } + } + + /// Advances until a block on the next superblock boundary has been sealed, + /// returning the boundary slot that triggered `finalize_superblock`. + pub async fn seal_superblock(&mut self) -> Slot { + let boundary = self.slot.next_multiple_of(SUPERBLOCK); + while self.slot <= boundary { + self.advance_once().await; + } + boundary + } + + /// Seals the next superblock and waits for its accountsdb snapshot to + /// finish archiving (a detached thread), returning the archive path. + pub async fn seal_and_archive(&mut self) -> PathBuf { + let mut snapshots = self.accounts().subscribe_snapshots(); + self.seal_superblock().await; + time::timeout(Duration::from_secs(8), snapshots.recv()) + .await + .expect("snapshot archived in time") + .expect("snapshot broadcast delivers") + } + + /// Directory paths, for corruption/archival helpers. + pub fn dirs(&self) -> &Dirs { + &self.dirs + } + + /// Signals shutdown, drains every background service, and releases the last + /// keeper handle so the directories can be reopened. Returns the directories + /// and authority for a subsequent restart. + pub async fn close(self) -> (Dirs, Arc) { + let Self { + mut shutdown, + dirs, + authority, + engine, + .. + } = self; + engine.flush().expect("should flush the state"); + drop(engine); + shutdown.terminate().await; + (dirs, authority) + } +} diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs new file mode 100644 index 0000000..8ddcb14 --- /dev/null +++ b/engine/tests/recovery.rs @@ -0,0 +1,130 @@ +//! 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}; +use solana_pubkey::Pubkey; +use tokio::time; +use v42_calculator_interface::builder::Expr as E; + +mod harness; +use harness::TestEngine; + +/// 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.expect("commits"); + 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.seed_v42_account(0); + + // 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.tar.zst"), + "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 live accountsdb and drop the newest archive: the reopen falls + // back to snapshot 1 (K = 10) and replays everything after it. + keeper::testkit::corrupt(dirs.accounts.path()); + std::fs::remove_file(&s2).expect("newest archive removed"); + + // 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.v42_u64(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.seed_v42_account(0); + commit_and_seal(&mut te, key, 10).await; + // Direct store: lands in persisted state (and superblock 2's checksum) + // without a ledger entry. + te.seed_v42_account(7); + let s2 = commit_and_seal(&mut te, key, 20).await; + let (dirs, authority) = te.close().await; + + keeper::testkit::corrupt(dirs.accounts.path()); + std::fs::remove_file(&s2).expect("newest archive removed"); + + 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 the persisted state as-is. The direct-stored account +// is the probe: it exists only in persisted state — in no archived snapshot and +// in no ledger entry — so it survives the reopen only if nothing was restored; +// the post-seal transaction write pins the persisted tip alongside it. +#[tokio::test(flavor = "multi_thread")] +async fn clean_restart_reopens_persisted_state() { + let mut te = TestEngine::new().await; + let key = te.seed_v42_account(0); + commit_and_seal(&mut te, key, 10).await; + te.execute(&[E::lit(20).compose(key, &[])]).await.expect("post-seal commit"); + let direct = te.seed_v42_account(7); + let (dirs, authority) = te.close().await; + + let te2 = TestEngine::with(dirs, authority).await; + assert_eq!( + te2.v42_u64(key), + Some(20), + "persisted tip state reopened as-is" + ); + assert_eq!( + te2.v42_u64(direct), + Some(7), + "ledger-invisible account intact, so no snapshot was restored" + ); + + te2.close().await; +} diff --git a/engine/tests/security.rs b/engine/tests/security.rs new file mode 100644 index 0000000..8d8cc88 --- /dev/null +++ b/engine/tests/security.rs @@ -0,0 +1,138 @@ +//! 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`. + +use engine::Engine; +use keeper::testkit::{signed_view, store_v42, 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; + +mod harness; +use harness::TestEngine; + +// 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; + let engine: &Engine = &te; + + // A writable, non-payer immutable operand: `compose` marks it writable, so + // execution dirties it before the guard runs. + let operand = store_v42(engine, 5, AccountMode::ReadOnly); + assert_eq!( + te.execute(&[E::lit(9).compose(operand, &[])]).await, + Err(TransactionError::InvalidWritableAccount) + ); + assert_eq!( + te.v42_u64(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())]) + .expect("payer account stores"); + let (_sig, view) = signed_view( + &payer, + E::lit(9).compose(payer.pubkey(), &[]), + te.blockhash(), + ); + let result = te + .transaction(view) + .expect("transaction composes") + .execute() + .await + .expect("execution responds"); + assert_eq!(result, Err(TransactionError::InvalidAccountForFee)); + assert_eq!( + te.v42_u64(payer.pubkey()), + Some(5), + "fee-payer write discarded" + ); + + // Positive control: a delegated (mutable) account commits normally. + let mutable = te.seed_v42_account(0); + assert!(te.execute(&[E::lit(9).compose(mutable, &[])]).await.is_ok()); + assert_eq!(te.v42_u64(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).expect("compose patch")) + .collect(); + ixs.push(MagicRootInstruction::Finalize.compose(key).expect("compose finalize")); + // 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..cf3577f --- /dev/null +++ b/engine/tests/transactions.rs @@ -0,0 +1,101 @@ +//! 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. + +use keeper::testkit::decode_v42; +use solana_instruction_error::InstructionError; +use solana_transaction::TransactionError; +use v42_calculator_interface::builder::Expr as E; + +mod harness; +use harness::TestEngine; + +// 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.seed_v42_account(0); + let input = te.seed_v42_account(40); + 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("output account loaded"); + assert_eq!( + decode_v42(simulated), + 42, + "simulation computed the result on its own account copy" + ); + assert_eq!( + te.v42_u64(output), + Some(0), + "simulation leaves state untouched" + ); + + assert!(te.execute(&ixs).await.is_ok(), "execution resolves"); + assert_eq!( + te.v42_u64(output), + Some(42), + "execution commits the computed value" + ); + + 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.seed_v42_account(5); + // 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.v42_u64(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.seed_v42_account(0); + 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("account update delivered"); + assert_eq!( + decode_v42(&account), + 7, + "scheduled transaction commits the write" + ); + + te.close().await; +}