feat: adds top level engine crate for global orchestration

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "accountsdb",
+    "engine",
     "keeper",
     "ledger",
     "nucleus",
@@ -27,6 +28,7 @@ version = "0.1.0"
 
 [workspace.dependencies]
 accountsdb = { path = "accountsdb", package = "magicblock-accountsdb" }
+engine = { path = "engine", package = "magicblock-engine" }
 keeper = { path = "keeper", package = "magicblock-keeper" }
 ledger = { path = "ledger", package = "magicblock-ledger" }
 magic-root-interface = { path = "programs/magic-root-interface" }

clippy.toml

@@ -19,7 +19,6 @@ max-struct-bools = 2
 check-private-items = true
 
 disallowed-methods = [
-    { path = "std::option::Option::unwrap", reason = "Use expect with context or handle the None case explicitly" },
-    { path = "std::result::Result::unwrap", reason = "Use expect with context or propagate the error explicitly" },
-    { path = "std::thread::sleep", reason = "Avoid timing-based flakiness in workspace code; isolate retries/backoff behind abstractions" }
+    { path = "std::thread::sleep", reason = "Avoid timing-based flakiness in workspace code; use event driven logic" },
+    { path = "tokio::time::sleep", reason = "Avoid timing-based flakiness in workspace code; use event driven logic" }
 ]

engine/Cargo.toml

@@ -0,0 +1,50 @@
+[package]
+name = "magicblock-engine"
+
+authors.workspace = true
+edition.workspace = true
+homepage.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+version.workspace = true
+
+[lib]
+name = "engine"
+
+[dependencies]
+keeper = { workspace = true }
+ledger = { workspace = true }
+magic-root-interface = { workspace = true }
+magic-root-program = { workspace = true }
+nucleus = { workspace = true }
+processor = { workspace = true }
+
+derive_more = { workspace = true }
+num_cpus = { workspace = true }
+oneshot = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true, features = ["sync"] }
+tracing = { workspace = true }
+wincode = { workspace = true }
+
+agave-transaction-view = { workspace = true }
+solana-account = { workspace = true }
+solana-instruction = { workspace = true }
+solana-keypair = { workspace = true }
+solana-program-runtime = { workspace = true }
+solana-pubkey = { workspace = true }
+solana-transaction = { workspace = true, features = ["wincode"] }
+
+[dev-dependencies]
+keeper = { workspace = true, features = ["testkit"] }
+nucleus = { workspace = true, features = ["testkit"] }
+v42-calculator-interface = { workspace = true, features = ["builder"] }
+
+solana-instruction-error = { workspace = true }
+solana-signer = { workspace = true }
+solana-sysvar = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt", "rt-multi-thread", "time", "sync"] }
+
+[lints]
+workspace = true

engine/README.md

@@ -0,0 +1,47 @@
+# `magicblock-engine`
+
+This is the top of the stack — the crate a consumer actually instantiates. It
+wires the two layers below it together: the durable state (`keeper`, which is
+accountsdb + ledger) and the transaction sequencer (`processor`). An `Engine`
+owns the signing authority, the shared `Keeper`, and a sequencer handle, and
+nothing else in the workspace depends on it. Everything above this point is the
+caller's program.
+
+## Startup and recovery
+
+A healthy restart comes up on **persisted state** — the authoritative copy is
+already on disk, so the engine just opens it. It does *not* replay the ledger or
+rebuild volatile state on a normal startup.
+
+At startup the engine always registers MagicRoot as a native built-in before
+opening keeper state. That keeps the account mutation path available even when a
+caller supplies no built-ins of its own.
+
+Replay is the **corruption-recovery** path. When accountsdb opens corrupt, the
+keeper restores an older archived snapshot, which leaves accountsdb trailing the
+ledger tip. Only then does the engine replay the retained ledger entries from the
+restored slot up to the tip, re-executing them through a temporary replay
+sequencer to rebuild volatile state and catch back up. (When accountsdb is already
+current, `replay` is a no-op.)
+
+Replay is also where silent corruption gets caught. At each sealed superblock the
+replay is quiesced and the reconstructed account checksum is compared against the
+checksum that was sealed into that superblock. A mismatch is
+`ReplayError::StateMismatch` — it means the durable state diverged from what was
+recorded, and surfacing it loudly on restart is far better than running on top of
+quietly wrong state.
+
+Volatile state can also be dropped at runtime. Because it's a mirror of
+chain-owned accounts the engine can always re-fetch, a caller that can no
+longer trust the mirror is in sync with chain (for example, the chain-sync
+connection dropped) may reset it; persisted ER-exclusive state is left intact.
+
+## What it exposes
+
+- `account()` — CRUD over accounts, expressed as MagicRoot instruction sequences.
+  Account mutation is privileged, so it can't be an ordinary write; it goes
+  through the MagicRoot built-in.
+- `transaction()` — execute, schedule, or simulate a transaction.
+- `IntoTransactionView` — signs and sanitizes a `Message`, an `&[Instruction]`,
+  or an existing view, using the engine authority and the latest blockhash, so
+  callers can hand in whichever form is convenient.

engine/src/accessor.rs

@@ -0,0 +1,118 @@
+//! Account- and transaction-scoped operation facades.
+
+use std::time::Duration;
+
+use keeper::{ExecutionRecord, TransactionView};
+use magic_root_interface::MagicRootInstruction;
+use processor::{SequencerMessage, Simulation, SimulatorMessage};
+use solana_account::{AccountFieldPatch, OwnedAccount};
+use solana_instruction::Instruction;
+use solana_pubkey::Pubkey;
+use solana_transaction::TransactionResult;
+use tokio::time;
+
+use crate::{Engine, IntoTransactionView, error::Result};
+
+/// Upper bound on awaiting a submitted transaction's committed result.
+const EXECUTION_TIMEOUT: Duration = Duration::from_secs(8);
+
+/// Account-scoped operations bound to a single `pubkey`.
+pub struct AccountAccessor<'a> {
+    pub(crate) pubkey: Pubkey,
+    pub(crate) engine: &'a Engine,
+}
+
+/// Transaction-submission operations bound to an engine instance.
+pub struct TransactionAccessor<'a> {
+    pub(crate) engine: &'a Engine,
+    pub(crate) transaction: TransactionView,
+}
+
+impl AccountAccessor<'_> {
+    /// Creates the account by patching in every field and finalizing it,
+    /// optionally running follow-up `actions` once it is finalized.
+    pub async fn create(
+        &self,
+        account: OwnedAccount,
+        actions: Option<Vec<Instruction>>,
+    ) -> Result<()> {
+        let mut instructions = self.patch_instructions(account)?;
+        if let Some(actions) = actions {
+            instructions.push(MagicRootInstruction::PostFinalize(actions).compose(self.pubkey)?);
+        }
+        self.execute(instructions).await
+    }
+
+    /// Updates the account by patching in every field of `account`
+    pub async fn update(&self, account: OwnedAccount) -> Result<()> {
+        let instructions = self.patch_instructions(account)?;
+        self.execute(instructions).await
+    }
+
+    /// Applies the given field patches to the account.
+    pub async fn patch(&self, patches: Vec<AccountFieldPatch>) -> Result<()> {
+        let mut instructions = Vec::with_capacity(patches.len());
+        for patch in patches {
+            let ix = MagicRootInstruction::Patch(patch).compose(self.pubkey)?;
+            instructions.push(ix);
+        }
+        self.execute(instructions).await
+    }
+
+    /// Closes the account.
+    pub async fn delete(&self) -> Result<()> {
+        let instructions = vec![MagicRootInstruction::Delete.compose(self.pubkey)?];
+        self.execute(instructions).await
+    }
+
+    /// Builds the instruction sequence that patches every
+    /// field of `account` and finalizes it.
+    fn patch_instructions(&self, account: OwnedAccount) -> Result<Vec<Instruction>> {
+        let patches = AccountFieldPatch::sequence(account);
+        let mut instructions = Vec::with_capacity(patches.len() + 1);
+        for patch in patches {
+            instructions.push(MagicRootInstruction::Patch(patch).compose(self.pubkey)?);
+        }
+        instructions.push(MagicRootInstruction::Finalize.compose(self.pubkey)?);
+        Ok(instructions)
+    }
+
+    /// Composes the instructions into a signed engine transaction, executes it,
+    /// and flattens the committed transaction result into the engine error type.
+    async fn execute(&self, instructions: Vec<Instruction>) -> Result<()> {
+        let txn = instructions.compose(self.engine)?;
+        self.engine.transaction(txn)?.execute().await?.map_err(Into::into)
+    }
+}
+
+impl TransactionAccessor<'_> {
+    /// Submits `transaction` for execution and awaits its committed result.
+    pub async fn execute(self) -> Result<TransactionResult<()>> {
+        let signature = self.transaction.signatures()[0];
+        let msg = SequencerMessage::Transaction(self.transaction);
+        let mut rx = self.engine.transactions().subscribe_signature(signature).await;
+        self.engine.sequencer.send(msg).await?;
+        let status = time::timeout(EXECUTION_TIMEOUT, rx.recv())
+            .await
+            .map_err(|e| e.to_string())?
+            .map_err(|e| e.to_string())?;
+        Ok(status.result)
+    }
+
+    /// Submits `transaction` for execution without awaiting its result.
+    pub async fn schedule(self) -> Result<()> {
+        let msg = SequencerMessage::Transaction(self.transaction);
+        self.engine.sequencer.send(msg).await.map_err(Into::into)
+    }
+
+    /// Simulates `transaction` against current state without committing it.
+    pub async fn simulate(self) -> Result<TransactionResult<ExecutionRecord>> {
+        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)
+    }
+}

engine/src/error.rs

@@ -0,0 +1,73 @@
+//! Engine error types.
+
+use agave_transaction_view::result::TransactionViewError;
+use derive_more::From;
+use keeper::error::KeeperError;
+use ledger::{LedgerError, LedgerRequestError};
+use nucleus::shutdown::Service;
+use processor::ProcessorError;
+use solana_transaction::{SignerError, TransactionError};
+use tokio::sync::mpsc::error::SendError;
+
+/// Result type used by engine APIs.
+pub type Result<T> = std::result::Result<T, EngineError>;
+
+/// Failures surfaced by the top-level engine.
+#[derive(From, thiserror::Error, Debug)]
+pub enum EngineError {
+    /// A durable-state (keeper) operation failed.
+    #[error("state error: {0}")]
+    State(#[source] KeeperError),
+    /// Scheduling or executing a transaction failed.
+    #[error("processor error: {0}")]
+    Processor(#[source] ProcessorError),
+    /// Replaying the ledger into volatile state on startup failed.
+    #[error("replay error: {0}")]
+    Replay(#[source] ReplayError),
+    /// A background service is no longer reachable.
+    #[error("service became unavailable: {0:?}")]
+    ServiceUnavailable(Service),
+    /// Signing a transaction with the engine authority failed.
+    #[error("signature error: {0}")]
+    Signature(#[source] SignerError),
+    /// Serializing or deserializing a transaction failed.
+    #[error("serialization error: {0}")]
+    Serde(#[source] wincode::Error),
+    /// Sanitizing a serialized transaction into a transaction view failed.
+    #[error("transaction sanitization: {0:?}")]
+    Sanitization(TransactionViewError),
+    /// A submitted transaction was committed with an execution failure.
+    #[error("transaction execution failed: {0}")]
+    TransactionExecution(#[source] TransactionError),
+    /// An unexpected internal failure carrying a contextual message.
+    #[error("internal error: {0}")]
+    Internal(String),
+}
+
+impl<T> From<SendError<T>> for EngineError {
+    fn from(_: SendError<T>) -> Self {
+        Self::ServiceUnavailable(Service::Sequencer)
+    }
+}
+impl From<oneshot::RecvError> for EngineError {
+    fn from(_: oneshot::RecvError) -> Self {
+        Self::ServiceUnavailable(Service::Sequencer)
+    }
+}
+
+/// Failures raised while replaying retained ledger entries on startup.
+#[derive(From, thiserror::Error, Debug)]
+pub enum ReplayError {
+    /// A retained transaction could not be sanitized into a transaction view.
+    #[error("transaction sanitization: {0:?}")]
+    Sanitization(TransactionViewError),
+    /// The replayed account state checksum diverged from the sealed superblock.
+    #[error("replayed state checksum mismatch")]
+    StateMismatch,
+    /// Waiting for the ledger reader's replay response failed.
+    #[error("ledger replay request failed: {0}")]
+    Request(LedgerRequestError),
+    /// Reading or decoding retained ledger entries failed.
+    #[error("ledger replay failed: {0}")]
+    Ledger(LedgerError),
+}