wip

Author: kariy

Cargo.lock

@@ -20,7 +20,7 @@ katana-tracing.workspace = true
 katana-paymaster = { workspace = true, optional = true }
 katana-utils.workspace = true
 serde-utils.workspace = true
-cartridge = { path = "../cartridge", optional = true }
+cartridge = { workspace = true, optional = true }
 
 alloy-primitives.workspace = true
 anyhow.workspace = true

crates/cli/Cargo.toml

@@ -23,7 +23,6 @@ katana-rpc-server.workspace = true
 katana-rpc-api.workspace = true
 katana-rpc-client.workspace = true
 katana-rpc-types.workspace = true
-katana-sharding.workspace = true
 katana-stage.workspace = true
 katana-tasks.workspace = true
 katana-tee = { workspace = true, optional = true }

crates/node/sequencer/Cargo.toml

@@ -16,7 +16,6 @@ pub mod tee;
 pub mod grpc;
 
 use db::DbConfig;
-use dev::DevConfig;
 use fork::ForkingConfig;
 use gateway::GatewayConfig;
 #[cfg(feature = "grpc")]

crates/node/sequencer/src/config/mod.rs

@@ -39,3 +39,6 @@ jsonrpsee.workspace = true
 serde.workspace = true
 tracing.workspace = true
 url.workspace = true
+
+[features]
+native = ["katana-executor/native", "katana-node-config/native"]

crates/node/sharding/Cargo.toml

@@ -20,7 +20,7 @@ use katana_rpc_server::shard::{ShardProvider, ShardRpc};
 use katana_rpc_server::starknet::{StarknetApi, StarknetApiConfig};
 use katana_rpc_server::{RpcServer, RpcServerHandle};
 use katana_sharding::manager::{LazyShardManager, ShardManager};
-use katana_sharding::runtime::{RuntimeHandle, ShardRuntime};
+use katana_sharding::runtime::{Runtime, RuntimeHandle};
 use katana_sharding::types::NoPendingBlockProvider;
 use katana_tasks::TaskManager;
 use parking_lot::Mutex;
@@ -38,7 +38,7 @@ pub struct Node {
     pub handle: RuntimeHandle,
     pub task_manager: TaskManager,
     pub rpc_server: RpcServer,
-    pub runtime: Mutex<Option<ShardRuntime>>,
+    pub runtime: Mutex<Option<Runtime>>,
 }
 
 impl Node {
@@ -89,12 +89,10 @@ impl Node {
             max_concurrent_estimate_fee_requests: config.rpc.max_concurrent_estimate_fee_requests,
             simulation_flags: executor_factory.execution_flags().clone(),
             versioned_constant_overrides,
-            #[cfg(feature = "cartridge")]
-            paymaster: None,
         };
 
         // --- Build runtime (scheduler + workers)
-        let runtime = ShardRuntime::new(config.worker_count, config.time_quantum);
+        let runtime = Runtime::new(config.worker_count, config.time_quantum);
         let handle = runtime.handle();
 
         // --- Build shard manager

crates/node/sharding/src/lib.rs

@@ -228,74 +228,3 @@ impl Debug for LazyShardManager {
         f.debug_struct("LazyShardManager").field("shard_count", &self.len()).finish_non_exhaustive()
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use std::sync::Arc;
-
-    use katana_chain_spec::ChainSpec;
-    use katana_executor::blockifier::cache::ClassCache;
-    use katana_executor::blockifier::BlockifierFactory;
-    use katana_executor::{BlockLimits, ExecutionFlags, ExecutorFactory};
-    use katana_gas_price_oracle::GasPriceOracle;
-    use katana_primitives::block::FinalityStatus;
-    use katana_primitives::da::L1DataAvailabilityMode;
-    use katana_primitives::env::BlockEnv;
-    use katana_primitives::{address, Felt};
-    use katana_rpc_server::starknet::StarknetApiConfig;
-    use katana_rpc_types::block::{
-        BlockWithTxHashes, GetBlockWithTxHashesResponse, PreConfirmedBlockWithTxHashes,
-    };
-    use katana_tasks::TaskManager;
-
-    use super::{block_env_from_latest_block_response, LazyShardManager, ShardManager};
-
-    #[test]
-    fn rejects_invalid_starknet_version_from_latest_block() {
-        let response = GetBlockWithTxHashesResponse::PreConfirmed(PreConfirmedBlockWithTxHashes {
-            block_number: 7,
-            timestamp: 1_234_567_890,
-            sequencer_address: address!("0x1234"),
-            l1_gas_price: resource_price(101, 102),
-            l2_gas_price: resource_price(201, 202),
-            l1_data_gas_price: resource_price(301, 302),
-            l1_da_mode: L1DataAvailabilityMode::Blob,
-            starknet_version: "not.a.version".to_owned(),
-            transactions: vec![],
-        });
-
-        let err = block_env_from_latest_block_response(response).expect_err("must fail");
-        assert!(err.to_string().contains("invalid Starknet version"));
-    }
-
-    #[tokio::test]
-    async fn get_fails_and_does_not_register_shard_when_initial_env_fetch_fails() {
-        let chain_spec = Arc::new(ChainSpec::dev());
-        let manager = LazyShardManager::new_with_block_env_fetcher(
-            chain_spec.clone(),
-            test_executor_factory(chain_spec),
-            GasPriceOracle::create_for_testing(),
-            test_starknet_api_config(),
-            TaskManager::current().task_spawner(),
-            Arc::new(|| -> anyhow::Result<BlockEnv> { anyhow::bail!("base chain unavailable") }),
-        );
-
-        let shard_id = address!("0x42");
-        let err = manager.get(shard_id).expect_err("must fail");
-
-        assert!(err.to_string().contains("failed to generate initial block context"));
-        assert!(manager.is_empty());
-        assert!(manager.shard_ids().is_empty());
-    }
-
-    fn test_executor_factory(chain_spec: Arc<ChainSpec>) -> Arc<dyn ExecutorFactory> {
-        let class_cache = ClassCache::new().expect("class cache should initialize");
-        Arc::new(BlockifierFactory::new(
-            None,
-            ExecutionFlags::new(),
-            BlockLimits::default(),
-            class_cache,
-            chain_spec,
-        ))
-    }
-}

crates/sharding/src/manager.rs

@@ -2,9 +2,11 @@ use std::sync::Arc;
 use std::thread;
 use std::time::{Duration, Instant};
 
-use crate::scheduler::ShardScheduler;
+use anyhow::Result;
+
+use crate::scheduler::Scheduler;
 use crate::types::Shard;
-use crate::worker;
+use crate::worker::Worker;
 
 /// Owns the execution resources (scheduler + worker threads) for the shard node.
 ///
@@ -19,18 +21,18 @@ use crate::worker;
 ///   so workers detach immediately.
 /// - **Drop** — if the runtime has not been consumed by one of the above methods, the `Drop` impl
 ///   signals the scheduler and blocks until all workers exit.
-pub struct ShardRuntime {
+pub struct Runtime {
     handle: RuntimeHandle,
     worker_count: usize,
-    worker_handles: Vec<thread::JoinHandle<()>>,
+    worker_handles: Vec<thread::JoinHandle<Result<()>>>,
 }
 
 /// Cheap, cloneable reference for scheduling work on the [`ShardRuntime`].
 ///
 /// Analogous to `tokio::runtime::Handle`.
 #[derive(Clone, Debug)]
 pub struct RuntimeHandle {
-    scheduler: ShardScheduler,
+    scheduler: Scheduler,
 }
 
 // ---------------------------------------------------------------------------
@@ -44,7 +46,7 @@ impl RuntimeHandle {
     }
 
     /// Returns a reference to the underlying scheduler.
-    pub fn scheduler(&self) -> &ShardScheduler {
+    pub fn scheduler(&self) -> &Scheduler {
         &self.scheduler
     }
 }
@@ -53,10 +55,10 @@ impl RuntimeHandle {
 // ShardRuntime
 // ---------------------------------------------------------------------------
 
-impl ShardRuntime {
+impl Runtime {
     /// Creates a new runtime. Workers are **not** spawned until [`start`](Self::start) is called.
     pub fn new(worker_count: usize, time_quantum: Duration) -> Self {
-        let scheduler = ShardScheduler::new(time_quantum);
+        let scheduler = Scheduler::new(time_quantum);
         let handle = RuntimeHandle { scheduler };
         Self { handle, worker_count, worker_handles: Vec::new() }
     }
@@ -69,8 +71,21 @@ impl ShardRuntime {
     /// Spawns the worker threads. Must be called exactly once.
     pub fn start(&mut self) {
         assert!(self.worker_handles.is_empty(), "ShardRuntime::start called more than once");
-        self.worker_handles =
-            worker::spawn_workers(self.worker_count, self.handle.scheduler.clone());
+
+        let total_workers = self.worker_count;
+        let scheduler_handle = self.handle.scheduler.clone();
+
+        self.worker_handles = (0..total_workers)
+            .map(|worker_id| {
+                let worker = Worker::new(worker_id, scheduler_handle.clone());
+                let worker_thread_name = format!("shard-worker-{worker_id}");
+
+                std::thread::Builder::new()
+                    .name(worker_thread_name)
+                    .spawn(move || worker.run())
+                    .expect("failed to spawn shard worker thread")
+            })
+            .collect::<Vec<thread::JoinHandle<Result<()>>>>();
     }
 
     /// Signals the scheduler to shut down, then joins worker threads up to `duration`.
@@ -109,7 +124,7 @@ impl ShardRuntime {
     }
 }
 
-impl Drop for ShardRuntime {
+impl Drop for Runtime {
     fn drop(&mut self) {
         if !self.worker_handles.is_empty() {
             self.handle.scheduler.shutdown();
@@ -120,7 +135,7 @@ impl Drop for ShardRuntime {
     }
 }
 
-impl std::fmt::Debug for ShardRuntime {
+impl std::fmt::Debug for Runtime {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("ShardRuntime")
             .field("worker_count", &self.worker_count)

crates/sharding/src/runtime.rs

@@ -12,7 +12,7 @@ use crate::types::{Shard, ShardState};
 /// Workers block on [`next_task`](ShardScheduler::next_task) using a condvar,
 /// so they must run on dedicated OS threads (not async tasks).
 #[derive(Debug, Clone)]
-pub struct ShardScheduler {
+pub struct Scheduler {
     inner: Arc<SchedulerInner>,
 }
 
@@ -23,7 +23,7 @@ struct SchedulerInner {
     shutdown: AtomicBool,
 }
 
-impl ShardScheduler {
+impl Scheduler {
     pub fn new(time_quantum: Duration) -> Self {
         Self {
             inner: Arc::new(SchedulerInner {
@@ -50,13 +50,13 @@ impl ShardScheduler {
     pub fn next_task(&self) -> Option<Arc<Shard>> {
         let mut queue = self.inner.queue.lock();
 
-        if let Some(shard) = queue.pop_front() {
-            return Some(shard);
+        if let shard @ Some(..) = queue.pop_front() {
+            shard
+        } else {
+            // Block until notified (new work or shutdown wake) or timeout.
+            self.inner.condvar.wait_for(&mut queue, Duration::from_millis(100));
+            queue.pop_front()
         }
-
-        // Block until notified (new work or shutdown wake) or timeout.
-        self.inner.condvar.wait_for(&mut queue, Duration::from_millis(100));
-        queue.pop_front()
     }
 
     /// Returns the time quantum for worker preemption.

crates/sharding/src/scheduler.rs

@@ -9,9 +9,9 @@ use katana_executor::ExecutorFactory;
 use katana_gas_price_oracle::GasPriceOracle;
 use katana_pool::ordering::FiFo;
 use katana_pool::validation::stateful::TxValidator;
-use katana_pool::TxPool;
+use katana_pool::{TransactionPool, TxPool};
 use katana_primitives::env::BlockEnv;
-use katana_primitives::transaction::TxHash;
+use katana_primitives::transaction::{ExecutableTxWithHash, TxHash};
 use katana_primitives::ContractAddress;
 use katana_provider::api::env::BlockEnvProvider;
 use katana_provider::api::state::{StateFactoryProvider, StateProvider};
@@ -211,4 +211,38 @@ impl Shard {
             .compare_exchange(expected as u8, new as u8, Ordering::SeqCst, Ordering::SeqCst)
             .is_ok()
     }
+
+    pub fn execute(&self) -> Result<()> {
+        // // Collect pending transactions from the shard's pool
+        let txs = self
+            .pool
+            .pending_transactions()
+            .all
+            .map(|ptx| (*ptx.tx).clone())
+            .collect::<Vec<ExecutableTxWithHash>>();
+
+        if txs.is_empty() {
+            return Ok(());
+        }
+
+        let tx_hashes: Vec<_> = txs.iter().map(|tx| tx.hash).collect();
+        let _tx_count = txs.len();
+
+        // // Read block env from the shard's own context
+        let block_env = self.block_env.read().clone();
+
+        let state = self.provider.provider().latest()?;
+
+        let mut executor = self.backend.executor_factory.executor(state, block_env.clone());
+
+        executor.execute_transactions(txs)?;
+        let output = executor.take_execution_output()?;
+
+        self.backend.do_mine_block(&block_env, output)?;
+
+        // Remove executed txs from pool
+        self.pool.remove_transactions(&tx_hashes);
+
+        Ok(())
+    }
 }