feat: adds top level engine crate for global orchestration

Cargo.toml

@@ -1,6 +1,7 @@
 [workspace]
 members = [
     "accountsdb",
+    "engine",
     "keeper",
     "ledger",
     "nucleus",
@@ -28,6 +29,7 @@ version = "0.1.0"
 
 [workspace.dependencies]
 accountsdb = { path = "accountsdb", package = "magicblock-accountsdb" }
+engine = { path = "engine", package = "magicblock-engine" }
 keeper = { path = "keeper", package = "magicblock-keeper" }
 ledger = { path = "ledger", package = "magicblock-ledger" }
 magic-root-interface = { path = "programs/magic-root-interface" }
@@ -90,6 +92,7 @@ agave-transaction-view = "4.1.1"
 solana-account-info = "3.1.1"
 solana-clock = "3.1.0"
 solana-compute-budget-instruction = "=4.1.1"
+solana-compute-budget-program = "=4.1.1"
 solana-cpi = "3.1.0"
 solana-ed25519-program = "3.0.0"
 solana-epoch-rewards = "3.0.1"

README.md

@@ -201,38 +201,31 @@ 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();
 let owner = Pubkey::new_unique();
 let account = AccountBuilder::default()
     .lamports(2_000_000)
     .owner(owner)
-    .mode(AccountMode::Delegated)
+    .mode(AccountMode::ReadOnly)
+    .slot(1)
     .data(vec![1, 2, 3, 4])
     .build();
 
 engine.account(key).create(account, None).await?;
-let current = engine.accounts().loader().load(&key)?;
-
-engine
-    .account(key)
-    .patch(vec![AccountFieldPatch::DataAt {
-        offset: 0,
-        data: vec![9; 4],
-    }])
-    .await?;
 
 let replacement = AccountBuilder::default()
     .lamports(2_000_000)
     .owner(owner)
-    .mode(AccountMode::Delegated)
+    .mode(AccountMode::ReadOnly)
+    .slot(2)
     .data(vec![5; 4])
     .build();
 engine.account(key).update(replacement).await?;
@@ -241,7 +234,12 @@ engine.account(key).delete().await?;
 
 Each mutation is one committed transaction. `create` can also run optional
 post-finalize instructions in that transaction; if an instruction fails, the
-creation does not commit.
+creation does not commit. Complete-account patches cover non-flag fields, and
+finalization atomically installs the caller-supplied flags without changing
+lamports. Callers are responsible for supplying current state; later
+replacements remain subject to the account's slot and lifecycle rules. Internal
+create composition places post-finalize instructions immediately after
+finalization.
 
 Missing external accounts can be coordinated with `Engine::accounts().ensure`.
 The first caller receives `MissingAccount::Load`; concurrent callers receive a
@@ -289,20 +287,23 @@ async fn submit(
 No polling loops — subscribe to what you care about and the engine pushes
 updates as they happen.
 
-Keeper accessors expose Tokio broadcast receivers for live state:
+Keeper accessors expose dedicated Tokio channels for live state:
 
 ```rust
 let mut account_updates = engine.accounts().subscribe(key).await;
 let mut blocks = engine.blocks().subscribe();
 
-let account = account_updates.recv().await?;
-let block = blocks.recv().await?;
+let account = account_updates.recv().await.expect("account stream is open");
+let block = blocks.recv().await.expect("block stream is open");
 ```
 
 Related accessors subscribe to program-owned accounts, cache evictions,
 snapshot completion, transaction status, logs, processed transactions, and
-service messages. Broadcast consumers must handle `Lagged` when they fall
-behind and `Closed` during shutdown; retained reads are available separately.
+service messages. Signatures use terminal oneshot channels; other multicast
+streams give each consumer a bounded queue and disconnect a consumer that falls
+behind. Processed transactions, service messages, and cache evictions each have
+one process-lifetime consumer and apply producer backpressure when its queue is
+full.
 
 ---
 

engine/Cargo.toml

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

engine/README.md

@@ -0,0 +1,62 @@
+# `magicblock-engine`
+
+This crate exposes `Engine`, the consumer-facing handle over keeper state,
+transaction sequencing, simulation, block pacing, recovery, and MagicRoot
+account operations. It registers MagicRoot and the System Program as native
+builtins before keeper opens startup state.
+
+`Engine::signer` is always the local keypair. `Engine::authority` returns the
+configured remote authority for a replica, or the local identity when no
+override is configured. Replication uses that distinction to sign locally while
+authenticating its immediate upstream.
+
+## Account replacement
+
+`AccountAccessor::{create, update}` composes complete-account MagicRoot patch
+transactions. Replacement slots are monotonic: a newer slot is accepted, an
+equal slot requires a genuine account-mode transition, and an older slot is
+rejected even when the mode changes. Failed replacements are transactionally
+rolled back. Complete-account patch sequences cover non-flag fields, while
+finalization atomically installs the caller-supplied complete flag value without
+changing lamports. Callers are responsible for supplying current state; later
+replacements remain subject to the account's slot and lifecycle rules. `create`
+appends any `PostFinalize` actions immediately after finalization in the same
+transaction. Magicblock construction rejects instruction, address, account-meta,
+and instruction-data lengths that cannot be represented by the V1 wire fields.
+
+## Startup and recovery
+
+Keeper restores an accountsdb snapshot when the active store is corrupt, its
+sealed superblock trails the retained ledger, or its committed transaction count
+trails the ledger's durable count. Accountsdb's count is a checkpoint high-water
+mark, so a count ahead of the locally retained ledger is current, including for
+snapshots staged by a replication follower. Superblock lag remains recoverable
+independently of the counters.
+
+If accountsdb then trails the ledger tip, `Engine::new` replays retained entries
+from the successor of its sealed snapshot through a temporary sequencer. Replay
+quiesces at superblock seals and compares the reconstructed checksum with the
+recorded seal. A mismatch returns `ReplayError::StateMismatch`. Current state
+opens without replay when its slot and transaction count are each at least the
+ledger values. After replay actually runs, the final transaction counts must be
+equal or startup returns `ReplayError::StateMismatch`.
+
+Internal pacing appends one reset marker at the current slot and clears
+chain-mirrored volatile accounts before the pacemaker task starts. Internal
+system accounts remain available. Replicas use external pacing and retain
+restored volatile state. External block producers supply the slot and timestamp;
+the sequencer overwrites hash-chain metadata with its locally computed hash and
+parent.
+
+## Shutdown
+
+Shutdown behavior follows the pacing source. Internal pacing publishes a final
+block and flushes durable state. External pacing flushes the durable cursor
+before writing `CURRENT/volatile.db`, allowing the next open and replication
+handshake to resume from matching state. The pacemaker holds the sequencer
+barrier while issuing a terminal ledger sync, which closes the appender and
+reader workers without waiting for every engine handle to be dropped.
+
+The embedding service retains the `ShutdownManager` passed to `Engine::new` and
+calls `terminate` after stopping external ingress. The manager stops the
+replication client, pacemaker, sequencer, and backing services in order.

engine/src/accessor.rs

@@ -0,0 +1,106 @@
+//! Account- and transaction-scoped operation facades.
+
+use std::{sync::atomic::Ordering, time::Duration};
+
+use keeper::{ExecutionRecord, TransactionView};
+use magic_root_interface::MagicRootInstruction;
+use processor::{SequencerMessage, Simulation, SimulatorMessage};
+use solana_account::OwnedAccount;
+use solana_instruction::Instruction;
+use solana_pubkey::Pubkey;
+use solana_transaction::TransactionResult;
+use tokio::time;
+
+use crate::{Engine, error::EngineError, error::Result, transaction};
+
+/// Upper bound on awaiting a submitted transaction's committed result.
+const EXECUTION_TIMEOUT: Duration = Duration::from_secs(8);
+
+/// Account-scoped operations bound to a single `pubkey`.
+pub struct AccountAccessor<'a> {
+    pub(crate) pubkey: Pubkey,
+    pub(crate) engine: &'a Engine,
+}
+
+/// Transaction-submission operations bound to an engine instance.
+pub struct TransactionAccessor<'a> {
+    pub(crate) engine: &'a Engine,
+    pub(crate) transaction: TransactionView,
+}
+
+impl AccountAccessor<'_> {
+    /// Creates the account by patching in every field and finalizing it,
+    /// optionally running follow-up `actions` once it is finalized.
+    pub async fn create(
+        &self,
+        acc: impl Into<OwnedAccount>,
+        actions: Option<Vec<Instruction>>,
+    ) -> 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<OwnedAccount>) -> Result<()> {
+        let instructions = MagicRootInstruction::compose_account(self.pubkey, acc.into())?;
+        self.execute(instructions).await
+    }
+
+    /// Closes the account.
+    pub async fn delete(&self) -> Result<()> {
+        let instructions = vec![MagicRootInstruction::Delete.compose(self.pubkey)?];
+        self.execute(instructions).await
+    }
+
+    /// Composes the instructions into a signed engine transaction, executes it,
+    /// and flattens the committed transaction result into the engine error type.
+    async fn execute(&self, instructions: Vec<Instruction>) -> Result<()> {
+        let txn = transaction::magicblock(&instructions, self.engine)?;
+        self.engine.transaction(txn)?.execute().await?.map_err(Into::into)
+    }
+}
+
+impl TransactionAccessor<'_> {
+    /// Submits `transaction` for execution and awaits its committed result.
+    /// A timeout does not cancel the submitted transaction.
+    pub async fn execute(self) -> Result<TransactionResult<()>> {
+        if self.engine.terminating.load(Ordering::Acquire) {
+            return Err(EngineError::ShuttingDown);
+        }
+        let signature = self.transaction.signatures()[0];
+        let msg = SequencerMessage::Transaction(self.transaction);
+        let rx = self.engine.transactions().subscribe_signature(signature).await;
+        self.engine.sequencer.send(msg).await?;
+        let status = time::timeout(EXECUTION_TIMEOUT, rx)
+            .await
+            .map_err(|_| EngineError::TransactionTimeout)?
+            .map_err(|e| e.to_string())?;
+        Ok(status.result)
+    }
+
+    /// Submits `transaction` for execution without awaiting its result.
+    pub async fn schedule(self) -> Result<()> {
+        if self.engine.terminating.load(Ordering::Acquire) {
+            return Err(EngineError::ShuttingDown);
+        }
+        let msg = SequencerMessage::Transaction(self.transaction);
+        self.engine.sequencer.send(msg).await.map_err(Into::into)
+    }
+
+    /// Simulates `transaction` against current state without committing it.
+    pub async fn simulate(self) -> Result<TransactionResult<ExecutionRecord>> {
+        if self.engine.terminating.load(Ordering::Acquire) {
+            return Err(EngineError::ShuttingDown);
+        }
+        let (response, rx) = oneshot::channel();
+        let msg = SimulatorMessage::Transaction(Simulation {
+            transaction: self.transaction,
+            response,
+        });
+        self.engine.sequencer.simulation.send(msg).await?;
+        rx.await.map_err(Into::into)
+    }
+}