From b0fb4be7b905bbe1a4fcb5afd6ee9a39632a753a Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 23 Jun 2026 18:28:57 +0400 Subject: [PATCH 1/8] feat: adds top level engine crate for global orchestration --- Cargo.toml | 2 + README.md | 13 +- engine/Cargo.toml | 59 +++++ engine/README.md | 57 +++++ engine/src/accessor.rs | 112 +++++++++ engine/src/error.rs | 85 +++++++ engine/src/lib.rs | 200 ++++++++++++++++ engine/src/pacemaker.rs | 200 ++++++++++++++++ engine/src/testkit.rs | 177 ++++++++++++++ engine/src/transaction.rs | 122 ++++++++++ engine/tests/accounts.rs | 442 +++++++++++++++++++++++++++++++++++ engine/tests/builtins.rs | 62 +++++ engine/tests/recovery.rs | 134 +++++++++++ engine/tests/security.rs | 185 +++++++++++++++ engine/tests/transactions.rs | 249 ++++++++++++++++++++ 15 files changed, 2096 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/testkit.rs create mode 100644 engine/src/transaction.rs create mode 100644 engine/tests/accounts.rs create mode 100644 engine/tests/builtins.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 e57c074b..366eaf61 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,7 @@ [workspace] members = [ "accountsdb", + "engine", "keeper", "ledger", "nucleus", @@ -28,6 +29,7 @@ version = "0.1.0" [workspace.dependencies] accountsdb = { path = "accountsdb", package = "magicblock-accountsdb" } +engine = { path = "engine", package = "magicblock-engine" } keeper = { path = "keeper", package = "magicblock-keeper" } ledger = { path = "ledger", package = "magicblock-ledger" } magic-root-interface = { path = "programs/magic-root-interface" } diff --git a/README.md b/README.md index 01c073d3..d05c5ba5 100644 --- a/README.md +++ b/README.md @@ -214,7 +214,8 @@ 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(); @@ -232,7 +233,8 @@ engine 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 +243,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 diff --git a/engine/Cargo.toml b/engine/Cargo.toml new file mode 100644 index 00000000..0fa84946 --- /dev/null +++ b/engine/Cargo.toml @@ -0,0 +1,59 @@ +[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-instruction = { workspace = true } +solana-keypair = { workspace = true } +solana-message = { workspace = true } +solana-program-runtime = { workspace = true } +solana-pubkey = { 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..c80ca3b7 --- /dev/null +++ b/engine/README.md @@ -0,0 +1,57 @@ +# `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 or its +sealed superblock trails the retained ledger. The latter also covers snapshots +staged by a replication follower. + +If accountsdb then trails the ledger tip, `Engine::new` replays retained entries +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. + +Internal pacing appends one reset marker at the upcoming slot and clears +chain-mirrored volatile accounts before the pacemaker task starts. Internal +system accounts remain available. Replicas use external pacing and retain +restored volatile state. 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..563701d1 --- /dev/null +++ b/engine/src/accessor.rs @@ -0,0 +1,112 @@ +//! Account- and transaction-scoped operation facades. + +use std::{sync::atomic::Ordering, time::Duration}; + +use keeper::{ExecutionRecord, TransactionView}; +use magic_root_interface::MagicRootInstruction; +use processor::{SequencerMessage, Simulation, SimulatorMessage}; +use solana_account::{AccountFieldPatch, OwnedAccount}; +use solana_instruction::Instruction; +use solana_pubkey::Pubkey; +use solana_transaction::TransactionResult; +use tokio::time; + +use crate::{Engine, 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 + } + + /// Applies the given field patches to the account. + pub async fn patch(&self, patches: Vec) -> Result<()> { + let mut instructions = Vec::with_capacity(patches.len()); + for patch in patches { + let ix = MagicRootInstruction::Patch(patch).compose(self.pubkey)?; + instructions.push(ix); + } + self.execute(instructions).await + } + + /// Closes the account. + pub async fn delete(&self) -> Result<()> { + let instructions = vec![MagicRootInstruction::Delete.compose(self.pubkey)?]; + self.execute(instructions).await + } + + /// Composes the instructions into a signed engine transaction, executes it, + /// and flattens the committed transaction result into the engine error type. + async fn execute(&self, instructions: Vec) -> Result<()> { + let txn = 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. + pub async fn execute(self) -> Result> { + if self.engine.terminating.load(Ordering::Acquire) { + Err("engine is shutting down".to_string())?; + } + let signature = self.transaction.signatures()[0]; + let msg = SequencerMessage::Transaction(self.transaction); + let mut rx = self.engine.transactions().subscribe_signature(signature).await; + self.engine.sequencer.send(msg).await?; + let status = time::timeout(EXECUTION_TIMEOUT, rx.recv()) + .await + .map_err(|e| e.to_string())? + .map_err(|e| e.to_string())?; + Ok(status.result) + } + + /// Submits `transaction` for execution without awaiting its result. + pub async fn schedule(self) -> Result<()> { + if self.engine.terminating.load(Ordering::Acquire) { + Err("engine is shutting down".to_string())?; + } + let msg = SequencerMessage::Transaction(self.transaction); + self.engine.sequencer.send(msg).await.map_err(Into::into) + } + + /// Simulates `transaction` against current state without committing it. + pub async fn simulate(self) -> Result> { + let (response, rx) = oneshot::channel(); + let msg = SimulatorMessage::Transaction(Simulation { + transaction: self.transaction, + response, + }); + self.engine.sequencer.simulation.send(msg).await?; + rx.await.map_err(Into::into) + } +} diff --git a/engine/src/error.rs b/engine/src/error.rs new file mode 100644 index 00000000..2db7ef3b --- /dev/null +++ b/engine/src/error.rs @@ -0,0 +1,85 @@ +//! 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), + /// 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..b0d4655d --- /dev/null +++ b/engine/src/lib.rs @@ -0,0 +1,200 @@ +#![doc = include_str!("../README.md")] + +use std::{ + sync::{ + Arc, + atomic::{AtomicBool, Ordering}, + }, + time::Instant, +}; + +use derive_more::Deref; +use keeper::{Keeper, builder::KeeperBuilder, error::KeeperError}; +use ledger::schema::OwnedBlockestoreEntry; +use magic_root_program::entrypoint::MagicRootEntrypoint; +use nucleus::{ + runtime::{self, BarrierHandle, SequencerHandle}, + shutdown::{Service, ShutdownManager, ShutdownReason}, +}; +use processor::{SequencerMessage, sequencer::Sequencer}; +use solana_program_runtime::{ + loaded_programs::{ProgramCache, ProgramCacheEntry}, + solana_sbpf::program::BuiltinFunctionDefinition, +}; +use solana_pubkey::Pubkey; +use 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, + ), + ); + + 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: OwnedBlockestoreEntry) -> Result<()> { + match entry { + OwnedBlockestoreEntry::Transaction(txn) => self.transaction(txn)?.schedule().await?, + OwnedBlockestoreEntry::Block(block) => { + self.sequencer.send(SequencerMessage::Block(block)).await?; + } + OwnedBlockestoreEntry::Superblock(expected) => { + let _guard = self.barrier().await?; + self.accounts().set_superblock(expected.id); + self.sync(false)?; + let observed = self.superblocks().sealed(); + if observed != expected { + error!(?observed, ?expected, "state mismatch; aborting replay"); + Err(ReplayError::StateMismatch)?; + } + } + OwnedBlockestoreEntry::Reset(slot) => { + let _guard = self.barrier().await?; + self.reset(slot)?; + } + }; + Ok(()) + } + + /// Rebuilds state through a temporary replay sequencer when accountsdb trails + /// the retained ledger, then stops every temporary service before returning. + async fn try_replay(state: &Arc, cache: &Arc, cpus: usize) -> Result<()> { + let timer = Instant::now(); + let Some(mut replayer) = state.replay().await? else { + return Ok(()); + }; + let mut shutdown = ShutdownManager::default(); + let mut sh = shutdown.handle(Service::LedgerReplayer); + let (service, sequencer) = + Sequencer::new(cpus, state.clone(), cache.clone(), &mut shutdown, true)?; + service.spawn()?; + let engine = Self { + state: state.clone(), + sequencer, + terminating: Default::default(), + }; + while let Some(entry) = replayer.rx.recv().await { + engine.replay(entry).await?; + } + replayer + .response + .recv_timeout() + .await + .map_err(ReplayError::from)? + .map_err(ReplayError::from)?; + + drop(engine.barrier().await?); + engine.sync(false)?; + + 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..a2127628 --- /dev/null +++ b/engine/src/pacemaker.rs @@ -0,0 +1,200 @@ +//! 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 { + tokio::select! { + biased; + _ = shutdown.signalled() => { + break Ok(()); + } + success = self.pace() => { + if !*success.as_ref().unwrap_or(&false) { + break success.map(|_| ()); + } + } + } + }; + res = if let Pacer::Internal(ref mut t) = self.pacer { + // 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); + } + } + + /// Emits one block boundary and seals a superblock when the emitted slot is + /// on the configured interval. + async fn pace(&mut self) -> Result { + let (block, submission) = match &mut self.pacer { + Pacer::Internal(t) => { + t.ticker.tick().await; + (t.block(), None) + } + Pacer::External(rx) => { + let Some(msg) = rx.recv().await else { + return Ok(false); + }; + (msg.block, Some(msg.submitted)) + } + }; + let result = self.handle(block).await.map(|()| true); + if let Some(submission) = submission + && result.is_ok() + { + let _ = submission.send(()); + } + result + } + + /// Advances the execution and simulation environments to `block`, sealing a + /// superblock when the slot lands on the configured interval. + /// + /// The seal is taken behind a barrier and runs synchronously: it exports an + /// accountsdb snapshot, which is only coherent while no store operation can + /// race it. Holding the boundary here is what buys that exclusivity, at the + /// cost of stalling block production until the seal completes. + async fn handle(&self, block: Block) -> Result<()> { + self.sequencer.send(SequencerMessage::Block(block)).await?; + self.sequencer.simulation.send(SimulatorMessage::Block(block)).await?; + if block.slot.is_multiple_of(self.superblock.get()) { + let _guard = self.barrier().await?; + self.finalize_superblock()?; + } + Ok(()) + } +} diff --git a/engine/src/testkit.rs b/engine/src/testkit.rs new file mode 100644 index 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..662933ec --- /dev/null +++ b/engine/tests/accounts.rs @@ -0,0 +1,442 @@ +//! Account CRUD through the MagicRoot builtin — the privileged mutation path +//! exposed by `AccountAccessor`. This path is untested below the engine: it needs +//! the always-on MagicRoot builtin plus the executor's per-thread authority +//! (MagicRoot authorizes the transaction's fee payer against it). Asserts the +//! create/patch/update/delete round-trip and the sponsor-balance invariant, and +//! that post-finalize actions actually run. +#![cfg(test)] + +use engine::{Engine, EngineError, testkit::TestEngine}; +use keeper::testkit::{ + V42_ID, load_v42_data, load_v42_lamports, patterned_bytes, store_v42, v42_builder, +}; +use solana_account::{ + AccountBuilder, AccountFieldPatch, 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; `patch` mutates individual fields in place; +// `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" + ); + + // patch the data, growing the buffer (exercises the borrowed-image + // copy-on-write promotion), leaving other fields untouched. + te.account(key) + .patch(vec![AccountFieldPatch::DataAt { + offset: 0, + data: vec![9; 16], + }]) + .await + .unwrap(); + let acc = te.get_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(); + te.account(key).patch(vec![AccountFieldPatch::Owner(new_owner)]).await.unwrap(); + assert_eq!( + te.get_account(key).expect("still exists").owner(), + &new_owner + ); + + // update overwrites the existing account in place: same-length data (the + // patch sequence replaces the exact data length) and identical lamports. + // The owner reverting to `owner` proves the earlier patch was overwritten. + // 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..e439c8ed --- /dev/null +++ b/engine/tests/recovery.rs @@ -0,0 +1,134 @@ +//! Full-engine replay recovery — the engine's most distinctive orchestration. +//! After an accountsdb corruption the keeper restores an older archived snapshot, +//! leaving durable state behind the ledger tip; the engine then spins a temporary +//! replay sequencer to re-execute the retained ledger entries and rebuild the +//! missing state, checksum-verified at each sealed superblock. Nothing below the +//! engine wires this end to end. Covered here: the healthy restart that must not +//! recover, the replay that crosses a sealed checksum and succeeds, and the +//! replay that diverges from one and must refuse to start. +#![cfg(test)] + +use std::{path::PathBuf, time::Duration}; + +use engine::{EngineError, ReplayError, testkit::TestEngine}; +use keeper::testkit::{corrupt, 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_corruption() { + 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; + + // Corrupt the closed accountsdb and drop the newest archive: the reopen falls + // back to snapshot 1 (K = 10) and replays everything after it. Corruption must + // follow the close, whose flush would otherwise republish a valid checksum. + corrupt(dirs.accounts.path()); + std::fs::remove_file(&s2).unwrap(); + + // Guarded by a timeout so a replay regression fails fast instead of hanging. + let te2 = 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()); + 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. The direct-stored delegated account exists in neither snapshots +// nor ledger, while the read-only account exists only in the volatile dump; the +// post-seal transaction write pins the persisted tip alongside both probes. +#[tokio::test(flavor = "multi_thread")] +async fn clean_restart_reopens_persisted_and_volatile_state() { + let mut te = TestEngine::new().await; + let key = store_v42(&te, 0, AccountMode::Delegated); + commit_and_seal(&mut te, key, 10).await; + te.execute(&[E::lit(20).compose(key, &[])]).await.unwrap(); + 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..edf9a565 --- /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::UnsupportedProgramId + )) + ), + "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; +} From 3ac638e0f11d2d4f13952e5e51ec5bbcae558dcb Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Thu, 6 Aug 2026 23:52:39 +0400 Subject: [PATCH 2/8] fix: update type name after typo fix --- engine/src/lib.rs | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/engine/src/lib.rs b/engine/src/lib.rs index b0d4655d..1ef6cfe9 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -10,7 +10,7 @@ use std::{ use derive_more::Deref; use keeper::{Keeper, builder::KeeperBuilder, error::KeeperError}; -use ledger::schema::OwnedBlockestoreEntry; +use ledger::schema::OwnedBlockstoreEntry; use magic_root_program::entrypoint::MagicRootEntrypoint; use nucleus::{ runtime::{self, BarrierHandle, SequencerHandle}, @@ -137,13 +137,13 @@ impl Engine { /// Seal and reset entries quiesce execution before touching shared state; /// a reconstructed seal whose checksum differs returns /// [`ReplayError::StateMismatch`]. - pub async fn replay(&self, entry: OwnedBlockestoreEntry) -> Result<()> { + pub async fn replay(&self, entry: OwnedBlockstoreEntry) -> Result<()> { match entry { - OwnedBlockestoreEntry::Transaction(txn) => self.transaction(txn)?.schedule().await?, - OwnedBlockestoreEntry::Block(block) => { + OwnedBlockstoreEntry::Transaction(txn) => self.transaction(txn)?.schedule().await?, + OwnedBlockstoreEntry::Block(block) => { self.sequencer.send(SequencerMessage::Block(block)).await?; } - OwnedBlockestoreEntry::Superblock(expected) => { + OwnedBlockstoreEntry::Superblock(expected) => { let _guard = self.barrier().await?; self.accounts().set_superblock(expected.id); self.sync(false)?; @@ -153,7 +153,7 @@ impl Engine { Err(ReplayError::StateMismatch)?; } } - OwnedBlockestoreEntry::Reset(slot) => { + OwnedBlockstoreEntry::Reset(slot) => { let _guard = self.barrier().await?; self.reset(slot)?; } From 2be699eb4bd7afade60ce4805b584cfe1972e1a1 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Fri, 7 Aug 2026 19:35:14 +0400 Subject: [PATCH 3/8] fix: verify transaction count after replay --- engine/README.md | 8 +++++--- engine/src/lib.rs | 9 +++++++++ engine/tests/recovery.rs | 32 ++++++++++++++++++++++---------- 3 files changed, 36 insertions(+), 13 deletions(-) diff --git a/engine/README.md b/engine/README.md index c80ca3b7..4d121c6d 100644 --- a/engine/README.md +++ b/engine/README.md @@ -26,15 +26,17 @@ 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 or its -sealed superblock trails the retained ledger. The latter also covers snapshots +Keeper restores an accountsdb snapshot when the active store is corrupt, its +sealed superblock trails the retained ledger, or its committed transaction count +differs from the ledger's durable count. Superblock lag also covers snapshots staged by a replication follower. 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. +opens without replay. After the final replay sync, persistent transaction-count +divergence also returns `ReplayError::StateMismatch`. Internal pacing appends one reset marker at the upcoming slot and clears chain-mirrored volatile accounts before the pacemaker task starts. Internal diff --git a/engine/src/lib.rs b/engine/src/lib.rs index 1ef6cfe9..e68b5bf7 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -190,6 +190,15 @@ impl Engine { 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"); diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs index e439c8ed..ffeeca9e 100644 --- a/engine/tests/recovery.rs +++ b/engine/tests/recovery.rs @@ -1,5 +1,5 @@ //! Full-engine replay recovery — the engine's most distinctive orchestration. -//! After an accountsdb corruption the keeper restores an older archived snapshot, +//! 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 @@ -11,7 +11,7 @@ use std::{path::PathBuf, time::Duration}; use engine::{EngineError, ReplayError, testkit::TestEngine}; -use keeper::testkit::{corrupt, load_v42_data, store_v42}; +use keeper::testkit::{advance_transactions, corrupt, load_v42_data, store_v42}; use nucleus::ledger::ACCOUNTSDB_SNAPSHOT_FILE; use solana_account::AccountMode; use solana_pubkey::Pubkey; @@ -30,7 +30,7 @@ async fn commit_and_seal(te: &mut TestEngine, key: Pubkey, value: i64) -> PathBu // 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() { +async fn replay_rebuilds_state_after_counter_mismatch() { let mut te = TestEngine::new().await; let key = store_v42(&te, 0, AccountMode::Delegated); @@ -46,12 +46,13 @@ async fn replay_rebuilds_state_after_corruption() { 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; + // Advance only accountsdb's count. The mismatch must restore a snapshot even + // though account content and its checksum remain valid. + advance_transactions(&te); let (dirs, authority) = te.close().await; - // Corrupt the closed accountsdb and drop the newest archive: the reopen falls - // back to snapshot 1 (K = 10) and replays everything after it. Corruption must - // follow the close, whose flush would otherwise republish a valid checksum. - corrupt(dirs.accounts.path()); + // 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(); // Guarded by a timeout so a replay regression fails fast instead of hanging. @@ -100,15 +101,26 @@ async fn replay_aborts_on_checksum_mismatch() { } // A healthy restart opens persisted state as-is and restores the clean-shutdown -// volatile dump. The direct-stored delegated account exists in neither snapshots -// nor ledger, while the read-only account exists only in the volatile dump; the -// post-seal transaction write pins the persisted tip alongside both probes. +// 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; From bef65b7fb4ad58ff25653eedb0c8b5653f479048 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Sat, 8 Aug 2026 01:33:24 +0400 Subject: [PATCH 4/8] test: cover recovery from accountsdb count lag --- engine/README.md | 11 +++++++---- engine/tests/recovery.rs | 13 +++++++------ 2 files changed, 14 insertions(+), 10 deletions(-) diff --git a/engine/README.md b/engine/README.md index 4d121c6d..685b2639 100644 --- a/engine/README.md +++ b/engine/README.md @@ -28,15 +28,18 @@ and instruction-data lengths that cannot be represented by the V1 wire fields. Keeper restores an accountsdb snapshot when the active store is corrupt, its sealed superblock trails the retained ledger, or its committed transaction count -differs from the ledger's durable count. Superblock lag also covers snapshots -staged by a replication follower. +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. After the final replay sync, persistent transaction-count -divergence also returns `ReplayError::StateMismatch`. +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 upcoming slot and clears chain-mirrored volatile accounts before the pacemaker task starts. Internal diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs index ffeeca9e..0c04eee2 100644 --- a/engine/tests/recovery.rs +++ b/engine/tests/recovery.rs @@ -11,7 +11,7 @@ use std::{path::PathBuf, time::Duration}; use engine::{EngineError, ReplayError, testkit::TestEngine}; -use keeper::testkit::{advance_transactions, corrupt, load_v42_data, store_v42}; +use keeper::testkit::{corrupt, load_v42_data, store_v42}; use nucleus::ledger::ACCOUNTSDB_SNAPSHOT_FILE; use solana_account::AccountMode; use solana_pubkey::Pubkey; @@ -30,7 +30,7 @@ async fn commit_and_seal(te: &mut TestEngine, key: Pubkey, value: i64) -> PathBu // 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_mismatch() { +async fn replay_rebuilds_state_after_counter_lag() { let mut te = TestEngine::new().await; let key = store_v42(&te, 0, AccountMode::Delegated); @@ -46,11 +46,12 @@ async fn replay_rebuilds_state_after_counter_mismatch() { 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; - // Advance only accountsdb's count. The mismatch must restore a snapshot even - // though account content and its checksum remain valid. - advance_transactions(&te); 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(); @@ -84,7 +85,7 @@ async fn replay_aborts_on_checksum_mismatch() { let s2 = commit_and_seal(&mut te, key, 20).await; let (dirs, authority) = te.close().await; - corrupt(dirs.accounts.path()); + corrupt(dirs.accounts.path(), 8, 0xABAB_ABAB_ABAB_ABAB); std::fs::remove_file(&s2).unwrap(); let result = time::timeout( From a47acdb27902cffea833fb0a6fe09d0a9fb3ec79 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Mon, 10 Aug 2026 18:15:49 +0400 Subject: [PATCH 5/8] fix: address engine review feedback --- engine/README.md | 2 +- engine/src/accessor.rs | 12 +++++++---- engine/src/error.rs | 6 ++++++ engine/src/lib.rs | 3 +++ engine/src/pacemaker.rs | 44 ++++++++++++++++------------------------ engine/tests/recovery.rs | 1 - engine/tests/security.rs | 2 +- 7 files changed, 36 insertions(+), 34 deletions(-) diff --git a/engine/README.md b/engine/README.md index 685b2639..c68ac2e4 100644 --- a/engine/README.md +++ b/engine/README.md @@ -41,7 +41,7 @@ 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 upcoming slot and clears +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; diff --git a/engine/src/accessor.rs b/engine/src/accessor.rs index 563701d1..91646fa0 100644 --- a/engine/src/accessor.rs +++ b/engine/src/accessor.rs @@ -11,7 +11,7 @@ use solana_pubkey::Pubkey; use solana_transaction::TransactionResult; use tokio::time; -use crate::{Engine, error::Result, transaction}; +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); @@ -75,9 +75,10 @@ impl AccountAccessor<'_> { 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) { - Err("engine is shutting down".to_string())?; + return Err(EngineError::ShuttingDown); } let signature = self.transaction.signatures()[0]; let msg = SequencerMessage::Transaction(self.transaction); @@ -85,7 +86,7 @@ impl TransactionAccessor<'_> { self.engine.sequencer.send(msg).await?; let status = time::timeout(EXECUTION_TIMEOUT, rx.recv()) .await - .map_err(|e| e.to_string())? + .map_err(|_| EngineError::TransactionTimeout)? .map_err(|e| e.to_string())?; Ok(status.result) } @@ -93,7 +94,7 @@ impl TransactionAccessor<'_> { /// Submits `transaction` for execution without awaiting its result. pub async fn schedule(self) -> Result<()> { if self.engine.terminating.load(Ordering::Acquire) { - Err("engine is shutting down".to_string())?; + return Err(EngineError::ShuttingDown); } let msg = SequencerMessage::Transaction(self.transaction); self.engine.sequencer.send(msg).await.map_err(Into::into) @@ -101,6 +102,9 @@ impl TransactionAccessor<'_> { /// 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, diff --git a/engine/src/error.rs b/engine/src/error.rs index 2db7ef3b..0e297bee 100644 --- a/engine/src/error.rs +++ b/engine/src/error.rs @@ -28,6 +28,12 @@ pub enum EngineError { /// 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), diff --git a/engine/src/lib.rs b/engine/src/lib.rs index e68b5bf7..43a1d56a 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -145,11 +145,14 @@ impl Engine { } 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)?; } } diff --git a/engine/src/pacemaker.rs b/engine/src/pacemaker.rs index a2127628..dbea204c 100644 --- a/engine/src/pacemaker.rs +++ b/engine/src/pacemaker.rs @@ -128,16 +128,19 @@ impl PaceMaker { /// handshake. async fn run(mut self, mut shutdown: ShutdownHandle) { let mut res = loop { - tokio::select! { + let next = tokio::select! { biased; - _ = shutdown.signalled() => { - break Ok(()); - } - success = self.pace() => { - if !*success.as_ref().unwrap_or(&false) { - break success.map(|_| ()); - } - } + _ = 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 { @@ -157,28 +160,15 @@ impl PaceMaker { } } - /// 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 { + /// 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; - (t.block(), None) - } - Pacer::External(rx) => { - let Some(msg) = rx.recv().await else { - return Ok(false); - }; - (msg.block, Some(msg.submitted)) + Some((t.block(), None)) } - }; - let result = self.handle(block).await.map(|()| true); - if let Some(submission) = submission - && result.is_ok() - { - let _ = submission.send(()); + Pacer::External(rx) => rx.recv().await.map(|msg| (msg.block, Some(msg.submitted))), } - result } /// Advances the execution and simulation environments to `block`, sealing a diff --git a/engine/tests/recovery.rs b/engine/tests/recovery.rs index 0c04eee2..9130b995 100644 --- a/engine/tests/recovery.rs +++ b/engine/tests/recovery.rs @@ -56,7 +56,6 @@ async fn replay_rebuilds_state_after_counter_lag() { // replays both a sealed successor and the unsealed ledger head. std::fs::remove_file(&s2).unwrap(); - // Guarded by a timeout so a replay regression fails fast instead of hanging. let te2 = TestEngine::with(dirs, authority).await; assert_eq!( load_v42_data(&te2, key), diff --git a/engine/tests/security.rs b/engine/tests/security.rs index edf9a565..d8247a11 100644 --- a/engine/tests/security.rs +++ b/engine/tests/security.rs @@ -142,7 +142,7 @@ async fn post_finalize_rejects_magic_root_ix() { error, EngineError::TransactionExecution(TransactionError::InstructionError( _, - InstructionError::UnsupportedProgramId + InstructionError::CallDepth )) ), "unexpected recursive invocation error: {error:?}" From 33a0d2ae6652caeef9a4ebc2556df8c4708dc056 Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Mon, 10 Aug 2026 19:08:35 +0400 Subject: [PATCH 6/8] fix: register compute budget builtin --- Cargo.toml | 1 + engine/Cargo.toml | 2 ++ engine/src/lib.rs | 8 ++++++++ 3 files changed, 11 insertions(+) diff --git a/Cargo.toml b/Cargo.toml index 366eaf61..56168ddd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -92,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/engine/Cargo.toml b/engine/Cargo.toml index 0fa84946..00dc5f68 100644 --- a/engine/Cargo.toml +++ b/engine/Cargo.toml @@ -33,11 +33,13 @@ 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"] } diff --git a/engine/src/lib.rs b/engine/src/lib.rs index 43a1d56a..954737d0 100644 --- a/engine/src/lib.rs +++ b/engine/src/lib.rs @@ -17,6 +17,7 @@ use nucleus::{ 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, @@ -78,6 +79,13 @@ impl Engine { 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); From baf565ced4bebf2a95eeee04de777a43ebef88cb Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 11 Aug 2026 13:14:27 +0400 Subject: [PATCH 7/8] refactor: adapt engine subscription receivers --- README.md | 13 ++++++++----- engine/src/accessor.rs | 4 ++-- 2 files changed, 10 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index d05c5ba5..044f1334 100644 --- a/README.md +++ b/README.md @@ -296,20 +296,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/src/accessor.rs b/engine/src/accessor.rs index 91646fa0..2735106a 100644 --- a/engine/src/accessor.rs +++ b/engine/src/accessor.rs @@ -82,9 +82,9 @@ impl TransactionAccessor<'_> { } let signature = self.transaction.signatures()[0]; let msg = SequencerMessage::Transaction(self.transaction); - let mut rx = self.engine.transactions().subscribe_signature(signature).await; + let rx = self.engine.transactions().subscribe_signature(signature).await; self.engine.sequencer.send(msg).await?; - let status = time::timeout(EXECUTION_TIMEOUT, rx.recv()) + let status = time::timeout(EXECUTION_TIMEOUT, rx) .await .map_err(|_| EngineError::TransactionTimeout)? .map_err(|e| e.to_string())?; From 1e063a02f411ee3ba40154cc1d89373161d034ec Mon Sep 17 00:00:00 2001 From: Babur Makhmudov Date: Tue, 11 Aug 2026 16:09:30 +0400 Subject: [PATCH 8/8] refactor: require complete account replacements --- README.md | 17 ++++------------- engine/src/accessor.rs | 12 +----------- engine/tests/accounts.rs | 32 ++++---------------------------- 3 files changed, 9 insertions(+), 52 deletions(-) diff --git a/README.md b/README.md index 044f1334..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(); @@ -220,15 +220,6 @@ let account = AccountBuilder::default() .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) diff --git a/engine/src/accessor.rs b/engine/src/accessor.rs index 2735106a..87116c34 100644 --- a/engine/src/accessor.rs +++ b/engine/src/accessor.rs @@ -5,7 +5,7 @@ use std::{sync::atomic::Ordering, time::Duration}; use keeper::{ExecutionRecord, TransactionView}; use magic_root_interface::MagicRootInstruction; use processor::{SequencerMessage, Simulation, SimulatorMessage}; -use solana_account::{AccountFieldPatch, OwnedAccount}; +use solana_account::OwnedAccount; use solana_instruction::Instruction; use solana_pubkey::Pubkey; use solana_transaction::TransactionResult; @@ -49,16 +49,6 @@ impl AccountAccessor<'_> { 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)?]; diff --git a/engine/tests/accounts.rs b/engine/tests/accounts.rs index 662933ec..3e17cf96 100644 --- a/engine/tests/accounts.rs +++ b/engine/tests/accounts.rs @@ -2,7 +2,7 @@ //! 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 +//! create/update/delete round-trip and the sponsor-balance invariant, and //! that post-finalize actions actually run. #![cfg(test)] @@ -10,9 +10,7 @@ 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, AccountFieldPatch, AccountMode, OwnedAccount, ReadableAccount, -}; +use solana_account::{AccountBuilder, AccountMode, OwnedAccount, ReadableAccount}; use solana_instruction_error::InstructionError; use solana_pubkey::Pubkey; use solana_system_interface::MAX_PERMITTED_DATA_LENGTH; @@ -88,8 +86,8 @@ fn assert_invalid_mode_transition(error: EngineError) { // The full lifecycle. `create` materializes a fresh account by patching every // non-flag field, balancing lamport patches against the authority, then -// finalizing its flags; `patch` mutates individual fields in place; -// `update` overwrites an existing account or materializes a fresh key; and +// 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() { @@ -120,30 +118,8 @@ async fn account_crud_lifecycle() { "the lamport patch 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. - te.account(key) - .patch(vec![AccountFieldPatch::DataAt { - offset: 0, - data: vec![9; 16], - }]) - .await - .unwrap(); - let acc = te.get_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(); - te.account(key).patch(vec![AccountFieldPatch::Owner(new_owner)]).await.unwrap(); - assert_eq!( - te.get_account(key).expect("still exists").owner(), - &new_owner - ); - // update overwrites the existing account in place: same-length data (the // patch sequence replaces the exact data length) and identical lamports. - // The owner reverting to `owner` proves the earlier patch was overwritten. // Read-only accounts remain replaceable after finalization. te.account(key) .update(account(owner, vec![5; 16], AccountMode::ReadOnly, 11))