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",
@@ -31,6 +32,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,56 @@
+[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 = ["ledger-schema", "metrics", "runtime", "shutdown", "tls"] }
+
+ahash = { workspace = true }
+blake3 = { workspace = true }
+derive_more = { workspace = true, features = ["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"] }
+
+solana-instruction = { workspace = true }
+solana-keypair = { workspace = true }
+solana-sdk-ids = { workspace = true }
+solana-signer = { workspace = true }
+solana-transaction = { workspace = true, features = ["wincode"] }
+oneshot = { workspace = true }
+wincode = { workspace = true }
+
+[lints]
+workspace = true

processor/README.md

@@ -0,0 +1,29 @@
+# `magicblock-processor`
+
+The processor is where transactions actually get run. Its job is to execute as
+many of them as possible in parallel while still producing the result they'd have
+if run one at a time — and the thing standing in the way of free parallelism is
+*account conflicts*. Two transactions that touch the same account (with at least
+one writing it) can't safely run at once; two that touch disjoint accounts can.
+
+So the sequencer's core task is conflict detection. It tracks each `Pubkey` with a
+write bit plus a per-executor occupancy bitset (sized to `MAX_EXECUTORS`).
+Transactions whose account sets don't collide are fanned out across the pool of
+SVM executors and run concurrently; the rest are serialized behind the work
+they conflict with. Reads through to accounts go via `keeper`/`accountsdb`, and
+writes go back the same way.
+
+## Quiescence
+
+Some operations need a consistent, frozen view of state — taking a snapshot at a
+superblock seal, or replaying the ledger on restart. In-flight transactions would
+make that view inconsistent, so the processor has a **quiescence barrier** that
+drains outstanding work and holds new work off until the barrier is released.
+That's the hook the keeper's finalize and the engine's replay rely on.
+
+## Simulation
+
+The simulation path runs a transaction against current state on *owned copies* of
+the accounts and returns the execution record without committing anything. Because
+it never touches the real accounts, it can run without disturbing the live
+scheduling path.

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,167 @@
+//! Transaction execution coordination.
+
+use std::{
+    collections::VecDeque,
+    sync::{
+        Arc,
+        mpsc::{self, Receiver, SyncSender},
+    },
+    thread::{self, JoinHandle},
+};
+
+use ahash::HashMap;
+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 tokio::sync::mpsc::Sender;
+use tracing::{error, info};
+
+use crate::{
+    ExecutorMessage, ExecutorReady, ResolvedTransaction, Result, callback::SVMCallback,
+    svm::SvmContext,
+};
+
+/// Index identifying an executor within the pool.
+pub(crate) type ExecutorId = u32;
+
+/// 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.
+pub(crate) struct ExecutorHandle {
+    /// Index of the executor this handle controls.
+    pub(crate) id: ExecutorId,
+    /// Transactions accumulated for the next dispatch.
+    pub(crate) batch: Vec<ResolvedTransaction>,
+    /// Accounts held on this executor's behalf, each mapped to the number of
+    /// in-flight transactions referencing it (released when the count hits zero).
+    pub(crate) locks: HashMap<Pubkey, usize>,
+    /// Transactions queued behind this executor on lock contention.
+    pub(crate) blocked: VecDeque<ResolvedTransaction>,
+    /// 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,
+            batch: Vec::new(),
+            locks: Default::default(),
+            blocked: VecDeque::new(),
+        })
+    }
+
+    /// 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,
+                id = self.id,
+                "executor encountered catastrophic failure, 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 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>,
+}