diff --git a/Cargo.toml b/Cargo.toml index e57c074b..56168ddd 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" } @@ -90,6 +92,7 @@ agave-transaction-view = "4.1.1" solana-account-info = "3.1.1" solana-clock = "3.1.0" solana-compute-budget-instruction = "=4.1.1" +solana-compute-budget-program = "=4.1.1" solana-cpi = "3.1.0" solana-ed25519-program = "3.0.0" solana-epoch-rewards = "3.0.1" diff --git a/README.md b/README.md index 01c073d3..5f8eb4dd 100644 --- a/README.md +++ b/README.md @@ -201,12 +201,12 @@ copy from the other backend, so there is only ever one live copy. `Transient` accounts remain authoritative and persisted even though runtime code cannot mutate them. -To change accounts directly, use `Engine::account(pubkey)`. `create`, `update`, -`patch`, and `delete` each run as one signed, committed transaction and require -the local signer to match the engine authority. +To replace accounts directly, use `Engine::account(pubkey)`. `create`, `update`, +and `delete` each run as one signed, committed transaction and require the local +signer to match the engine authority. ```rust -use solana_account::{AccountBuilder, AccountFieldPatch, AccountMode}; +use solana_account::{AccountBuilder, AccountMode}; use solana_pubkey::Pubkey; let key = Pubkey::new_unique(); @@ -214,25 +214,18 @@ let owner = Pubkey::new_unique(); let account = AccountBuilder::default() .lamports(2_000_000) .owner(owner) - .mode(AccountMode::Delegated) + .mode(AccountMode::ReadOnly) + .slot(1) .data(vec![1, 2, 3, 4]) .build(); engine.account(key).create(account, None).await?; -let current = engine.accounts().loader().load(&key)?; - -engine - .account(key) - .patch(vec![AccountFieldPatch::DataAt { - offset: 0, - data: vec![9; 4], - }]) - .await?; let replacement = AccountBuilder::default() .lamports(2_000_000) .owner(owner) - .mode(AccountMode::Delegated) + .mode(AccountMode::ReadOnly) + .slot(2) .data(vec![5; 4]) .build(); engine.account(key).update(replacement).await?; @@ -241,7 +234,12 @@ engine.account(key).delete().await?; Each mutation is one committed transaction. `create` can also run optional post-finalize instructions in that transaction; if an instruction fails, the -creation does not commit. +creation does not commit. Complete-account patches cover non-flag fields, and +finalization atomically installs the caller-supplied flags without changing +lamports. Callers are responsible for supplying current state; later +replacements remain subject to the account's slot and lifecycle rules. Internal +create composition places post-finalize instructions immediately after +finalization. Missing external accounts can be coordinated with `Engine::accounts().ensure`. The first caller receives `MissingAccount::Load`; concurrent callers receive a @@ -289,20 +287,23 @@ async fn submit( No polling loops — subscribe to what you care about and the engine pushes updates as they happen. -Keeper accessors expose Tokio broadcast receivers for live state: +Keeper accessors expose dedicated Tokio channels for live state: ```rust let mut account_updates = engine.accounts().subscribe(key).await; let mut blocks = engine.blocks().subscribe(); -let account = account_updates.recv().await?; -let block = blocks.recv().await?; +let account = account_updates.recv().await.expect("account stream is open"); +let block = blocks.recv().await.expect("block stream is open"); ``` Related accessors subscribe to program-owned accounts, cache evictions, snapshot completion, transaction status, logs, processed transactions, and -service messages. Broadcast consumers must handle `Lagged` when they fall -behind and `Closed` during shutdown; retained reads are available separately. +service messages. Signatures use terminal oneshot channels; other multicast +streams give each consumer a bounded queue and disconnect a consumer that falls +behind. Processed transactions, service messages, and cache evictions each have +one process-lifetime consumer and apply producer backpressure when its queue is +full. --- diff --git a/engine/Cargo.toml b/engine/Cargo.toml new file mode 100644 index 00000000..00dc5f68 --- /dev/null +++ b/engine/Cargo.toml @@ -0,0 +1,61 @@ +[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 = ["config", "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-compute-budget-program = { workspace = true, features = ["agave-unstable-api"] } +solana-instruction = { workspace = true } +solana-keypair = { workspace = true } +solana-message = { workspace = true } +solana-program-runtime = { workspace = true } +solana-pubkey = { workspace = true } +solana-sdk-ids = { workspace = true } +solana-signer = { workspace = true } +solana-system-program = { workspace = true, features = ["agave-unstable-api"] } +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-packet = { workspace = true } +solana-signer = { workspace = true } +solana-system-interface = { workspace = true, features = ["bincode"] } +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 00000000..c68ac2e4 --- /dev/null +++ b/engine/README.md @@ -0,0 +1,62 @@ +# `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 and the System Program as native +builtins 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. + +## Account replacement + +`AccountAccessor::{create, update}` composes complete-account MagicRoot patch +transactions. Replacement slots are monotonic: a newer slot is accepted, an +equal slot requires a genuine account-mode transition, and an older slot is +rejected even when the mode changes. Failed replacements are transactionally +rolled back. Complete-account patch sequences cover non-flag fields, while +finalization atomically installs the caller-supplied complete flag value without +changing lamports. Callers are responsible for supplying current state; later +replacements remain subject to the account's slot and lifecycle rules. `create` +appends any `PostFinalize` actions immediately after finalization in the same +transaction. Magicblock construction rejects instruction, address, account-meta, +and instruction-data lengths that cannot be represented by the V1 wire fields. + +## Startup and recovery + +Keeper restores an accountsdb snapshot when the active store is corrupt, its +sealed superblock trails the retained ledger, or its committed transaction count +trails the ledger's durable count. Accountsdb's count is a checkpoint high-water +mark, so a count ahead of the locally retained ledger is current, including for +snapshots staged by a replication follower. Superblock lag remains recoverable +independently of the counters. + +If accountsdb then trails the ledger tip, `Engine::new` replays retained entries +from the successor of its sealed snapshot 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 when its slot and transaction count are each at least the +ledger values. After replay actually runs, the final transaction counts must be +equal or startup returns `ReplayError::StateMismatch`. + +Internal pacing appends one reset marker at the current 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. External block producers supply the slot and timestamp; +the sequencer overwrites hash-chain metadata with its locally computed hash and +parent. + +## 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 pacemaker holds the sequencer +barrier while issuing a terminal ledger sync, which closes the appender and +reader workers without waiting for every engine handle to be dropped. + +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 00000000..87116c34 --- /dev/null +++ b/engine/src/accessor.rs @@ -0,0 +1,106 @@ +//! 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::OwnedAccount; +use solana_instruction::Instruction; +use solana_pubkey::Pubkey; +use solana_transaction::TransactionResult; +use tokio::time; + +use crate::{Engine, error::EngineError, error::Result, transaction}; + +/// 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, + acc: impl Into, + actions: Option>, + ) -> Result<()> { + let mut instructions = MagicRootInstruction::compose_account(self.pubkey, acc.into())?; + 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, acc: impl Into) -> Result<()> { + let instructions = MagicRootInstruction::compose_account(self.pubkey, acc.into())?; + 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 = transaction::magicblock(&instructions, self.engine)?; + self.engine.transaction(txn)?.execute().await?.map_err(Into::into) + } +} + +impl TransactionAccessor<'_> { + /// Submits `transaction` for execution and awaits its committed result. + /// A timeout does not cancel the submitted transaction. + pub async fn execute(self) -> Result> { + if self.engine.terminating.load(Ordering::Acquire) { + return Err(EngineError::ShuttingDown); + } + let signature = self.transaction.signatures()[0]; + let msg = SequencerMessage::Transaction(self.transaction); + let rx = self.engine.transactions().subscribe_signature(signature).await; + self.engine.sequencer.send(msg).await?; + let status = time::timeout(EXECUTION_TIMEOUT, rx) + .await + .map_err(|_| EngineError::TransactionTimeout)? + .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) { + return Err(EngineError::ShuttingDown); + } + 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> { + if self.engine.terminating.load(Ordering::Acquire) { + return Err(EngineError::ShuttingDown); + } + 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 00000000..0e297bee --- /dev/null +++ b/engine/src/error.rs @@ -0,0 +1,91 @@ +//! 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_message::CompileError; +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), + /// The engine has begun coordinated shutdown and rejects new work. + #[error("engine is shutting down")] + ShuttingDown, + /// Timed out waiting for a submitted transaction's committed result. + #[error("timed out waiting for transaction result")] + TransactionTimeout, + /// 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), + /// Compiling instructions into a versioned transaction message failed. + #[error("transaction compilation failed: {0}")] + TransactionCompile(#[source] CompileError), + /// 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(#[source] LedgerRequestError), + /// Reading or decoding retained ledger entries failed. + #[error("ledger replay failed: {0}")] + Ledger(#[source] LedgerError), +} diff --git a/engine/src/lib.rs b/engine/src/lib.rs new file mode 100644 index 00000000..954737d0 --- /dev/null +++ b/engine/src/lib.rs @@ -0,0 +1,220 @@ +#![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::OwnedBlockstoreEntry; +use magic_root_program::entrypoint::MagicRootEntrypoint; +use nucleus::{ + runtime::{self, BarrierHandle, SequencerHandle}, + shutdown::{Service, ShutdownManager, ShutdownReason}, +}; +use processor::{SequencerMessage, sequencer::Sequencer}; +use solana_compute_budget_program::Entrypoint as ComputeBudgetEntrypoint; +use solana_program_runtime::{ + loaded_programs::{ProgramCache, ProgramCacheEntry}, + solana_sbpf::program::BuiltinFunctionDefinition, +}; +use solana_pubkey::Pubkey; +use solana_system_program::system_processor::Entrypoint as SystemProgramEntrypoint; +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), + ); + builder.builtins.insert( + solana_system_program::id(), + ( + SystemProgramEntrypoint::vm, + SystemProgramEntrypoint::codegen, + ), + ); + builder.builtins.insert( + solana_sdk_ids::compute_budget::id(), + ( + ComputeBudgetEntrypoint::vm, + ComputeBudgetEntrypoint::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.sync(true).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: OwnedBlockstoreEntry) -> Result<()> { + match entry { + OwnedBlockstoreEntry::Transaction(txn) => self.transaction(txn)?.schedule().await?, + OwnedBlockstoreEntry::Block(block) => { + self.sequencer.send(SequencerMessage::Block(block)).await?; + } + OwnedBlockstoreEntry::Superblock(expected) => { + let _guard = self.barrier().await?; + let previous = self.superblocks().sealed().id; + self.accounts().set_superblock(expected.id); + self.sync(false)?; + let observed = self.superblocks().sealed(); + if observed != expected { + error!(?observed, ?expected, "state mismatch; aborting replay"); + self.accounts().set_superblock(previous); + self.sync(false)?; + Err(ReplayError::StateMismatch)?; + } + } + OwnedBlockstoreEntry::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.sync(false)?; + let accountsdb = state.accounts().transactions(); + let ledger = state.ledger().transactions(); + if accountsdb != ledger { + error!( + accountsdb, + ledger, "transaction count mismatch; aborting replay" + ); + Err(ReplayError::StateMismatch)?; + } + + let slot = state.blocks().latest().slot; + info!(slot, duration = ?timer.elapsed(), "ledger replay complete"); + 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 00000000..dbea204c --- /dev/null +++ b/engine/src/pacemaker.rs @@ -0,0 +1,190 @@ +//! Block-boundary pacing. + +use std::{num::NonZeroU64, time::Duration}; + +use derive_more::Deref; +use ledger::schema::Block; +use nucleus::{ + Slot, + config::BlockstoreParams, + 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 time = unix_time().as_secs() as i64; + let block = Block::new(self.slot, time); + self.slot += 1; + block + } +} + +/// Block boundary submitted by an external producer. +/// +/// The caller supplies its slot and timestamp. The sequencer overwrites the +/// hash and parent with locally computed hash-chain metadata. +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 { + let next = tokio::select! { + biased; + _ = shutdown.signalled() => None, + next = self.next() => next, + }; + let Some((block, submission)) = next else { + break Ok(()); + }; + if let Err(error) = self.handle(block).await { + break Err(error); + } + if let Some(submission) = submission { + let _ = submission.send(()); + } + }; + res = if let Pacer::Internal(ref mut t) = self.pacer { + // Await every shutdown step even after an earlier failure. + let b = t.block(); + res.and(self.handle(b).await).and(self.shutdown(false).await) + } else { + res.and(self.shutdown(true).await) + }; + // Release engine storage before the manager can reopen it. + drop(self); + if let Err(error) = res { + error!(?error, "pace maker terminated with critical failure"); + shutdown.terminate(ShutdownReason::Error(error.into())); + } else { + shutdown.terminate(ShutdownReason::Signalled); + } + } + + /// Waits for the next block boundary without applying it. + async fn next(&mut self) -> Option<(Block, Option>)> { + match &mut self.pacer { + Pacer::Internal(t) => { + t.ticker.tick().await; + Some((t.block(), None)) + } + Pacer::External(rx) => rx.recv().await.map(|msg| (msg.block, Some(msg.submitted))), + } + } + + /// 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 00000000..7939834c --- /dev/null +++ b/engine/src/testkit.rs @@ -0,0 +1,177 @@ +//! 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::KeeperBuilder, + testkit::{Dirs, SUPERBLOCK, await_archive, block, keeper_builder}, +}; +use nucleus::{Slot, config::Authority, ledger::BlockstorePosition, shutdown::ShutdownManager}; +use solana_account::AccountSharedData; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_transaction::TransactionResult; +use tokio::{sync::mpsc, time}; + +use crate::{Engine, IntoTransactionView, 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(false).unwrap(); + self.superblocks().position() + } + + /// Full committed account, or `None` when absent/closed. + pub fn get_account(&self, key: Pubkey) -> Option { + self.engine.accounts().loader().load(&key).unwrap() + } + + /// Executes instructions and returns the committed transaction result. + pub async fn execute(&self, txn: impl IntoTransactionView) -> TransactionResult<()> { + self.transaction(txn).unwrap().execute().await.unwrap() + } + + /// Simulates instructions without committing them. + pub async fn simulate( + &self, + txn: impl IntoTransactionView, + ) -> TransactionResult { + self.transaction(txn).unwrap().simulate().await.unwrap() + } + + /// Schedules instructions without awaiting commit. + pub async fn schedule(&self, txn: impl IntoTransactionView) { + self.transaction(txn).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(); + time::timeout(TIMEOUT, submitted) + .await + .expect("pacemaker accepts the block in time") + .expect("pacemaker reports block submission"); + self.slot += 1; + } + } + + /// Seals the next superblock and waits for its snapshot archive. + pub async fn seal_and_archive(&mut self) -> PathBuf { + let boundary = self.slot.next_multiple_of(SUPERBLOCK.into()); + while self.slot <= boundary { + self.advance(1).await; + } + await_archive(self).await + } + + /// Stops every service and returns the directories and authority for reopen. + pub async fn close(self) -> (Dirs, Authority) { + let Self { + mut shutdown, dirs, authority, .. + } = self; + shutdown.terminate().await; + (dirs, authority) + } +} diff --git a/engine/src/transaction.rs b/engine/src/transaction.rs new file mode 100644 index 00000000..d4be48c3 --- /dev/null +++ b/engine/src/transaction.rs @@ -0,0 +1,122 @@ +//! Composing values into sanitized transaction views. + +use agave_transaction_view::{ + MAGICBLOCK_INSTRUCTION_TRACE_LENGTH, MAX_MAGICBLOCK_ACCOUNT_LOCKS, + transaction_version::{MAGICBLOCK_PREFIX, TransactionVersion}, +}; +use keeper::TransactionView; +use solana_instruction::Instruction; +use solana_message::{ + VersionedMessage, + v1::{self, SIGNATURE_SIZE}, +}; +use solana_signer::Signer; +use solana_transaction::{Message, Transaction, TransactionError, versioned::VersionedTransaction}; + +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.signer()], engine.blockhash())?; + transaction.compose(engine) + } +} + +impl IntoTransactionView for Transaction { + fn compose(self, engine: &Engine) -> Result { + let data = wincode::serialize(&self).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 &[Instruction; N] { + fn compose(self, engine: &Engine) -> Result { + self.as_slice().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: &Engine) -> Result { + if matches!(self.version(), TransactionVersion::Magicblock) + && self.static_account_keys()[0] != engine.authority() + { + return Err(EngineError::SignatureVerification); + } + 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, so downstream code may assume the fee payer and every +/// required signer actually signed. +fn sigverify(view: &TransactionView) -> Result<()> { + // Sanitization guarantees one static key for every required signature. + let message = view.message_data(); + for (signature, key) in view.signatures().iter().zip(view.static_account_keys()) { + if !signature.verify(key.as_ref(), message) { + return Err(EngineError::SignatureVerification); + } + } + Ok(()) +} + +/// Composes an Engine-private transaction and signs its final +/// Magicblock wire representation with the Engine authority. +pub(crate) fn magicblock(instructions: &[Instruction], engine: &Engine) -> Result> { + let message = v1::Message::try_compile(&engine.authority(), instructions, engine.blockhash())?; + let message = VersionedMessage::V1(message); + // These checks are merely future proof defenses, currently it should be + // impossible to construct a transaction which might violate any of them + if message.instructions().len() > MAGICBLOCK_INSTRUCTION_TRACE_LENGTH { + Err(TransactionError::SanitizeFailure)?; + } else if message.static_account_keys().len() > MAX_MAGICBLOCK_ACCOUNT_LOCKS { + Err(TransactionError::TooManyAccountLocks)?; + } + for ix in message.instructions() { + if ix.accounts.len() > MAX_MAGICBLOCK_ACCOUNT_LOCKS { + Err(TransactionError::TooManyAccountLocks)?; + } + } + + // Reserve the trailing signature slot without signing the V1 prefix, which + // is replaced below before the only signing operation. + let transaction = VersionedTransaction { + signatures: vec![Default::default()], + message, + }; + let mut data = wincode::serialize(&transaction).map_err(wincode::Error::from)?; + // Patch the transaction prefix to allow for larger tranaction limits + data[0] = MAGICBLOCK_PREFIX; + + let signature_offset = data.len() - SIGNATURE_SIZE; + let signature = engine.signer().sign_message(&data[..signature_offset]); + data[signature_offset..].copy_from_slice(signature.as_ref()); + Ok(data) +} diff --git a/engine/tests/accounts.rs b/engine/tests/accounts.rs new file mode 100644 index 00000000..3e17cf96 --- /dev/null +++ b/engine/tests/accounts.rs @@ -0,0 +1,418 @@ +//! 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/update/delete round-trip and the sponsor-balance invariant, and +//! that post-finalize actions actually run. +#![cfg(test)] + +use engine::{Engine, EngineError, testkit::TestEngine}; +use keeper::testkit::{ + V42_ID, load_v42_data, load_v42_lamports, patterned_bytes, store_v42, v42_builder, +}; +use solana_account::{AccountBuilder, AccountMode, OwnedAccount, ReadableAccount}; +use solana_instruction_error::InstructionError; +use solana_pubkey::Pubkey; +use solana_system_interface::MAX_PERMITTED_DATA_LENGTH; +use solana_sysvar::rent::Rent; +use solana_transaction::TransactionError; +use v42_calculator_interface::builder::{Expr as E, transfer}; + +/// 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; +const SLOT: u64 = 42; + +/// Account with explicit lifecycle state, funded at the shared rent-exempt balance. +fn account(owner: Pubkey, data: Vec, mode: AccountMode, slot: u64) -> OwnedAccount { + AccountBuilder::default() + .lamports(LAMPORTS) + .owner(owner) + .mode(mode) + .slot(slot) + .data(data) + .build() +} + +/// Delegated account with `data` at `slot`. +fn delegated(owner: Pubkey, data: Vec, slot: u64) -> OwnedAccount { + account(owner, data, AccountMode::Delegated, slot) +} + +/// Materializes `mode`, entering transient through its required delegated state. +async fn create_with(engine: &Engine, key: Pubkey, owner: Pubkey, mode: AccountMode) { + let initial = if mode == AccountMode::Transient { + delegated(owner, vec![1], SLOT - 1) + } else { + account(owner, vec![1], mode, SLOT) + }; + engine + .account(key) + .create(initial, None) + .await + .expect("initial account is created"); + if mode == AccountMode::Transient { + engine + .account(key) + .update(account(owner, vec![1], mode, SLOT)) + .await + .expect("delegated account enters transient"); + } +} + +/// Asserts MagicRoot rejected the slot patch in a complete-account sequence. +fn assert_non_advancing_slot(error: EngineError) { + let errored = matches!( + error, + EngineError::TransactionExecution(TransactionError::InstructionError( + 2, + InstructionError::InvalidArgument + )) + ); + assert!(errored, "unexpected replacement error: {error:?}"); +} + +/// Asserts MagicRoot rejected the mode patch in a complete-account sequence. +fn assert_invalid_mode_transition(error: EngineError) { + let errored = matches!( + error, + EngineError::TransactionExecution(TransactionError::InstructionError( + 1, + InstructionError::InvalidArgument + )) + ); + assert!(errored, "unexpected replacement error: {error:?}"); +} + +// The full lifecycle. `create` materializes a fresh account by patching every +// non-flag field, balancing lamport patches against the authority, then +// finalizing its flags; `update` overwrites an existing account or materializes +// a fresh key; and +// `delete` closes it. Mutations here keep the balance constant after creation. +#[tokio::test(flavor = "multi_thread")] +async fn account_crud_lifecycle() { + let te = TestEngine::new().await; + let key = Pubkey::new_unique(); + let owner = Pubkey::new_unique(); + + let created = account( + owner, + vec![1, 2, 3, 4, 5, 6, 7, 8], + AccountMode::ReadOnly, + 10, + ); + let authority_before = te.get_account(te.authority()).expect("sponsor exists").lamports(); + + te.account(key).create(created, None).await.unwrap(); + + let acc = te.get_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.is(AccountMode::ReadOnly)); + + let authority_after = te.get_account(te.authority()).expect("sponsor exists").lamports(); + assert_eq!( + authority_before - authority_after, + LAMPORTS, + "the lamport patch sponsors the created balance from the authority" + ); + + // update overwrites the existing account in place: same-length data (the + // patch sequence replaces the exact data length) and identical lamports. + // Read-only accounts remain replaceable after finalization. + te.account(key) + .update(account(owner, vec![5; 16], AccountMode::ReadOnly, 11)) + .await + .unwrap(); + let acc = te.get_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"); + assert!(acc.is(AccountMode::ReadOnly)); + + // delete: the account is gone from storage. + te.account(key).delete().await.unwrap(); + assert!(te.get_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(); + te.account(key2).update(delegated(owner, vec![3; 8], 10)).await.unwrap(); + assert_eq!(te.get_account(key2).expect("materialized").data(), &[3; 8]); + + te.close().await; +} + +// Account cloning reconstructs every field and data chunk in one atomic private +// transaction. Growing the same clone through the 64 KiB boundary and beyond, +// then shrinking it below the boundary and to empty, proves replacement keeps +// the exact data length. The caller supplies each successive current state. +#[tokio::test(flavor = "multi_thread")] +async fn account_clone_create_and_update_accept_large_data() { + const MAX_DATA_LEN: usize = 128 * 1024 + 1; + + let te = TestEngine::new().await; + let key = Pubkey::new_unique(); + let owner = Pubkey::new_unique(); + let lamports = Rent::default().minimum_balance(MAX_DATA_LEN); + + for (index, (len, seed)) in [ + (u16::MAX as usize, 1), + (64 * 1024, 2), + (MAX_DATA_LEN, 3), + (32 * 1024, 4), + (0, 5), + ] + .into_iter() + .enumerate() + { + let data = patterned_bytes(len, seed); + let account = AccountBuilder::default() + .lamports(lamports) + .owner(owner) + .mode(AccountMode::ReadOnly) + .slot(SLOT + index as u64) + .data(data.clone()); + + if index == 0 { + te.account(key).create(account, None).await.unwrap(); + } else { + te.account(key).update(account).await.unwrap(); + } + + let stored = te.get_account(key).expect("large account exists"); + assert_eq!(stored.lamports(), lamports); + assert_eq!(stored.owner(), &owner); + assert!(stored.is(AccountMode::ReadOnly)); + assert_eq!(stored.slot(), SLOT + index as u64); + assert_eq!(stored.data(), data); + } + + te.close().await; +} + +/// Proves an exact maximum-sized Solana account can run a PostFinalize SBPF +/// action above trace index 64, while 257 subsequent V42 self-CPIs hit the CPI +/// trace limit and roll back both the account creation and an earlier action. +#[tokio::test(flavor = "multi_thread")] +async fn account_create_accepts_max_data_with_post_finalize() { + const CPI_CALLS: usize = 257; + + let te = TestEngine::new().await; + let key = Pubkey::new_unique(); + let source = store_v42(&te, 7, AccountMode::Delegated); + let output = store_v42(&te, 0, AccountMode::Ephemeral); + let data = patterned_bytes(MAX_PERMITTED_DATA_LENGTH as usize, 42); + let account = AccountBuilder::default() + .lamports(Rent::default().minimum_balance(data.len())) + .owner(Pubkey::new_unique()) + .mode(AccountMode::Delegated) + .slot(SLOT) + .data(data.clone()); + let action = transfer(source, output, 1); + + te.account(key) + .create(account, Some(vec![action])) + .await + .expect("maximum-sized account and post-finalize action execute atomically"); + + let stored = te.get_account(key).expect("maximum-sized account exists"); + assert_eq!(stored.data().len(), MAX_PERMITTED_DATA_LENGTH as usize); + assert!(stored.data() == data, "maximum-sized account data differs"); + assert_eq!(load_v42_data(&te, source), Some(6)); + assert_eq!(load_v42_data(&te, output), Some(1)); + + let failed_key = Pubkey::new_unique(); + let failed_account = AccountBuilder::default() + .lamports(Rent::default().minimum_balance(data.len())) + .owner(Pubkey::new_unique()) + .mode(AccountMode::Delegated) + .slot(SLOT) + .data(data); + let excessive_cpis = (1..CPI_CALLS) + .fold(E::lit(1).cpi(), |expr, _| expr + E::lit(1).cpi()) + .compose(output, &[]); + let error = te + .account(failed_key) + .create( + failed_account, + Some(vec![transfer(source, output, 1), excessive_cpis]), + ) + .await + .expect_err("257 V42 self-CPIs exceed the trace limit"); + + assert!( + matches!( + error, + EngineError::TransactionExecution(TransactionError::InstructionError( + _, + InstructionError::MaxInstructionTraceLengthExceeded + )) + ), + "unexpected CPI trace error: {error:?}" + ); + assert!( + te.get_account(failed_key).is_none(), + "failed creation was rolled back" + ); + assert_eq!(load_v42_data(&te, source), Some(6)); + assert_eq!(load_v42_data(&te, output), Some(1)); + + te.close().await; +} + +// Replacing the seeded executable at a newer slot exercises MagicRoot's large +// account reconstruction and executable finalization, then proves the resulting +// program remains usable by executing it through the normal client path. +#[tokio::test(flavor = "multi_thread")] +async fn account_update_replaces_v42_program_at_new_slot() { + let te = TestEngine::new().await; + let program = te.get_account(V42_ID).expect("v42 program is seeded"); + let data = program.data().to_vec(); + let owner = *program.owner(); + let lamports = program.lamports(); + let slot = program.slot() + 1; + let replacement = AccountBuilder::from(program).slot(slot); + + te.account(V42_ID).update(replacement).await.unwrap(); + + let replaced = te.get_account(V42_ID).expect("replaced v42 program exists"); + assert_eq!(replaced.lamports(), lamports); + assert_eq!(replaced.owner(), &owner); + assert_eq!(replaced.slot(), slot); + assert!(replaced.executable()); + assert_eq!(replaced.data(), data); + + let output = store_v42(&te, 0, AccountMode::Ephemeral); + te.execute(&[E::lit(42).compose(output, &[])]) + .await + .expect("replaced v42 program executes"); + assert_eq!(load_v42_data(&te, output), Some(42)); + + te.close().await; +} + +// Complete replacements are monotonic by slot. An equal-slot replacement is +// meaningful only when it performs a real lifecycle transition; mode is patched +// before slot, and no-op mode writes deliberately leave the mode marker clean. +#[tokio::test(flavor = "multi_thread")] +async fn account_replacement_slot_ordering() { + let te = TestEngine::new().await; + let owner = Pubkey::new_unique(); + + for (from, to) in [ + (AccountMode::ReadOnly, AccountMode::Delegated), + (AccountMode::Placeholder, AccountMode::Ephemeral), + ] { + let key = Pubkey::new_unique(); + create_with(&te, key, owner, from).await; + te.account(key) + .update(account(owner, vec![2], to, SLOT)) + .await + .expect("equal-slot mode transition is accepted"); + + let updated = te.get_account(key).expect("transitioned account exists"); + assert!(updated.is(to), "{from:?} transitions to {to:?}"); + assert_eq!(updated.slot(), SLOT); + assert_eq!(updated.data(), &[2]); + } + + for (from, to) in [ + (AccountMode::Placeholder, AccountMode::Transient), + (AccountMode::Ephemeral, AccountMode::Delegated), + (AccountMode::System, AccountMode::ReadOnly), + ] { + let key = Pubkey::new_unique(); + // Seed the source directly so only the mode-transition invariant is + // under test. + te.accounts() + .store(&[(key, account(owner, vec![1], from, SLOT).into())]) + .unwrap(); + let error = te + .account(key) + .update(account(owner, vec![2], to, SLOT)) + .await + .expect_err("invalid mode transition is rejected"); + assert_invalid_mode_transition(error); + + let unchanged = te.get_account(key).expect("rejected transition preserves the account"); + assert!(unchanged.is(from), "{from:?} does not transition to {to:?}"); + assert_eq!(unchanged.slot(), SLOT); + assert_eq!(unchanged.data(), &[1]); + } + + let key = Pubkey::new_unique(); + te.account(key) + .create(account(owner, vec![3], AccountMode::ReadOnly, SLOT), None) + .await + .expect("baseline account is created"); + + let error = te + .account(key) + .update(account(owner, vec![4], AccountMode::ReadOnly, SLOT)) + .await + .expect_err("equal-slot replacement without a mode change is rejected"); + assert_non_advancing_slot(error); + + let error = te + .account(key) + .update(account(owner, vec![5], AccountMode::Delegated, SLOT - 1)) + .await + .expect_err("a mode change never authorizes an older slot"); + assert_non_advancing_slot(error); + + let unchanged = te.get_account(key).expect("rejected replacements preserve the account"); + assert!(unchanged.is(AccountMode::ReadOnly)); + assert_eq!(unchanged.slot(), SLOT); + assert_eq!(unchanged.data(), &[3]); + + 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; + + // A successful v42 transfer proves the post-finalize action ran after the + // new account became writable and program-owned. + let source = store_v42(&te, 0, AccountMode::Delegated); + let source_before = load_v42_lamports(&te, source).expect("source exists"); + let ok_key = Pubkey::new_unique(); + let acc = v42_builder(0, AccountMode::Delegated); + let benign = transfer(source, ok_key, 1); + te.account(ok_key) + .create(acc, Some(vec![benign])) + .await + .expect("create with a succeeding post-finalize action"); + assert_eq!( + load_v42_lamports(&te, source).expect("source remains"), + source_before - 1, + "post-finalize action debited its source" + ); + assert_eq!( + load_v42_lamports(&te, ok_key).expect("created account exists"), + source_before + 1, + "post-finalize action credited the created account" + ); + + // An overflowing 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(i64::MIN) - E::lit(1)).compose(bad_key, &[]); + let acc = v42_builder(0, AccountMode::Delegated); + let result = te.account(bad_key).create(acc, Some(vec![failing])).await; + assert!( + result.is_err(), + "failing post-finalize action surfaces an error" + ); + assert!( + te.get_account(bad_key).is_none(), + "nothing commits when the action fails" + ); + + te.close().await; +} diff --git a/engine/tests/builtins.rs b/engine/tests/builtins.rs new file mode 100644 index 00000000..595da1c4 --- /dev/null +++ b/engine/tests/builtins.rs @@ -0,0 +1,62 @@ +//! Full-engine coverage for native builtins registered during startup. +#![cfg(test)] + +use engine::testkit::TestEngine; +use solana_account::{AccountBuilder, AccountMode, AccountSharedData, ReadableAccount}; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_signer::Signer; +use solana_system_interface::{ + instruction::{allocate, assign, transfer}, + program, +}; +use solana_sysvar::rent::Rent; +use solana_transaction::Transaction; + +#[tokio::test(flavor = "multi_thread")] +async fn system_program_executes_transfer_allocate_and_assign() { + const LAMPORTS: u64 = 42; + const SPACE: usize = 8; + + let te = TestEngine::new().await; + let source = te.authority(); + + let source_before = te.get_account(source).expect("authority account remains"); + assert_eq!(source_before.owner(), &program::ID); + + let destination = Keypair::new(); + let destination_before = Rent::default().minimum_balance(SPACE); + let account: AccountSharedData = AccountBuilder::default() + .lamports(destination_before) + .mode(AccountMode::Delegated) + .build(); + assert_eq!(account.owner(), &program::ID); + te.accounts().store(&[(destination.pubkey(), account)]).unwrap(); + + let owner = Pubkey::new_unique(); + let instructions = [ + transfer(&source, &destination.pubkey(), LAMPORTS), + allocate(&destination.pubkey(), SPACE as u64), + assign(&destination.pubkey(), &owner), + ]; + let transaction = Transaction::new_signed_with_payer( + &instructions, + Some(&source), + &[te.signer(), &destination], + te.blockhash(), + ); + te.execute(transaction).await.expect("failed to execute system ixs"); + + let source_after = te.get_account(source).expect("authority account remains"); + assert_eq!(source_after.lamports(), source_before.lamports() - LAMPORTS); + assert_eq!(source_after.data(), source_before.data()); + assert_eq!(source_after.owner(), source_before.owner()); + + let destination_after = + te.get_account(destination.pubkey()).expect("destination account remains"); + assert_eq!(destination_after.lamports(), destination_before + LAMPORTS); + assert_eq!(destination_after.data(), &[0; SPACE]); + assert_eq!(destination_after.owner(), &owner); + + te.close().await; +} diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs new file mode 100644 index 00000000..9130b995 --- /dev/null +++ b/engine/tests/recovery.rs @@ -0,0 +1,146 @@ +//! Full-engine replay recovery — the engine's most distinctive orchestration. +//! After an accountsdb inconsistency 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, load_v42_data, store_v42}; +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: i64) -> 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_counter_lag() { + let mut te = TestEngine::new().await; + let key = store_v42(&te, 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; + + // Lag only accountsdb's durable checkpoint in the closed store, preserving + // valid account content and its checksum. + corrupt(dirs.accounts.path(), 32, 2); + + // Drop the newest archive so recovery falls back to snapshot 1 (K = 10) and + // replays both a sealed successor and the unsealed ledger head. + std::fs::remove_file(&s2).unwrap(); + + let te2 = TestEngine::with(dirs, authority).await; + assert_eq!( + load_v42_data(&te2, 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 = store_v42(&te, 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. + store_v42(&te, 7, AccountMode::Delegated); + let s2 = commit_and_seal(&mut te, key, 20).await; + let (dirs, authority) = te.close().await; + + corrupt(dirs.accounts.path(), 8, 0xABAB_ABAB_ABAB_ABAB); + std::fs::remove_file(&s2).unwrap(); + + let result = time::timeout( + Duration::from_secs(4), + 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. A failed execution still counts on both durable sides without +// writing accounts. 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 them. +#[tokio::test(flavor = "multi_thread")] +async fn clean_restart_reopens_persisted_and_volatile_state() { + let mut te = TestEngine::new().await; + let key = store_v42(&te, 0, AccountMode::Delegated); + commit_and_seal(&mut te, key, 10).await; + te.execute(&[E::lit(20).compose(key, &[])]).await.unwrap(); + let failed = (E::lit(i64::MIN) - E::lit(1)).compose(key, &[]); + assert!( + te.execute(&[failed]).await.is_err(), + "overflow execution fails" + ); + assert_eq!( + load_v42_data(&te, key), + Some(20), + "failed execution writes no state" + ); + let direct = store_v42(&te, 7, AccountMode::Delegated); + let volatile = store_v42(&te, 8, AccountMode::ReadOnly); + let (dirs, authority) = te.close().await; + + let te2 = TestEngine::with(dirs, authority).await; + assert_eq!( + load_v42_data(&te2, key), + Some(20), + "persisted tip state reopened as-is" + ); + assert_eq!( + load_v42_data(&te2, direct), + Some(7), + "ledger-invisible account intact, so no snapshot was restored" + ); + assert_eq!( + load_v42_data(&te2, 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 00000000..d8247a11 --- /dev/null +++ b/engine/tests/security.rs @@ -0,0 +1,185 @@ +//! 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::{EngineError, testkit::TestEngine}; +use keeper::testkit::{load_v42_data, load_v42_lamports, signed_view, store_v42, v42_builder}; +use magic_root_interface::MagicRootInstruction; +use solana_account::{AccountFieldPatch, AccountMode}; +use solana_instruction::Instruction; +use solana_instruction_error::InstructionError; +use solana_keypair::Keypair; +use solana_pubkey::Pubkey; +use solana_signer::Signer; +use solana_transaction::TransactionError; +use v42_calculator_interface::builder::{Expr as E, transfer}; + +/// Complete v42 account replacement at an explicit non-default slot. +fn compose_v42_replacement(key: Pubkey, mode: AccountMode, slot: u64) -> Vec { + let account = v42_builder(0, mode).slot(slot).build(); + MagicRootInstruction::compose_account(key, account).unwrap() +} + +// 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 source: the transfer dirties both balance + // fields before the guard rejects the source account's engine mode. + let operand = store_v42(&te, 5, AccountMode::ReadOnly); + let recipient = store_v42(&te, 0, AccountMode::Delegated); + let operand_before = load_v42_lamports(&te, operand).expect("operand exists"); + let recipient_before = load_v42_lamports(&te, recipient).expect("recipient exists"); + assert_eq!( + te.execute(&[transfer(operand, recipient, 1)]).await, + Err(TransactionError::InvalidWritableAccount) + ); + assert_eq!( + load_v42_lamports(&te, operand).expect("operand remains"), + operand_before, + "immutable source debit discarded" + ); + assert_eq!( + load_v42_lamports(&te, recipient).expect("recipient remains"), + recipient_before, + "recipient credit rolled back with the transaction" + ); + + // 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(); + let acc = v42_builder(5, AccountMode::ReadOnly).build(); + te.accounts().store(&[(payer.pubkey(), acc)]).unwrap(); + let (_sig, view) = signed_view(&te, 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!( + load_v42_data(&te, payer.pubkey()), + Some(5), + "fee-payer write discarded" + ); + + // Positive control: a delegated (mutable) account commits normally. + let mutable = store_v42(&te, 0, AccountMode::Delegated); + assert!(te.execute(&[E::lit(9).compose(mutable, &[])]).await.is_ok()); + assert_eq!( + load_v42_data(&te, 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 key = Pubkey::new_unique(); + let mut ixs = compose_v42_replacement(key, AccountMode::ReadOnly, 1); + let post_finalize_idx = ixs.len(); + let post_finalize = MagicRootInstruction::PostFinalize(vec![E::lit(9).compose(key, &[])]); + ixs.push(post_finalize.compose(key).unwrap()); + assert_eq!( + te.execute(ixs.as_slice()).await, + Err(TransactionError::InstructionError( + post_finalize_idx as u8, + InstructionError::Immutable + )), + "MagicRoot's PostFinalize guard rejects the immutable writable account" + ); + assert!( + te.get_account(key).is_none(), + "the rejected creation commits nothing" + ); + + te.close().await; +} + +/// Proves PostFinalize rejects a recursive MagicRoot instruction of a delegated +/// account owned by an unrelated program and rolls back its creation. +#[tokio::test(flavor = "multi_thread")] +async fn post_finalize_rejects_magic_root_ix() { + let te = TestEngine::new().await; + + let key = Pubkey::new_unique(); + let owner = Pubkey::new_unique(); + let account = v42_builder(0, AccountMode::Delegated).owner(owner); + let patch = MagicRootInstruction::Patch(AccountFieldPatch::DataAt { + offset: 0, + data: 9_i64.to_le_bytes().to_vec(), + }) + .compose(key) + .unwrap(); + + let error = te + .account(key) + .create(account, Some(vec![patch])) + .await + .expect_err("PostFinalize rejects a recursive MagicRoot patch"); + assert!( + matches!( + error, + EngineError::TransactionExecution(TransactionError::InstructionError( + _, + InstructionError::CallDepth + )) + ), + "unexpected recursive invocation error: {error:?}" + ); + assert!( + te.get_account(key).is_none(), + "the rejected recursive action rolls back account creation" + ); + + 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(); + // A missing account starts as ReadOnly at slot zero. Advance the replacement + // slot so this test reaches the access guard rather than MagicRoot's + // duplicate-replacement guard. + let mut ixs = compose_v42_replacement(key, AccountMode::ReadOnly, 1); + // The foreign instruction that makes the whole transaction non-privileged. + ixs.push(E::lit(9).compose(key, &[])); + + assert_eq!( + te.execute(ixs.as_slice()).await, + Err(TransactionError::InvalidWritableAccount) + ); + assert!( + te.get_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 00000000..e0c3ee30 --- /dev/null +++ b/engine/tests/transactions.rs @@ -0,0 +1,249 @@ +//! 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 agave_transaction_view::MAX_STANDARD_TRANSACTION_SIZE; +use engine::testkit::TestEngine; +use keeper::testkit::{ + WireVersion, decode_v42, load_v42_data, load_v42_lamports, sign_versioned_instructions, + signed_view, store_v42, v42_padded_value, v42_sum, +}; +use nucleus::KB; +use solana_account::{AccountMode, ReadableAccount}; +use solana_instruction_error::InstructionError; +use solana_packet::PACKET_DATA_SIZE; +use solana_transaction::TransactionError; +use v42_calculator_interface::builder::{Expr as E, transfer}; + +// The Engine accepts the same standard wire formats produced by Solana clients +// on either side of the canonical packet boundary. The wide form reads every +// supplied account, while the batched form independently exercises instruction +// framing instead of relying on one large payload. +#[tokio::test(flavor = "multi_thread")] +async fn client_transaction_formats_execute_below_and_above_packet_limit() { + const WIDE_ACCOUNTS: usize = 32; + const BATCHED_INSTRUCTIONS: usize = 32; + const BATCHED_TERMS: usize = 16; + const FOUR_KIB: usize = 4 * KB; + + let te = TestEngine::new().await; + let operands: Vec<_> = + (0..WIDE_ACCOUNTS).map(|_| store_v42(&te, 1, AccountMode::Delegated)).collect(); + + for version in [WireVersion::Legacy, WireVersion::V0, WireVersion::V1] { + let output = store_v42(&te, 0, AccountMode::Ephemeral); + let (_, small) = sign_versioned_instructions( + te.signer(), + version, + [E::lit(42).compose(output, &[])], + te.blockhash(), + ); + assert!(small.len() < PACKET_DATA_SIZE,); + te.execute(small).await.expect("small v42 transaction succeeds"); + assert_eq!(load_v42_data(&te, output), Some(42)); + + let output = store_v42(&te, 0, AccountMode::Ephemeral); + let (_, wide) = sign_versioned_instructions( + te.signer(), + version, + [v42_sum(output, &operands)], + te.blockhash(), + ); + assert!(wide.len() > PACKET_DATA_SIZE); + assert!(wide.len() < MAX_STANDARD_TRANSACTION_SIZE); + te.execute(wide).await.expect("wide v42 transaction succeeds"); + assert_eq!(load_v42_data(&te, output), Some(WIDE_ACCOUNTS as i64)); + + let output = store_v42(&te, 0, AccountMode::Ephemeral); + let instructions: Vec<_> = (0..BATCHED_INSTRUCTIONS) + .map(|value| v42_padded_value(output, value as i64, BATCHED_TERMS)) + .collect(); + let (_, batched) = + sign_versioned_instructions(te.signer(), version, &instructions, te.blockhash()); + assert!(batched.len() > FOUR_KIB); + assert!(batched.len() < MAX_STANDARD_TRANSACTION_SIZE); + te.execute(batched).await.expect("batched v42 transaction succeeds"); + assert_eq!( + load_v42_data(&te, output), + Some((BATCHED_INSTRUCTIONS - 1) as i64) + ); + } + + te.close().await; +} + +// Simulation runs against live state through the dedicated simulation channel +// but must not commit; execution of the same transfer does. +#[tokio::test(flavor = "multi_thread")] +async fn simulate_does_not_commit_execute_does() { + let te = TestEngine::new().await; + let source = store_v42(&te, 0, AccountMode::Delegated); + let recipient = store_v42(&te, 0, AccountMode::Ephemeral); + let source_before = load_v42_lamports(&te, source).expect("source exists"); + let recipient_before = load_v42_lamports(&te, recipient).expect("recipient exists"); + let ixs = [transfer(source, recipient, 42)]; + + // 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_source) = executed + .loaded_transaction + .accounts + .iter() + .find(|(key, _)| *key == source) + .expect("simulation loaded the source account"); + assert_eq!( + simulated_source.lamports(), + source_before - 42, + "simulation debited its source copy" + ); + let (_, simulated_recipient) = executed + .loaded_transaction + .accounts + .iter() + .find(|(key, _)| *key == recipient) + .expect("simulation loaded the recipient account"); + assert_eq!( + simulated_recipient.lamports(), + recipient_before + 42, + "simulation credited its recipient copy" + ); + assert_eq!( + load_v42_lamports(&te, source).expect("source remains"), + source_before, + "simulation leaves the live source untouched" + ); + assert_eq!( + load_v42_lamports(&te, recipient).expect("recipient remains"), + recipient_before, + "simulation leaves the live recipient untouched" + ); + + assert!(te.execute(&ixs).await.is_ok(), "execution resolves"); + assert_eq!( + load_v42_lamports(&te, source).expect("source remains"), + source_before - 42, + "execution commits the source debit" + ); + assert_eq!( + load_v42_lamports(&te, recipient).expect("recipient remains"), + recipient_before + 42, + "execution commits the recipient credit" + ); + + 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 = store_v42(&te, 0, AccountMode::Ephemeral); + let ix = E::lit(value).cpi().compose(output, &[]); + let (signature, transaction) = signed_view(&te, 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 = store_v42(&te, 5, AccountMode::Ephemeral); + // MIN - 1 overflows the program's checked_sub before any write. + let ixs = [(E::lit(i64::MIN) - E::lit(1)).compose(output, &[])]; + + let error = te.execute(&ixs).await.expect_err("overflow 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!( + load_v42_data(&te, 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 = store_v42(&te, 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; +}