feat: implement transaction processor crate

Cargo.toml

@@ -4,6 +4,7 @@ members = [
     "keeper",
     "ledger",
     "nucleus",
+    "processor",
     "programs/magic-root-interface",
     "programs/magic-root-program",
     "programs/v42-calculator-interface",
@@ -33,6 +34,9 @@ ledger = { path = "ledger", package = "magicblock-ledger" }
 magic-root-interface = { path = "programs/magic-root-interface" }
 magic-root-program = { path = "programs/magic-root-program" }
 nucleus = { path = "nucleus", package = "magicblock-engine-nucleus" }
+processor = { path = "processor", package = "magicblock-processor" }
+v42-calculator-interface = { path = "programs/v42-calculator-interface", default-features = false }
+
 solana-account = { path = "solana/account" }
 solana-program-runtime = { path = "solana/program-runtime" }
 solana-svm = { path = "solana/svm" }

processor/Cargo.toml

@@ -0,0 +1,53 @@
+[package]
+name = "magicblock-processor"
+
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[lib]
+name = "processor"
+
+[dependencies]
+accountsdb = { workspace = true }
+keeper = { workspace = true }
+nucleus = { workspace = true, features = ["runtime"] }
+
+ahash = { workspace = true }
+blake3 = { workspace = true }
+derive_more = { workspace = true, features = ["deref", "deref_mut", "from"] }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt", "sync", "time"] }
+tracing = { workspace = true }
+
+agave-feature-set = { workspace = true }
+agave-precompiles = { workspace = true, features = ["agave-unstable-api"] }
+agave-syscalls = { workspace = true, features = ["agave-unstable-api"] }
+agave-transaction-view = { workspace = true }
+solana-account = { workspace = true }
+solana-compute-budget-instruction = { workspace = true, features = ["agave-unstable-api"] }
+solana-hash = { workspace = true }
+solana-precompile-error = { workspace = true }
+solana-program-runtime = { workspace = true }
+solana-pubkey = { workspace = true }
+solana-signature = { workspace = true }
+solana-svm = { workspace = true }
+solana-svm-transaction = { workspace = true }
+solana-sysvar = { workspace = true }
+solana-transaction-error = { workspace = true }
+
+[dev-dependencies]
+keeper = { workspace = true, features = ["testkit"] }
+v42-calculator-interface = { workspace = true, features = ["builder"] }
+
+oneshot = { workspace = true }
+solana-instruction = { workspace = true }
+solana-keypair = { workspace = true }
+solana-sdk-ids = { workspace = true }
+
+[lints]
+workspace = true

processor/README.md

@@ -0,0 +1,26 @@
+# `magicblock-processor`
+
+The processor schedules transactions across a fixed pool of SVM executors and
+commits their results through keeper.
+
+The sequencer tracks each static account key with executor-holder bits and an
+exclusive-write bit. Transactions with non-conflicting read/write sets run in
+parallel. Conflicting transactions queue behind the executor holding the
+blocking account and are retried when that executor becomes available.
+
+Block finalization drains executor work before publishing the ledger boundary,
+so every transaction's execution metadata precedes the block that contains it.
+
+## Quiescence
+
+The sequencer barrier drains all executor work, acknowledges the caller, and
+holds new execution until its guard is released. Engine uses the barrier for
+coherent superblock snapshots, replay seal checks, replication handshakes, and
+shutdown.
+
+## Simulation
+
+Simulation has a separate worker and SVM context. It resolves a transaction,
+loads owned account copies, executes against the current block environment, and
+returns an `ExecutionRecord` without appending to the ledger or storing account
+changes.

processor/src/callback.rs

@@ -0,0 +1,54 @@
+use agave_feature_set::FeatureSet;
+use nucleus::Slot;
+use solana_account::AccountSharedData;
+use solana_precompile_error::PrecompileError;
+use solana_pubkey::Pubkey;
+use solana_svm::transaction_processing_callback::{
+    InvokeContextCallback, TransactionProcessingCallback,
+};
+use tracing::error;
+
+use accountsdb::AccountLoader;
+
+/// Bridges the SVM to keeper-backed account loads.
+///
+/// When `LOAD_OWNED` is set, loaded accounts are copied to owned data; this is
+/// used by simulation, which must always operate on owned account copies.
+pub(crate) struct SVMCallback<'a, const LOAD_OWNED: bool> {
+    /// Reads accounts from the engine's accounts store.
+    pub(crate) loader: AccountLoader<'a>,
+    /// Active feature set governing precompiles and runtime behavior.
+    pub(crate) featureset: &'a FeatureSet,
+}
+
+impl<'a, const LO: bool> InvokeContextCallback for SVMCallback<'a, LO> {
+    fn is_precompile(&self, program_id: &Pubkey) -> bool {
+        agave_precompiles::is_precompile(program_id, |id| self.featureset.is_active(id))
+    }
+
+    fn process_precompile(
+        &self,
+        program_id: &Pubkey,
+        data: &[u8],
+        instruction_datas: Vec<&[u8]>,
+    ) -> Result<(), PrecompileError> {
+        agave_precompiles::get_precompile(program_id, |id| self.featureset.is_active(id))
+            .ok_or(PrecompileError::InvalidPublicKey)
+            .and_then(|p| p.verify(data, &instruction_datas, self.featureset))
+    }
+}
+
+impl<'a, const LOAD_OWNED: bool> TransactionProcessingCallback for SVMCallback<'a, LOAD_OWNED> {
+    fn get_account_shared_data(&self, pubkey: &Pubkey) -> Option<(AccountSharedData, Slot)> {
+        self.loader
+            .load(pubkey)
+            .inspect_err(|error| error!(?error, "accountsdb load error"))
+            .unwrap_or_default()
+            .map(|mut acc| {
+                if LOAD_OWNED {
+                    acc = acc.owned().into();
+                }
+                (acc, 0)
+            })
+    }
+}

processor/src/error.rs

@@ -0,0 +1,27 @@
+//! Error type for the transaction processor.
+
+use std::io;
+
+use derive_more::From;
+use keeper::error::KeeperError;
+use nucleus::shutdown::Service;
+
+/// Convenience result alias for processor operations.
+pub type Result<T> = std::result::Result<T, ProcessorError>;
+
+/// Failures raised while scheduling or executing transactions.
+#[derive(Debug, thiserror::Error, From)]
+pub enum ProcessorError {
+    /// An I/O error, e.g. while spawning a worker thread or building a runtime.
+    #[error("i/o error: {0}")]
+    Io(#[source] io::Error),
+    /// A failure surfaced by the underlying keeper-backed state.
+    #[error("state error: {0}")]
+    State(#[source] KeeperError),
+    /// A background service is no longer reachable, so work could not be handed off.
+    #[error("service became unavailable: {0:?}")]
+    ServiceUnavailable(Service),
+    /// An internal invariant was violated; carries a human-readable description.
+    #[error("internal error: {0}")]
+    Internal(String),
+}

processor/src/executor.rs

@@ -0,0 +1,177 @@
+//! Transaction execution coordination.
+
+use std::{
+    collections::VecDeque,
+    sync::{
+        Arc,
+        mpsc::{self, Receiver, SyncSender},
+    },
+    thread::{self, JoinHandle},
+};
+
+use ahash::HashMap;
+use derive_more::{Deref, DerefMut};
+use keeper::{ExecutionRecord, FullTransaction, Keeper};
+use nucleus::{
+    shutdown::{Service, ShutdownHandle, ShutdownManager, ShutdownReason},
+    tls::AUTHORITY,
+};
+use solana_program_runtime::loaded_programs::ProgramCache;
+use solana_pubkey::Pubkey;
+use solana_svm::transaction_processing_result::TransactionProcessingResultExtensions;
+use tokio::sync::mpsc::Sender;
+use tracing::{error, info};
+
+use crate::{
+    ExecutorMessage, ExecutorReady, ResolvedTransaction, Result,
+    callback::SVMCallback,
+    metrics::{self, FailureKind},
+    svm::SvmContext,
+};
+
+/// Index identifying an executor within the pool.
+pub(crate) type ExecutorId = u32;
+
+/// Sequencer-owned work accumulated for an executor between dispatches.
+#[derive(Default)]
+pub(crate) struct ExecutorWork {
+    /// Transactions accumulated for the next dispatch.
+    pub(crate) batch: Vec<ResolvedTransaction>,
+    /// Accounts held on this executor's behalf, mapped to their reference count.
+    pub(crate) locks: HashMap<Pubkey, usize>,
+    /// Transactions queued behind this executor on lock contention.
+    pub(crate) blocked: VecDeque<ResolvedTransaction>,
+}
+
+/// Worker that drives the SVM for batches of conflict-free transactions.
+pub(crate) struct TransactionExecutor {
+    /// This executor's index within the pool.
+    id: ExecutorId,
+    /// Inbound channel of execution batches and block boundaries.
+    rx: Receiver<ExecutorMessage>,
+    /// Channel used to report back to the sequencer once a batch completes.
+    ready: Sender<ExecutorReady>,
+    /// SVM batch processor and per-block environment driving execution.
+    svm: SvmContext,
+    /// Durable engine state used for account loads and commits.
+    state: Arc<Keeper>,
+    /// Handle used to report cooperative shutdown for this worker.
+    shutdown: ShutdownHandle,
+    /// Whether the executor runs in ledger-replay mode: when set, only state
+    /// transitions are committed directly instead of recording full execution.
+    replay: bool,
+}
+
+/// Sequencer-side handle to a spawned executor: its dispatch channel, pending
+/// batch, held locks, and join handle.
+#[derive(Deref, DerefMut)]
+pub(crate) struct ExecutorHandle {
+    /// Index of the executor this handle controls.
+    pub(crate) id: ExecutorId,
+    /// Mutable sequencing state moved out while completed work is reclaimed.
+    #[deref]
+    #[deref_mut]
+    pub(crate) work: ExecutorWork,
+    /// Channel for dispatching batches and block boundaries to the worker.
+    pub(crate) tx: SyncSender<ExecutorMessage>,
+    /// Join handle for the worker thread, taken during shutdown.
+    pub(crate) task: Option<JoinHandle<()>>,
+}
+
+impl TransactionExecutor {
+    /// Builds the SVM environment, spawns the worker thread, and returns its
+    /// sequencer-side handle.
+    pub(crate) fn spawn(
+        id: ExecutorId,
+        state: Arc<Keeper>,
+        cache: Arc<ProgramCache>,
+        shutdown: &mut ShutdownManager,
+        ready: Sender<ExecutorReady>,
+        replay: bool,
+    ) -> Result<ExecutorHandle> {
+        let svm = SvmContext::new(&state, cache)?;
+        let shutdown = shutdown.handle(Service::TransactionExecutor(id));
+        let (tx, rx) = mpsc::sync_channel(3);
+        let executor = Self {
+            id,
+            rx,
+            svm,
+            ready,
+            state,
+            shutdown,
+            replay,
+        };
+        let task = thread::Builder::new()
+            .name(format!("transaction-executor-{id}"))
+            .spawn(move || executor.run())?;
+        Ok(ExecutorHandle {
+            id,
+            task: Some(task),
+            tx,
+            work: Default::default(),
+        })
+    }
+
+    /// Worker loop: executes batches and applies block transitions until the
+    /// channel closes or a batch fails, then reports the termination reason.
+    fn run(mut self) {
+        // MagicRoot authorizes callers against this thread-local; publish the
+        // engine authority before executing any transaction on this thread.
+        AUTHORITY.set(self.state.authority());
+        let mut error = None;
+        while let Ok(msg) = self.rx.recv() {
+            match msg {
+                ExecutorMessage::Transactions(mut batch) => {
+                    let result = self.process(&mut batch);
+                    if let Err(e) = result {
+                        error.replace(e);
+                        break;
+                    }
+                    let signal = ExecutorReady { id: self.id, batch };
+                    if self.ready.blocking_send(signal).is_err() {
+                        info!(id = self.id, "ready channel closed, executor exiting");
+                        break;
+                    }
+                }
+                ExecutorMessage::Block(block) => self.svm.transition(block),
+            };
+        }
+        let reason = if let Some(error) = error {
+            error!(?error, self.id, "executor failed, terminating");
+            ShutdownReason::Error(Box::new(error))
+        } else {
+            ShutdownReason::Signalled
+        };
+        self.shutdown.terminate(reason);
+    }
+
+    /// Loads and executes each transaction in the batch through the SVM,
+    /// committing either raw state transitions (replay) or full execution.
+    fn process(&mut self, transactions: &mut Vec<ResolvedTransaction>) -> Result<()> {
+        let accessor = self.state.accounts();
+        let callback = SVMCallback::<false> {
+            loader: accessor.loader(),
+            featureset: self.state.features(),
+        };
+        for txn in transactions.drain(..) {
+            let output = self.svm.execute(&callback, &txn, self.state.features());
+            if !output.processing_result.was_processed_with_successful_result() {
+                metrics::failed_transaction(FailureKind::Execution);
+            }
+            if self.replay {
+                self.state.transactions().commit_state_transitions(&output.processing_result)?;
+            } else {
+                let txn = FullTransaction {
+                    transaction: txn.into_view(),
+                    execution: ExecutionRecord {
+                        result: output.processing_result,
+                        balances: output.balance_collector,
+                        slot: self.svm.slot(),
+                    },
+                };
+                self.state.transactions().commit_execution(txn)?;
+            }
+        }
+        Ok(())
+    }
+}

processor/src/lib.rs

@@ -0,0 +1,36 @@
+#![doc = include_str!("../README.md")]
+
+use keeper::ResolvedTransaction;
+use nucleus::ledger::Block;
+
+use crate::executor::ExecutorId;
+
+mod callback;
+mod error;
+mod executor;
+mod metrics;
+pub mod sequencer;
+pub mod simulator;
+mod svm;
+
+#[cfg(test)]
+mod tests;
+
+pub use error::{ProcessorError, Result};
+pub use nucleus::runtime::{SequencerMessage, Simulation, SimulatorMessage};
+
+/// Batch of work delivered from the sequencer to a transaction executor.
+pub enum ExecutorMessage {
+    /// A set of conflict-free transactions to execute together.
+    Transactions(Vec<ResolvedTransaction>),
+    /// A block boundary advancing the executor's processing environment.
+    Block(Block),
+}
+
+/// Notification that an executor finished its batch and is idle again.
+struct ExecutorReady {
+    /// Identifier of the executor that became ready.
+    id: ExecutorId,
+    /// The drained batch handed back for reuse
+    batch: Vec<ResolvedTransaction>,
+}