Feat: multi threaded scheduler

Cargo.toml

@@ -91,7 +91,7 @@ jsonrpc-pubsub = "18.0.0"
 jsonrpc-ws-server = "18.0.0"
 lazy_static = "1.4.0"
 libc = "0.2.153"
-log = { version = "0.4.20", features = ["release_max_level_info"] }
+log = { version = "0.4.20" }
 lru = "0.16.0"
 macrotest = "1"
 magic-domain-program = { git = "https://github.com/magicblock-labs/magic-domain-program.git", rev = "ea04d46", default-features = false }
@@ -140,11 +140,12 @@ prost = "0.11.9"
 json = { package = "sonic-rs", version = "0.5.3" }
 protobuf-src = "1.1"
 quote = "1.0"
-rand = "0.8.5"
+rand = "0.9"
 rayon = "1.10.0"
 # bundled sqlite 3.44
 rusqlite = { version = "0.37.0", features = ["bundled"] }
 rustc_version = "0.4"
+rustc-hash = "2.1"
 scc = "2.4"
 semver = "1.0.22"
 serde = "1.0.217"

magicblock-aperture/Cargo.toml

@@ -66,7 +66,7 @@ log = { workspace = true }
 serde = { workspace = true }
 
 [dev-dependencies]
-rand = "0.9"
+rand = { workspace = true }
 test-kit = { workspace = true }
 solana-rpc-client = { workspace = true }
 solana-pubsub-client = { workspace = true }

magicblock-aperture/src/requests/http/get_blocks_with_limit.rs

@@ -21,9 +21,8 @@ impl HttpDispatcher {
             .min(MAX_DEFAULT_BLOCKS_LIMIT);
         let end_slot = start_slot + limit;
         // Calculate the end slot, ensuring it does not exceed the latest block height.
-        let end_slot = (end_slot).min(self.blocks.block_height());
+        let end_slot = (end_slot).min(self.blocks.block_height() + 1);
 
-        // The range is exclusive of the end slot, so `(start..end)` is correct.
         let slots = (start_slot..end_slot).collect::<Vec<Slot>>();
 
         Ok(ResponsePayload::encode_no_context(&request.id, slots))

magicblock-aperture/src/requests/http/get_token_account_balance.rs

@@ -1,7 +1,10 @@
 use std::mem::size_of;
 
 use solana_account::AccountSharedData;
-use solana_account_decoder::parse_token::UiTokenAmount;
+use solana_account_decoder::{
+    parse_account_data::SplTokenAdditionalDataV2,
+    parse_token::token_amount_to_ui_amount_v3,
+};
 
 use super::{
     prelude::*, MINT_DECIMALS_OFFSET, SPL_MINT_RANGE, SPL_TOKEN_AMOUNT_RANGE,
@@ -58,13 +61,14 @@ impl HttpDispatcher {
             u64::from_le_bytes(buffer)
         };
 
-        let ui_amount = (token_amount as f64) / 10f64.powi(decimals as i32);
-        let ui_token_amount = UiTokenAmount {
-            amount: token_amount.to_string(),
-            ui_amount: Some(ui_amount),
-            ui_amount_string: ui_amount.to_string(),
-            decimals,
-        };
+        let ui_token_amount = token_amount_to_ui_amount_v3(
+            token_amount,
+            &SplTokenAdditionalDataV2 {
+                decimals,
+                interest_bearing_config: None,
+                scaled_ui_amount_config: None,
+            },
+        );
 
         let slot = self.blocks.block_height();
         Ok(ResponsePayload::encode(&request.id, ui_token_amount, slot))

magicblock-aperture/src/requests/http/mocked.rs

@@ -84,7 +84,7 @@ impl HttpDispatcher {
         Ok(ResponsePayload::encode(
             &request.id,
             Vec::<()>::new(),
-            self.blocks.get_latest().slot,
+            self.blocks.block_height(),
         ))
     }
 
@@ -103,7 +103,7 @@ impl HttpDispatcher {
         Ok(ResponsePayload::encode(
             &request.id,
             supply,
-            self.blocks.get_latest().slot,
+            self.blocks.block_height(),
         ))
     }
 
@@ -120,7 +120,7 @@ impl HttpDispatcher {
         Ok(ResponsePayload::encode(
             &request.id,
             supply,
-            self.blocks.get_latest().slot,
+            self.blocks.block_height(),
         ))
     }
 

magicblock-aperture/src/requests/http/request_airdrop.rs

@@ -15,13 +15,18 @@ impl HttpDispatcher {
         // Airdrops are only supported if a faucet keypair is configured.
         // Which is never the case with *ephemeral* running mode of the validator
         let Some(ref faucet) = self.context.faucet else {
-            return Err(RpcError::invalid_request("method is not supported"));
+            return Err(RpcError::invalid_request(
+                "free airdrop faucet is disabled",
+            ));
         };
 
         let (pubkey, lamports) =
             parse_params!(request.params()?, Serde32Bytes, u64);
         let pubkey = some_or_err!(pubkey);
         let lamports = some_or_err!(lamports);
+        if lamports == 0 {
+            return Err(RpcError::invalid_params("lamports must be > 0"));
+        }
 
         // Build and execute the airdrop transfer transaction.
         let txn = solana_system_transaction::transfer(

magicblock-api/src/magic_validator.rs

@@ -302,8 +302,14 @@ impl MagicValidator {
             base_fee: config.validator.base_fees.unwrap_or_default(),
             featureset: txn_scheduler_state.environment.feature_set.clone(),
         };
-        let transaction_scheduler =
-            TransactionScheduler::new(1, txn_scheduler_state);
+        // We dedicate half of the available resources to the execution
+        // runtime, -1 is taken up by the transaction scheduler itself
+        let transaction_executors =
+            (num_cpus::get() / 2).saturating_sub(1).max(1) as u32;
+        let transaction_scheduler = TransactionScheduler::new(
+            transaction_executors,
+            txn_scheduler_state,
+        );
         transaction_scheduler.spawn();
 
         let shared_state = SharedState::new(

magicblock-core/src/link/transactions.rs

@@ -51,6 +51,7 @@ pub type TxnReplayResultTx = oneshot::Sender<TransactionResult>;
 /// Contains the final, committed status of an executed
 /// transaction, including its result and metadata.
 /// This is the message type that is communicated to subscribers via event processors.
+#[derive(Debug)]
 pub struct TransactionStatus {
     pub signature: Signature,
     pub slot: Slot,
@@ -76,6 +77,7 @@ pub enum TransactionProcessingMode {
 }
 
 /// The detailed outcome of a standard transaction execution.
+#[derive(Debug)]
 pub struct TransactionExecutionResult {
     pub result: TransactionResult,
     pub accounts: Box<[Pubkey]>,

magicblock-processor/Cargo.toml

@@ -39,8 +39,11 @@ solana-transaction = { workspace = true }
 solana-transaction-status = { workspace = true }
 solana-transaction-error = { workspace = true }
 
+rustc-hash = { workspace = true }
+
 [dev-dependencies]
 guinea = { workspace = true }
+solana-keypair = { workspace = true }
 solana-signature = { workspace = true }
 solana-signer = { workspace = true }
 test-kit = { workspace = true }

magicblock-processor/src/executor/mod.rs

@@ -20,7 +20,8 @@ use solana_svm::transaction_processor::{
 use tokio::{runtime::Builder, sync::mpsc::Sender};
 
 use crate::{
-    builtins::BUILTINS, scheduler::state::TransactionSchedulerState, WorkerId,
+    builtins::BUILTINS,
+    scheduler::{locks::ExecutorId, state::TransactionSchedulerState},
 };
 
 /// A dedicated, single-threaded worker responsible for processing transactions using
@@ -30,7 +31,7 @@ use crate::{
 /// executors can be spawned to process transactions in parallel.
 pub(super) struct TransactionExecutor {
     /// A unique identifier for this worker instance.
-    id: WorkerId,
+    id: ExecutorId,
     /// A handle to the global accounts database for reading and writing account state.
     accountsdb: Arc<AccountsDb>,
     /// A handle to the global ledger for writing committed transaction history.
@@ -50,7 +51,7 @@ pub(super) struct TransactionExecutor {
     /// A channel to send out account state updates after processing.
     accounts_tx: AccountUpdateTx,
     /// A back-channel to notify the `TransactionScheduler` that this worker is ready for more work.
-    ready_tx: Sender<WorkerId>,
+    ready_tx: Sender<ExecutorId>,
     /// A read lock held during a slot's processing to synchronize with critical global
     /// operations like `AccountsDb` snapshots.
     sync: StWLock,
@@ -63,10 +64,10 @@ impl TransactionExecutor {
     /// with a globally shared one. This allows updates made by one executor to be immediately
     /// visible to all others, preventing redundant program loads.
     pub(super) fn new(
-        id: WorkerId,
+        id: ExecutorId,
         state: &TransactionSchedulerState,
         rx: TransactionToProcessRx,
-        ready_tx: Sender<WorkerId>,
+        ready_tx: Sender<ExecutorId>,
         programs_cache: Arc<RwLock<ProgramCache<SimpleForkGraph>>>,
     ) -> Self {
         let slot = state.accountsdb.slot();

magicblock-processor/src/lib.rs

@@ -10,8 +10,6 @@ use solana_program::feature;
 use solana_rent_collector::RentCollector;
 use solana_svm::transaction_processor::TransactionProcessingEnvironment;
 
-type WorkerId = u8;
-
 /// Initialize an SVM enviroment for transaction processing
 pub fn build_svm_env(
     accountsdb: &AccountsDb,