lib.rs

//! Reusable dev-node runner for Evolve applications.
//!
//! This module provides a complete dev-node implementation that includes:
//! - Block production with configurable intervals
//! - JSON-RPC server for Ethereum-compatible queries
//! - Chain indexing for block/transaction/receipt queries
//! - Persistent storage across restarts

pub mod cli;
pub mod config;

use std::fmt::Debug;
use std::net::SocketAddr;
use std::path::Path;
use std::sync::Arc;
use std::time::Duration;

use alloy_primitives::U256;
use borsh::{BorshDeserialize, BorshSerialize};
use commonware_runtime::tokio::{Config as TokioConfig, Context as TokioContext, Runner};
use commonware_runtime::{Runner as RunnerTrait, Spawner};
use evolve_chain_index::{
    ChainStateProvider, ChainStateProviderConfig, PersistentChainIndex, StateQuerier,
    StorageStateQuerier, DEFAULT_PROTOCOL_VERSION,
};
use evolve_core::encoding::Encodable;
use evolve_core::{AccountId, ReadonlyKV};
use evolve_eth_jsonrpc::{start_server_with_subscriptions, RpcServerConfig, SubscriptionManager};
use evolve_grpc::{GrpcServer, GrpcServerConfig};
use evolve_mempool::{new_shared_mempool, Mempool, MempoolTx, SharedMempool};
use evolve_rpc_types::SyncStatus;
use evolve_server::{
    load_chain_state, save_chain_state, state_changes_to_operations, ChainState, DevConfig,
    DevConsensus, CHAIN_STATE_KEY,
};
use evolve_server::{OnBlockArchive, StfExecutor};
use evolve_stf_traits::{AccountsCodeStorage, StateChange, Transaction};
use evolve_storage::types::BlockHash as ArchiveBlockHash;
use evolve_storage::{
    BlockStorage, BlockStorageConfig, MockStorage, Operation, Storage, StorageConfig,
};
use evolve_tx_eth::TxContext;
use std::future::Future;

pub use cli::*;
pub use config::*;

/// Default data directory for persistent storage.
pub const DEFAULT_DATA_DIR: &str = "./data";

/// Default RPC server address.
pub const DEFAULT_RPC_ADDR: &str = "127.0.0.1:8545";

fn parse_env_u64(var: &str, default: u64) -> u64 {
    std::env::var(var)
        .ok()
        .and_then(|v| v.parse::<u64>().ok())
        .filter(|v| *v > 0)
        .unwrap_or(default)
}

fn parse_env_usize(var: &str, default: usize) -> usize {
    std::env::var(var)
        .ok()
        .and_then(|v| v.parse::<usize>().ok())
        .filter(|v| *v > 0)
        .unwrap_or(default)
}

fn configured_block_interval() -> Duration {
    Duration::from_millis(parse_env_u64("EVOLVE_BLOCK_INTERVAL_MS", 1_000))
}

fn configured_max_txs_per_block() -> usize {
    parse_env_usize("EVOLVE_MAX_TXS_PER_BLOCK", 1_000)
}

/// Convenience handles for a dev node wired with a mempool.
pub struct DevNodeMempoolHandles<Stf, S, Codes, Tx: MempoolTx> {
    /// Dev consensus engine wired to mempool transactions.
    pub dev: Arc<DevConsensus<Stf, S, Codes, Tx, evolve_server::NoopChainIndex>>,
    /// Shared mempool instance.
    pub mempool: SharedMempool<Mempool<Tx>>,
}

/// Build a dev consensus + mempool pair for testing and tools.
///
/// Generic over transaction type `Tx`. For ETH transactions, use `TxContext`.
pub fn build_dev_node_with_mempool<Stf, Codes, S, Tx>(
    stf: Stf,
    storage: S,
    codes: Codes,
    config: DevConfig,
) -> DevNodeMempoolHandles<Stf, S, Codes, Tx>
where
    Tx: Transaction + MempoolTx + Encodable + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    Stf: StfExecutor<Tx, S, Codes> + Send + Sync + 'static,
{
    let mempool: SharedMempool<Mempool<Tx>> = new_shared_mempool();
    let dev = Arc::new(DevConsensus::with_mempool(
        stf,
        storage,
        codes,
        config,
        mempool.clone(),
    ));

    DevNodeMempoolHandles { dev, mempool }
}

/// Configuration for the dev node RPC server.
#[derive(Debug, Clone)]
pub struct RpcConfig {
    /// Address to bind the HTTP server to.
    pub http_addr: SocketAddr,
    /// Chain ID for eth_chainId.
    pub chain_id: u64,
    /// Whether RPC is enabled.
    pub enabled: bool,
    /// Whether block indexing is enabled while producing blocks.
    pub enable_block_indexing: bool,
    /// Optional gRPC server address. When set, a gRPC server is started
    /// alongside JSON-RPC, sharing the same state provider and subscriptions.
    pub grpc_addr: Option<SocketAddr>,
    /// Enable gzip compression for the gRPC server.
    pub grpc_enable_gzip: bool,
    /// Maximum gRPC request/response size in bytes.
    pub grpc_max_message_size: usize,
    /// Graceful shutdown timeout in seconds.
    pub shutdown_timeout_secs: u64,
}

impl Default for RpcConfig {
    fn default() -> Self {
        Self {
            http_addr: DEFAULT_RPC_ADDR.parse().unwrap(),
            chain_id: 1,
            enabled: true,
            enable_block_indexing: true,
            grpc_addr: None,
            grpc_enable_gzip: true,
            grpc_max_message_size: 4 * 1024 * 1024,
            shutdown_timeout_secs: 10,
        }
    }
}

impl RpcConfig {
    /// Create a disabled RPC config.
    pub fn disabled() -> Self {
        Self {
            enabled: false,
            ..Default::default()
        }
    }

    /// Set the HTTP address.
    pub fn with_addr(mut self, addr: SocketAddr) -> Self {
        self.http_addr = addr;
        self
    }

    /// Set the chain ID.
    pub fn with_chain_id(mut self, chain_id: u64) -> Self {
        self.chain_id = chain_id;
        self
    }

    /// Enable or disable block indexing while keeping RPC enabled.
    pub fn with_block_indexing(mut self, enabled: bool) -> Self {
        self.enable_block_indexing = enabled;
        self
    }

    /// Enable the gRPC server on the given address.
    pub fn with_grpc(mut self, addr: SocketAddr) -> Self {
        self.grpc_addr = Some(addr);
        self
    }

    /// Configure gRPC compression and message sizing.
    pub fn with_grpc_settings(mut self, enable_gzip: bool, max_message_size: usize) -> Self {
        self.grpc_enable_gzip = enable_gzip;
        self.grpc_max_message_size = max_message_size;
        self
    }

    /// Set the graceful shutdown timeout in seconds.
    pub fn with_shutdown_timeout_secs(mut self, shutdown_timeout_secs: u64) -> Self {
        self.shutdown_timeout_secs = shutdown_timeout_secs;
        self
    }
}

/// Result of a genesis run, including the state changes to commit.
pub struct GenesisOutput<G> {
    /// Application-specific genesis result.
    pub genesis_result: G,
    /// State changes produced by genesis.
    pub changes: Vec<StateChange>,
}

/// Trait for extracting the token account ID from a genesis result.
///
/// Implementing this trait on your genesis result type enables
/// `eth_getBalance` queries via the RPC server.
pub trait HasTokenAccountId {
    fn token_account_id(&self) -> AccountId;
}

type RuntimeContext = TokioContext;

/// Build the block archive callback.
///
/// Creates a `BlockStorage` backed by the commonware archive and returns
/// an `OnBlockArchive` callback that writes each produced block into it.
///
/// # Panics
///
/// Panics if block storage initialization fails. Block archival is a required
/// subsystem — all produced blocks must be persisted.
async fn build_block_archive(context: TokioContext) -> OnBlockArchive {
    let config = BlockStorageConfig::default();
    let retention = config.retention_blocks;
    let prune_interval = config.blocks_per_section;
    let store = BlockStorage::new(context, config)
        .await
        .expect("failed to initialize block archive storage");

    let (tx, mut rx) = tokio::sync::mpsc::channel::<(u64, ArchiveBlockHash, bytes::Bytes)>(64);

    // Single consumer task ensures blocks are written in order.
    tokio::spawn(async move {
        let mut store = store;
        while let Some((block_number, block_hash, block_bytes)) = rx.recv().await {
            if let Err(e) = store.put_sync(block_number, block_hash, block_bytes).await {
                tracing::warn!("Failed to archive block {}: {:?}", block_number, e);
            }

            // Prune old blocks at section boundaries to bound disk usage.
            if retention > 0 && block_number > retention && block_number % prune_interval == 0 {
                let min_block = block_number.saturating_sub(retention);
                if let Err(e) = store.prune(min_block).await {
                    tracing::warn!(min_block, "Failed to prune block archive: {:?}", e);
                }
            }
        }
    });

    tracing::info!(retention, "Block archive storage enabled");

    Arc::new(move |block_number, block_hash, block_bytes| {
        let hash_bytes = ArchiveBlockHash::new(block_hash.0);
        if let Err(e) = tx.try_send((block_number, hash_bytes, block_bytes)) {
            tracing::warn!(
                "Block archive channel full or closed for block {}: {}",
                block_number,
                e
            );
        }
    })
}

fn grpc_server_config(rpc_config: &RpcConfig, addr: SocketAddr) -> GrpcServerConfig {
    GrpcServerConfig {
        addr,
        chain_id: rpc_config.chain_id,
        enable_gzip: rpc_config.grpc_enable_gzip,
        max_message_size: rpc_config.grpc_max_message_size,
    }
}

fn shutdown_timeout(rpc_config: &RpcConfig) -> Duration {
    Duration::from_secs(rpc_config.shutdown_timeout_secs)
}

#[derive(Clone, Copy)]
enum EthRunnerMode {
    PersistentSidecars,
    EphemeralSidecars,
}

impl EthRunnerMode {
    fn enable_block_archive(self) -> bool {
        matches!(self, Self::PersistentSidecars)
    }

    fn persistent_chain_index_path(self, data_dir: &Path) -> Option<std::path::PathBuf> {
        matches!(self, Self::PersistentSidecars).then(|| data_dir.join("chain-index.sqlite"))
    }
}

#[derive(Clone)]
struct EthRunnerConfig {
    rpc: RpcConfig,
    runner_mode: EthRunnerMode,
}

/// Run the dev node with default settings (RPC enabled).
pub fn run_dev_node<
    Stf,
    Codes,
    Tx,
    G,
    S,
    BuildGenesisStf,
    BuildStf,
    BuildCodes,
    RunGenesis,
    BuildStorage,
    BuildStorageFut,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_stf: BuildStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    build_storage: BuildStorage,
) where
    Tx: Transaction + MempoolTx + Encodable + Send + Sync + 'static,
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Stf: StfExecutor<Tx, S, Codes> + Send + Sync + 'static,
    G: BorshSerialize
        + BorshDeserialize
        + Clone
        + Debug
        + HasTokenAccountId
        + Send
        + Sync
        + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildStf: Fn(&G) -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Clone + Send + Sync + 'static,
    RunGenesis: Fn(&Stf, &Codes, &S) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
    BuildStorage: Fn(RuntimeContext, StorageConfig) -> BuildStorageFut + Send + Sync + 'static,
    BuildStorageFut:
        Future<Output = Result<S, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
{
    run_dev_node_with_rpc(
        data_dir,
        build_genesis_stf,
        build_stf,
        build_codes,
        run_genesis,
        build_storage,
        RpcConfig::default(),
    )
}

/// Run the dev node with custom RPC configuration.
pub fn run_dev_node_with_rpc<
    Stf,
    Codes,
    Tx,
    G,
    S,
    BuildGenesisStf,
    BuildStf,
    BuildCodes,
    RunGenesis,
    BuildStorage,
    BuildStorageFut,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_stf: BuildStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    build_storage: BuildStorage,
    rpc_config: RpcConfig,
) where
    Tx: Transaction + MempoolTx + Encodable + Send + Sync + 'static,
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Stf: StfExecutor<Tx, S, Codes> + Send + Sync + 'static,
    G: BorshSerialize
        + BorshDeserialize
        + Clone
        + Debug
        + HasTokenAccountId
        + Send
        + Sync
        + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildStf: Fn(&G) -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Clone + Send + Sync + 'static,
    RunGenesis: Fn(&Stf, &Codes, &S) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
    BuildStorage: Fn(RuntimeContext, StorageConfig) -> BuildStorageFut + Send + Sync + 'static,
    BuildStorageFut:
        Future<Output = Result<S, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
{
    tracing::info!("=== Evolve Dev Node ===");
    let data_dir = data_dir.as_ref();
    std::fs::create_dir_all(data_dir).expect("failed to create data directory");

    let storage_config = StorageConfig {
        path: data_dir.to_path_buf(),
        ..Default::default()
    };
    let chain_index_db_path = data_dir.join("chain-index.sqlite");

    let runtime_config = TokioConfig::default()
        .with_storage_directory(data_dir)
        .with_worker_threads(4); // More threads for RPC handling

    let runner = Runner::new(runtime_config);

    let build_genesis_stf = Arc::new(build_genesis_stf);
    let build_stf = Arc::new(build_stf);
    let build_codes = Arc::new(build_codes);
    let run_genesis = Arc::new(run_genesis);
    let build_storage = Arc::new(build_storage);

    runner.start(move |context| {
        let build_genesis_stf = Arc::clone(&build_genesis_stf);
        let build_stf = Arc::clone(&build_stf);
        let build_codes = Arc::clone(&build_codes);
        let run_genesis = Arc::clone(&run_genesis);
        let build_storage = Arc::clone(&build_storage);
        let rpc_config = rpc_config.clone();
        let chain_index_db_path = chain_index_db_path.clone();

        async move {
            // Clone context early since build_storage takes ownership
            let context_for_shutdown = context.clone();
            let context_for_archive = context.clone();
            let storage = (build_storage)(context, storage_config)
                .await
                .expect("failed to create storage");

            // Create account codes
            let codes = build_codes();

            let (genesis_result, initial_height) = match load_chain_state::<G, _>(&storage) {
                Some(state) => {
                    tracing::info!("Resuming from existing state at height {}", state.height);
                    tracing::info!("Genesis state: {:?}", state.genesis_result);
                    (state.genesis_result, state.height)
                }
                None => {
                    tracing::info!("No existing state found, running genesis...");
                    let bootstrap_stf = (build_genesis_stf)();
                    let output =
                        (run_genesis)(&bootstrap_stf, &codes, &storage).expect("genesis failed");

                    commit_genesis(&storage, output.changes, &output.genesis_result)
                        .await
                        .expect("genesis commit failed");

                    tracing::info!("Genesis complete. Result: {:?}", output.genesis_result);
                    (output.genesis_result, 1)
                }
            };

            // Build STF for normal execution
            let stf = (build_stf)(&genesis_result);

            // Create DevConsensus config
            let block_interval = configured_block_interval();
            let dev_config = DevConfig {
                block_interval: Some(block_interval),
                initial_height,
                chain_id: rpc_config.chain_id,
                ..Default::default()
            };

            // Build block archive callback (always on)
            let archive_cb = build_block_archive(context_for_archive).await;

            // Set up RPC infrastructure if enabled
            let rpc_handle = if rpc_config.enabled {
                // Create chain index backed by SQLite
                let chain_index = Arc::new(
                    PersistentChainIndex::new(&chain_index_db_path)
                        .expect("failed to open chain index database"),
                );

                // Initialize from existing data
                if let Err(e) = chain_index.initialize() {
                    tracing::warn!("Failed to initialize chain index: {:?}", e);
                }

                // Create subscription manager for real-time events
                let subscriptions = Arc::new(SubscriptionManager::new());

                // Create state provider for RPC
                let codes_for_rpc = Arc::new(build_codes());
                let state_provider_config = ChainStateProviderConfig {
                    chain_id: rpc_config.chain_id,
                    protocol_version: DEFAULT_PROTOCOL_VERSION.to_string(),
                    gas_price: U256::ZERO,
                    sync_status: SyncStatus::NotSyncing(false),
                };
                let state_querier: Arc<dyn StateQuerier> = Arc::new(StorageStateQuerier::new(
                    storage.clone(),
                    genesis_result.token_account_id(),
                ));
                let state_provider = ChainStateProvider::with_account_codes(
                    Arc::clone(&chain_index),
                    state_provider_config.clone(),
                    Arc::clone(&codes_for_rpc),
                )
                .with_state_querier(Arc::clone(&state_querier));

                // Start JSON-RPC server
                let server_config = RpcServerConfig {
                    http_addr: rpc_config.http_addr,
                    chain_id: rpc_config.chain_id,
                };

                tracing::info!("Starting JSON-RPC server on {}", rpc_config.http_addr);
                let handle = start_server_with_subscriptions(
                    server_config,
                    state_provider,
                    Arc::clone(&subscriptions),
                )
                .await
                .expect("failed to start RPC server");

                let grpc_handle = if let Some(grpc_addr) = rpc_config.grpc_addr {
                    let grpc_state_provider = ChainStateProvider::with_account_codes(
                        Arc::clone(&chain_index),
                        state_provider_config,
                        codes_for_rpc,
                    )
                    .with_state_querier(state_querier);
                    let grpc_config = grpc_server_config(&rpc_config, grpc_addr);
                    tracing::info!("Starting gRPC server on {}", grpc_addr);
                    let grpc_server = GrpcServer::with_subscription_manager(
                        grpc_config,
                        grpc_state_provider,
                        Arc::clone(&subscriptions),
                    );
                    Some(tokio::spawn(async move {
                        if let Err(e) = grpc_server.serve().await {
                            tracing::error!("gRPC server error: {}", e);
                        }
                    }))
                } else {
                    None
                };

                // Create DevConsensus with RPC support
                let consensus = DevConsensus::with_rpc(
                    stf,
                    storage,
                    codes,
                    dev_config,
                    chain_index,
                    subscriptions,
                )
                .with_indexing_enabled(rpc_config.enable_block_indexing)
                .with_block_archive(archive_cb);
                let dev: Arc<DevConsensus<Stf, S, Codes, Tx, PersistentChainIndex>> =
                    Arc::new(consensus);

                tracing::info!(
                    "Block interval: {:?}, starting at height {}",
                    block_interval,
                    initial_height
                );

                tracing::info!("Starting block production... (Ctrl+C to stop)");

                // Run block production and Ctrl+C handling concurrently using Spawner pattern.
                // When Ctrl+C is received, stop() triggers shutdown signal via context.stopped().
                tokio::select! {
                    _ = dev.run_block_production(context_for_shutdown.clone()) => {
                        // Block production exited
                    }
                    _ = tokio::signal::ctrl_c() => {
                        tracing::info!("Received Ctrl+C, initiating graceful shutdown...");
                        context_for_shutdown
                            .stop(0, Some(shutdown_timeout(&rpc_config)))
                            .await
                            .expect("shutdown failed");
                    }
                }

                // Save chain state
                let final_height = dev.height();
                tracing::info!("Stopped at height: {}", final_height);

                let chain_state = ChainState {
                    height: final_height,
                    genesis_result,
                };
                if let Err(e) = save_chain_state(dev.storage(), &chain_state).await {
                    tracing::error!("Failed to save chain state: {}", e);
                } else {
                    tracing::info!("Saved chain state at height {}", final_height);
                }

                Some((handle, grpc_handle))
            } else {
                // No RPC - use simple DevConsensus
                let consensus = DevConsensus::new(stf, storage, codes, dev_config)
                    .with_block_archive(archive_cb);
                let dev: Arc<DevConsensus<Stf, S, Codes, Tx, evolve_server::NoopChainIndex>> =
                    Arc::new(consensus);

                tracing::info!(
                    "Block interval: {:?}, starting at height {}",
                    block_interval,
                    initial_height
                );

                tracing::info!("Starting block production... (Ctrl+C to stop)");

                // Run block production and Ctrl+C handling concurrently using Spawner pattern
                tokio::select! {
                    _ = dev.run_block_production(context_for_shutdown.clone()) => {
                        // Block production exited
                    }
                    _ = tokio::signal::ctrl_c() => {
                        tracing::info!("Received Ctrl+C, initiating graceful shutdown...");
                        context_for_shutdown
                            .stop(0, Some(shutdown_timeout(&rpc_config)))
                            .await
                            .expect("shutdown failed");
                    }
                }

                let final_height = dev.height();
                tracing::info!("Stopped at height: {}", final_height);

                let chain_state = ChainState {
                    height: final_height,
                    genesis_result,
                };
                if let Err(e) = save_chain_state(dev.storage(), &chain_state).await {
                    tracing::error!("Failed to save chain state: {}", e);
                } else {
                    tracing::info!("Saved chain state at height {}", final_height);
                }

                None
            };

            // Stop RPC server if running
            if let Some((handle, grpc_handle)) = rpc_handle {
                tracing::info!("Stopping RPC server...");
                handle.stop().expect("failed to stop RPC server");
                tracing::info!("RPC server stopped");
                if let Some(grpc_handle) = grpc_handle {
                    tracing::info!("Stopping gRPC server...");
                    grpc_handle.abort();
                    tracing::info!("gRPC server stopped");
                }
            }
        }
    });
}

/// Run the dev node with RPC and mempool-enabled transaction ingestion.
///
/// Generic over transaction type `Tx`. For ETH transactions, use
/// `run_dev_node_with_rpc_and_mempool_eth` for convenience.
///
/// Note: When using a custom `Tx` type with RPC enabled, the RPC layer
/// will still use `TxContext` for `eth_sendRawTransaction`. For custom
/// transaction types, consider disabling RPC or providing a custom gateway.
pub fn run_dev_node_with_rpc_and_mempool<
    Stf,
    Codes,
    Tx,
    G,
    S,
    BuildGenesisStf,
    BuildStf,
    BuildCodes,
    RunGenesis,
    BuildStorage,
    BuildStorageFut,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_stf: BuildStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    build_storage: BuildStorage,
    rpc_config: RpcConfig,
) where
    Tx: Transaction + MempoolTx + Encodable + Send + Sync + 'static,
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Stf: StfExecutor<Tx, S, Codes> + Send + Sync + 'static,
    G: BorshSerialize + BorshDeserialize + Clone + Debug + Send + Sync + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildStf: Fn(&G) -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Clone + Send + Sync + 'static,
    RunGenesis: Fn(&Stf, &Codes, &S) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
    BuildStorage: Fn(RuntimeContext, StorageConfig) -> BuildStorageFut + Send + Sync + 'static,
    BuildStorageFut:
        Future<Output = Result<S, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
{
    tracing::info!("=== Evolve Dev Node (mempool) ===");
    let data_dir = data_dir.as_ref();
    std::fs::create_dir_all(data_dir).expect("failed to create data directory");

    let storage_config = StorageConfig {
        path: data_dir.to_path_buf(),
        ..Default::default()
    };

    let runtime_config = TokioConfig::default()
        .with_storage_directory(data_dir)
        .with_worker_threads(4);

    let runner = Runner::new(runtime_config);

    let build_genesis_stf = Arc::new(build_genesis_stf);
    let build_stf = Arc::new(build_stf);
    let build_codes = Arc::new(build_codes);
    let run_genesis = Arc::new(run_genesis);
    let build_storage = Arc::new(build_storage);

    runner.start(move |context| {
        let build_genesis_stf = Arc::clone(&build_genesis_stf);
        let build_stf = Arc::clone(&build_stf);
        let build_codes = Arc::clone(&build_codes);
        let run_genesis = Arc::clone(&run_genesis);
        let build_storage = Arc::clone(&build_storage);
        let rpc_config = rpc_config.clone();

        async move {
            let context_for_shutdown = context.clone();
            let storage = (build_storage)(context, storage_config)
                .await
                .expect("failed to create storage");

            let codes = build_codes();

            let (genesis_result, initial_height) = match load_chain_state::<G, _>(&storage) {
                Some(state) => {
                    tracing::info!("Resuming from existing state at height {}", state.height);
                    tracing::info!("Genesis state: {:?}", state.genesis_result);
                    (state.genesis_result, state.height)
                }
                None => {
                    tracing::info!("No existing state found, running genesis...");
                    let bootstrap_stf = (build_genesis_stf)();
                    let output =
                        (run_genesis)(&bootstrap_stf, &codes, &storage).expect("genesis failed");

                    commit_genesis(&storage, output.changes, &output.genesis_result)
                        .await
                        .expect("genesis commit failed");

                    tracing::info!("Genesis complete. Result: {:?}", output.genesis_result);
                    (output.genesis_result, 1)
                }
            };

            let stf = (build_stf)(&genesis_result);

            let block_interval = configured_block_interval();
            let max_txs_per_block = configured_max_txs_per_block();
            let dev_config = DevConfig {
                block_interval: Some(block_interval),
                initial_height,
                chain_id: rpc_config.chain_id,
                ..Default::default()
            };

            let mempool: SharedMempool<Mempool<Tx>> = new_shared_mempool();

            // Note: RPC with custom Tx types is not fully supported.
            // The RPC layer requires TxContext for eth_sendRawTransaction.
            // For custom Tx types, use run_dev_node_with_rpc_and_mempool_eth instead.
            if rpc_config.enabled {
                tracing::warn!(
                    "RPC enabled with generic Tx type. eth_sendRawTransaction will not work. \
                    Use run_dev_node_with_rpc_and_mempool_eth for ETH transactions with full RPC support."
                );
            }

            let dev: Arc<DevConsensus<Stf, S, Codes, Tx, evolve_server::NoopChainIndex>> =
                Arc::new(DevConsensus::with_mempool(stf, storage, codes, dev_config, mempool));

            tracing::info!(
                "Block interval: {:?}, max_txs_per_block: {}, starting at height {}",
                block_interval,
                max_txs_per_block,
                initial_height
            );

            tracing::info!("Starting block production... (Ctrl+C to stop)");

            tokio::select! {
                _ = dev.run_block_production_with_mempool(context_for_shutdown.clone(), max_txs_per_block) => {
                }
                _ = tokio::signal::ctrl_c() => {
                    tracing::info!("Received Ctrl+C, initiating graceful shutdown...");
                    context_for_shutdown
                        .stop(0, Some(shutdown_timeout(&rpc_config)))
                        .await
                        .expect("shutdown failed");
                }
            }

            let final_height = dev.height();
            tracing::info!("Stopped at height: {}", final_height);

            let chain_state = ChainState {
                height: final_height,
                genesis_result,
            };
            if let Err(e) = save_chain_state(dev.storage(), &chain_state).await {
                tracing::error!("Failed to save chain state: {}", e);
            } else {
                tracing::info!("Saved chain state at height {}", final_height);
            }
        }
    });
}

/// Run the dev node with RPC and mempool for ETH transactions (TxContext).
///
/// This is a convenience wrapper around `run_dev_node_with_rpc_and_mempool`
/// that uses `TxContext` as the transaction type and sets up the full
/// ETH JSON-RPC server with `eth_sendRawTransaction` support.
pub fn run_dev_node_with_rpc_and_mempool_eth<
    Stf,
    Codes,
    G,
    S,
    BuildGenesisStf,
    BuildStf,
    BuildCodes,
    RunGenesis,
    BuildStorage,
    BuildStorageFut,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_stf: BuildStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    build_storage: BuildStorage,
    rpc_config: RpcConfig,
) where
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Stf: StfExecutor<TxContext, S, Codes> + Send + Sync + 'static,
    G: BorshSerialize
        + BorshDeserialize
        + Clone
        + Debug
        + HasTokenAccountId
        + Send
        + Sync
        + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildStf: Fn(&G) -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Clone + Send + Sync + 'static,
    RunGenesis: Fn(&Stf, &Codes, &S) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
    BuildStorage: Fn(RuntimeContext, StorageConfig) -> BuildStorageFut + Send + Sync + 'static,
    BuildStorageFut:
        Future<Output = Result<S, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
{
    run_dev_node_with_rpc_and_mempool_eth_impl(
        data_dir,
        build_genesis_stf,
        build_stf,
        build_codes,
        run_genesis,
        build_storage,
        EthRunnerConfig {
            rpc: rpc_config,
            runner_mode: EthRunnerMode::PersistentSidecars,
        },
    )
}

fn run_dev_node_with_rpc_and_mempool_eth_impl<
    Stf,
    Codes,
    G,
    S,
    BuildGenesisStf,
    BuildStf,
    BuildCodes,
    RunGenesis,
    BuildStorage,
    BuildStorageFut,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_stf: BuildStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    build_storage: BuildStorage,
    runner_config: EthRunnerConfig,
) where
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Stf: StfExecutor<TxContext, S, Codes> + Send + Sync + 'static,
    G: BorshSerialize
        + BorshDeserialize
        + Clone
        + Debug
        + HasTokenAccountId
        + Send
        + Sync
        + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildStf: Fn(&G) -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Clone + Send + Sync + 'static,
    RunGenesis: Fn(&Stf, &Codes, &S) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
    BuildStorage: Fn(RuntimeContext, StorageConfig) -> BuildStorageFut + Send + Sync + 'static,
    BuildStorageFut:
        Future<Output = Result<S, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
{
    tracing::info!("=== Evolve Dev Node (ETH mempool) ===");
    let data_dir = data_dir.as_ref();
    std::fs::create_dir_all(data_dir).expect("failed to create data directory");

    let storage_config = StorageConfig {
        path: data_dir.to_path_buf(),
        ..Default::default()
    };
    let chain_index_db_path = runner_config
        .runner_mode
        .persistent_chain_index_path(data_dir);

    let runtime_config = TokioConfig::default()
        .with_storage_directory(data_dir)
        .with_worker_threads(4);

    let runner = Runner::new(runtime_config);

    let build_genesis_stf = Arc::new(build_genesis_stf);
    let build_stf = Arc::new(build_stf);
    let build_codes = Arc::new(build_codes);
    let run_genesis = Arc::new(run_genesis);
    let build_storage = Arc::new(build_storage);

    runner.start(move |context| {
        let build_genesis_stf = Arc::clone(&build_genesis_stf);
        let build_stf = Arc::clone(&build_stf);
        let build_codes = Arc::clone(&build_codes);
        let run_genesis = Arc::clone(&run_genesis);
        let build_storage = Arc::clone(&build_storage);
        let rpc_config = runner_config.rpc.clone();
        let chain_index_db_path = chain_index_db_path.clone();
        let runner_mode = runner_config.runner_mode;

        async move {
            let context_for_shutdown = context.clone();
            let context_for_archive = context.clone();
            let storage = (build_storage)(context, storage_config)
                .await
                .expect("failed to create storage");

            let codes = build_codes();

            let (genesis_result, initial_height) = match load_chain_state::<G, _>(&storage) {
                Some(state) => {
                    tracing::info!("Resuming from existing state at height {}", state.height);
                    tracing::info!("Genesis state: {:?}", state.genesis_result);
                    (state.genesis_result, state.height)
                }
                None => {
                    tracing::info!("No existing state found, running genesis...");
                    let bootstrap_stf = (build_genesis_stf)();
                    let output =
                        (run_genesis)(&bootstrap_stf, &codes, &storage).expect("genesis failed");

                    commit_genesis(&storage, output.changes, &output.genesis_result)
                        .await
                        .expect("genesis commit failed");

                    tracing::info!("Genesis complete. Result: {:?}", output.genesis_result);
                    (output.genesis_result, 1)
                }
            };

            let stf = (build_stf)(&genesis_result);

            let block_interval = configured_block_interval();
            let max_txs_per_block = configured_max_txs_per_block();
            let dev_config = DevConfig {
                block_interval: Some(block_interval),
                initial_height,
                chain_id: rpc_config.chain_id,
                ..Default::default()
            };

            let mempool: SharedMempool<Mempool<TxContext>> = new_shared_mempool();

            let archive_cb = if runner_mode.enable_block_archive() {
                Some(build_block_archive(context_for_archive).await)
            } else {
                tracing::info!(
                    "Block archive disabled for mock-storage runner to keep sidecars ephemeral"
                );
                None
            };

            let rpc_handle = if rpc_config.enabled {
                let chain_index = Arc::new(
                    match chain_index_db_path.as_ref() {
                        Some(path) => PersistentChainIndex::new(path),
                        None => PersistentChainIndex::in_memory(),
                    }
                    .expect("failed to open chain index database"),
                );

                if let Err(e) = chain_index.initialize() {
                    tracing::warn!("Failed to initialize chain index: {:?}", e);
                }

                let subscriptions = Arc::new(SubscriptionManager::new());

                let codes_for_rpc = Arc::new(build_codes());
                let state_provider_config = ChainStateProviderConfig {
                    chain_id: rpc_config.chain_id,
                    protocol_version: DEFAULT_PROTOCOL_VERSION.to_string(),
                    gas_price: U256::ZERO,
                    sync_status: SyncStatus::NotSyncing(false),
                };

                // Create state querier for balance/nonce reads
                let state_querier: Arc<dyn crate::StateQuerier> = Arc::new(
                    StorageStateQuerier::new(
                        storage.clone(),
                        genesis_result.token_account_id(),
                    ),
                );

                let state_provider = ChainStateProvider::with_mempool(
                    Arc::clone(&chain_index),
                    state_provider_config.clone(),
                    Arc::clone(&codes_for_rpc),
                    mempool.clone(),
                )
                .with_state_querier(Arc::clone(&state_querier));

                let server_config = RpcServerConfig {
                    http_addr: rpc_config.http_addr,
                    chain_id: rpc_config.chain_id,
                };

                tracing::info!("Starting JSON-RPC server on {}", rpc_config.http_addr);
                let handle = start_server_with_subscriptions(
                    server_config,
                    state_provider,
                    Arc::clone(&subscriptions),
                )
                .await
                .expect("failed to start RPC server");

                // Start gRPC server if configured
                let grpc_handle = if let Some(grpc_addr) = rpc_config.grpc_addr {
                    let grpc_state_provider = ChainStateProvider::with_mempool(
                        Arc::clone(&chain_index),
                        state_provider_config,
                        codes_for_rpc,
                        mempool.clone(),
                    )
                    .with_state_querier(state_querier);
                    let grpc_config = grpc_server_config(&rpc_config, grpc_addr);
                    tracing::info!("Starting gRPC server on {}", grpc_addr);
                    let grpc_server = GrpcServer::with_subscription_manager(
                        grpc_config,
                        grpc_state_provider,
                        Arc::clone(&subscriptions),
                    );
                    Some(tokio::spawn(async move {
                        if let Err(e) = grpc_server.serve().await {
                            tracing::error!("gRPC server error: {}", e);
                        }
                    }))
                } else {
                    None
                };
                let consensus = DevConsensus::with_rpc_and_mempool(
                    stf,
                    storage,
                    codes,
                    dev_config,
                    chain_index,
                    subscriptions,
                    mempool.clone(),
                )
                .with_indexing_enabled(rpc_config.enable_block_indexing);
                let consensus = match archive_cb {
                    Some(archive_cb) => consensus.with_block_archive(archive_cb),
                    None => consensus,
                };
                let dev: Arc<DevConsensus<Stf, S, Codes, TxContext, PersistentChainIndex>> =
                    Arc::new(consensus);

                tracing::info!(
                    "Block interval: {:?}, max_txs_per_block: {}, starting at height {}",
                    block_interval,
                    max_txs_per_block,
                    initial_height
                );

                tracing::info!("Starting block production... (Ctrl+C to stop)");

                tokio::select! {
                    _ = dev.run_block_production_with_mempool(context_for_shutdown.clone(), max_txs_per_block) => {
                    }
                    _ = tokio::signal::ctrl_c() => {
                        tracing::info!("Received Ctrl+C, initiating graceful shutdown...");
                        context_for_shutdown
                            .stop(0, Some(shutdown_timeout(&rpc_config)))
                            .await
                            .expect("shutdown failed");
                    }
                }

                let final_height = dev.height();
                tracing::info!("Stopped at height: {}", final_height);

                let chain_state = ChainState {
                    height: final_height,
                    genesis_result,
                };
                if let Err(e) = save_chain_state(dev.storage(), &chain_state).await {
                    tracing::error!("Failed to save chain state: {}", e);
                } else {
                    tracing::info!("Saved chain state at height {}", final_height);
                }

                Some((handle, grpc_handle))
            } else {
                let consensus = DevConsensus::with_mempool(stf, storage, codes, dev_config, mempool);
                let consensus = match archive_cb {
                    Some(archive_cb) => consensus.with_block_archive(archive_cb),
                    None => consensus,
                };
                let dev: Arc<DevConsensus<Stf, S, Codes, TxContext, evolve_server::NoopChainIndex>> =
                    Arc::new(consensus);

                tracing::info!(
                    "Block interval: {:?}, max_txs_per_block: {}, starting at height {}",
                    block_interval,
                    max_txs_per_block,
                    initial_height
                );

                tracing::info!("Starting block production... (Ctrl+C to stop)");

                tokio::select! {
                    _ = dev.run_block_production_with_mempool(context_for_shutdown.clone(), max_txs_per_block) => {
                    }
                    _ = tokio::signal::ctrl_c() => {
                        tracing::info!("Received Ctrl+C, initiating graceful shutdown...");
                        context_for_shutdown
                            .stop(0, Some(shutdown_timeout(&rpc_config)))
                            .await
                            .expect("shutdown failed");
                    }
                }

                let final_height = dev.height();
                tracing::info!("Stopped at height: {}", final_height);

                let chain_state = ChainState {
                    height: final_height,
                    genesis_result,
                };
                if let Err(e) = save_chain_state(dev.storage(), &chain_state).await {
                    tracing::error!("Failed to save chain state: {}", e);
                } else {
                    tracing::info!("Saved chain state at height {}", final_height);
                }

                None
            };

            if let Some((handle, grpc_handle)) = rpc_handle {
                tracing::info!("Stopping RPC server...");
                handle.stop().expect("failed to stop RPC server");
                tracing::info!("RPC server stopped");
                if let Some(grpc_handle) = grpc_handle {
                    tracing::info!("Stopping gRPC server...");
                    grpc_handle.abort();
                    tracing::info!("gRPC server stopped");
                }
            }
        }
    });
}

/// Run the dev node with RPC and ETH mempool using in-memory mock storage.
pub fn run_dev_node_with_rpc_and_mempool_mock_storage<
    Stf,
    Codes,
    G,
    BuildGenesisStf,
    BuildStf,
    BuildCodes,
    RunGenesis,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_stf: BuildStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    rpc_config: RpcConfig,
) where
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    Stf: StfExecutor<TxContext, MockStorage, Codes> + Send + Sync + 'static,
    G: BorshSerialize
        + BorshDeserialize
        + Clone
        + Debug
        + HasTokenAccountId
        + Send
        + Sync
        + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildStf: Fn(&G) -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Clone + Send + Sync + 'static,
    RunGenesis: Fn(
            &Stf,
            &Codes,
            &MockStorage,
        ) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
{
    run_dev_node_with_rpc_and_mempool_eth_impl(
        data_dir,
        build_genesis_stf,
        build_stf,
        build_codes,
        run_genesis,
        |_context, _config| async { Ok(MockStorage::new()) },
        EthRunnerConfig {
            rpc: rpc_config,
            runner_mode: EthRunnerMode::EphemeralSidecars,
        },
    )
}

/// Initialize genesis in the data directory and exit.
pub fn init_dev_node<
    Stf,
    Codes,
    Tx,
    G,
    S,
    BuildGenesisStf,
    BuildCodes,
    RunGenesis,
    BuildStorage,
    BuildStorageFut,
>(
    data_dir: impl AsRef<Path>,
    build_genesis_stf: BuildGenesisStf,
    build_codes: BuildCodes,
    run_genesis: RunGenesis,
    build_storage: BuildStorage,
) where
    Tx: Transaction + MempoolTx + Encodable + Send + Sync + 'static,
    Codes: AccountsCodeStorage + Send + Sync + 'static,
    S: ReadonlyKV + Storage + Clone + Send + Sync + 'static,
    Stf: StfExecutor<Tx, S, Codes> + Send + Sync + 'static,
    G: BorshSerialize + BorshDeserialize + Clone + Debug + Send + Sync + 'static,
    BuildGenesisStf: Fn() -> Stf + Send + Sync + 'static,
    BuildCodes: Fn() -> Codes + Send + Sync + 'static,
    RunGenesis: Fn(&Stf, &Codes, &S) -> Result<GenesisOutput<G>, Box<dyn std::error::Error + Send + Sync>>
        + Send
        + Sync
        + 'static,
    BuildStorage: Fn(RuntimeContext, StorageConfig) -> BuildStorageFut + Send + Sync + 'static,
    BuildStorageFut:
        Future<Output = Result<S, Box<dyn std::error::Error + Send + Sync>>> + Send + 'static,
{
    tracing::info!("=== Evolve Dev Node (init) ===");
    let data_dir = data_dir.as_ref();
    std::fs::create_dir_all(data_dir).expect("failed to create data directory");

    let storage_config = StorageConfig {
        path: data_dir.to_path_buf(),
        ..Default::default()
    };

    let runtime_config = TokioConfig::default()
        .with_storage_directory(data_dir)
        .with_worker_threads(1);

    let runner = Runner::new(runtime_config);

    let build_genesis_stf = Arc::new(build_genesis_stf);
    let build_codes = Arc::new(build_codes);
    let run_genesis = Arc::new(run_genesis);
    let build_storage = Arc::new(build_storage);

    runner.start(move |context| {
        let build_genesis_stf = Arc::clone(&build_genesis_stf);
        let build_codes = Arc::clone(&build_codes);
        let run_genesis = Arc::clone(&run_genesis);
        let build_storage = Arc::clone(&build_storage);

        async move {
            let storage = (build_storage)(context, storage_config)
                .await
                .expect("failed to create storage");

            if load_chain_state::<G, _>(&storage).is_some() {
                tracing::error!("State already initialized; refusing to re-run genesis");
                return;
            }

            let codes = build_codes();
            let bootstrap_stf = (build_genesis_stf)();
            let output = (run_genesis)(&bootstrap_stf, &codes, &storage).expect("genesis failed");

            commit_genesis(&storage, output.changes, &output.genesis_result)
                .await
                .expect("genesis commit failed");

            tracing::info!("Genesis complete. Result: {:?}", output.genesis_result);
        }
    });
}

async fn commit_genesis<G: BorshSerialize + Clone, S: Storage>(
    storage: &S,
    changes: Vec<StateChange>,
    genesis_result: &G,
) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    let mut operations = state_changes_to_operations(changes);

    let chain_state: ChainState<G> = ChainState {
        height: 1,
        genesis_result: genesis_result.clone(),
    };
    operations.push(Operation::Set {
        key: CHAIN_STATE_KEY.to_vec(),
        value: borsh::to_vec(&chain_state).map_err(|e| format!("serialize chain state: {e}"))?,
    });

    storage
        .batch(operations)
        .await
        .map_err(|e| format!("batch failed: {:?}", e))?;

    storage
        .commit()
        .await
        .map_err(|e| format!("commit failed: {:?}", e))?;
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use alloy_consensus::{SignableTransaction, TxLegacy};
    use alloy_primitives::{Address, Bytes, U256, U64};
    use evolve_chain_index::NoopAccountCodes;
    use evolve_eth_jsonrpc::StateProvider;
    use evolve_mempool::MempoolTx;
    use evolve_rpc_types::block::BlockTransactions;
    use evolve_rpc_types::SyncStatus;
    use evolve_server::{Block, DevConsensus, NoopChainIndex, StfExecutor};
    use evolve_stf::execution_state::ExecutionState;
    use evolve_stf::results::{BlockResult, TxResult};
    use evolve_stf_traits::{AccountsCodeStorage, Block as _};
    use k256::ecdsa::SigningKey;
    use rand::rngs::OsRng;
    use tokio::time::{sleep, timeout, Duration};

    #[derive(Default, Clone, Copy)]
    struct MockStf;

    #[derive(Default)]
    struct MockCodes;

    impl AccountsCodeStorage for MockCodes {
        fn with_code<F, R>(&self, _identifier: &str, f: F) -> Result<R, evolve_core::ErrorCode>
        where
            F: FnOnce(Option<&dyn evolve_core::AccountCode>) -> R,
        {
            Ok(f(None))
        }

        fn list_identifiers(&self) -> Vec<String> {
            Vec::new()
        }
    }

    impl StfExecutor<TxContext, MockStorage, MockCodes> for MockStf {
        fn execute_block<'a>(
            &self,
            storage: &'a MockStorage,
            _codes: &'a MockCodes,
            block: &Block<TxContext>,
        ) -> (BlockResult, ExecutionState<'a, MockStorage>) {
            let tx_results: Vec<TxResult> = block
                .txs()
                .iter()
                .map(|tx| TxResult {
                    events: vec![],
                    gas_used: MempoolTx::gas_limit(tx),
                    response: evolve_core::InvokeResponse::new(&()),
                })
                .collect();

            let gas_used = tx_results.iter().map(|r| r.gas_used).sum();
            (
                BlockResult {
                    begin_block_events: vec![],
                    tx_results,
                    end_block_events: vec![],
                    gas_used,
                    txs_skipped: 0,
                },
                ExecutionState::new(storage),
            )
        }
    }

    use evolve_tx_eth::sign_hash;

    fn make_signed_legacy_tx(chain_id: u64) -> Vec<u8> {
        make_signed_legacy_tx_with_gas(chain_id, 21_000)
    }

    fn make_signed_legacy_tx_with_gas(chain_id: u64, gas_limit: u64) -> Vec<u8> {
        let signing_key = SigningKey::random(&mut OsRng);
        let tx = TxLegacy {
            chain_id: Some(chain_id),
            nonce: 0,
            gas_price: 1,
            gas_limit,
            to: alloy_primitives::TxKind::Call(Address::repeat_byte(0x11)),
            value: U256::from(1u64),
            input: Bytes::new(),
        };

        let signature = sign_hash(&signing_key, tx.signature_hash());
        let signed = tx.into_signed(signature);

        let mut encoded = Vec::new();
        signed.rlp_encode(&mut encoded);
        encoded
    }

    fn test_provider_config() -> ChainStateProviderConfig {
        ChainStateProviderConfig {
            chain_id: 1,
            protocol_version: "evolve-test".to_string(),
            gas_price: U256::from(1u64),
            sync_status: SyncStatus::NotSyncing(false),
        }
    }

    #[tokio::test]
    async fn mempool_to_block_to_provider_query_roundtrip() {
        let chain_index = Arc::new(PersistentChainIndex::in_memory().expect("in-memory index"));
        let mempool = new_shared_mempool::<TxContext>();
        let provider = ChainStateProvider::with_mempool(
            Arc::clone(&chain_index),
            test_provider_config(),
            Arc::new(NoopAccountCodes),
            mempool.clone(),
        );
        let subscriptions = Arc::new(SubscriptionManager::new());

        let dev = DevConsensus::with_rpc_and_mempool(
            MockStf,
            MockStorage::new(),
            MockCodes,
            DevConfig::default(),
            Arc::clone(&chain_index),
            subscriptions,
            mempool.clone(),
        );

        let raw_tx = make_signed_legacy_tx(1);
        let tx_hash = provider
            .send_raw_transaction(&raw_tx)
            .await
            .expect("raw tx should be accepted into mempool");

        assert_eq!(mempool.read().await.len(), 1);

        let produced = dev
            .produce_block_from_mempool(100)
            .await
            .expect("block production should succeed");
        assert_eq!(produced.tx_count, 1);

        assert_eq!(mempool.read().await.len(), 0);

        timeout(Duration::from_secs(2), async {
            loop {
                if provider
                    .get_transaction_by_hash(tx_hash)
                    .await
                    .expect("provider query should not fail")
                    .is_some()
                {
                    break;
                }
                sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("indexed transaction should become visible");

        let tx = provider
            .get_transaction_by_hash(tx_hash)
            .await
            .expect("query should succeed")
            .expect("transaction should be indexed");
        assert_eq!(tx.hash, tx_hash);
        assert_eq!(tx.block_number, Some(U64::from(1u64)));
        assert_eq!(tx.nonce, U64::ZERO);
        assert_eq!(tx.gas_price, Some(U256::from(1u64)));
        assert_eq!(tx.tx_type, U64::ZERO);
        assert_ne!(tx.r, U256::ZERO);
        assert_ne!(tx.s, U256::ZERO);

        let receipt = provider
            .get_transaction_receipt(tx_hash)
            .await
            .expect("receipt query should succeed")
            .expect("receipt should be indexed");
        assert_eq!(receipt.transaction_hash, tx_hash);
        assert_eq!(receipt.block_number, U64::from(1u64));
        assert_eq!(receipt.status, U64::from(1u64));
        assert_eq!(receipt.effective_gas_price, U256::from(1u64));
        assert_eq!(receipt.tx_type, U64::ZERO);

        let block = provider
            .get_block_by_number(1, false)
            .await
            .expect("block query should succeed")
            .expect("block should be indexed");
        match block.transactions {
            Some(BlockTransactions::Hashes(hashes)) => assert_eq!(hashes, vec![tx_hash]),
            _ => panic!("expected block transaction hashes"),
        }
    }

    #[tokio::test]
    async fn produce_block_from_mempool_keeps_over_budget_transactions_queued() {
        let chain_index = Arc::new(PersistentChainIndex::in_memory().expect("in-memory index"));
        let mempool = new_shared_mempool::<TxContext>();
        let provider = ChainStateProvider::with_mempool(
            Arc::clone(&chain_index),
            test_provider_config(),
            Arc::new(NoopAccountCodes),
            mempool.clone(),
        );

        let low_gas_dev: DevConsensus<_, _, _, _, NoopChainIndex> = DevConsensus::with_mempool(
            MockStf,
            MockStorage::new(),
            MockCodes,
            DevConfig {
                gas_limit: 21_000,
                ..DevConfig::manual()
            },
            mempool.clone(),
        );
        let high_gas_dev: DevConsensus<_, _, _, _, NoopChainIndex> = DevConsensus::with_mempool(
            MockStf,
            MockStorage::new(),
            MockCodes,
            DevConfig {
                gas_limit: 30_000,
                ..DevConfig::manual()
            },
            mempool.clone(),
        );

        let executed_hash = provider
            .send_raw_transaction(&make_signed_legacy_tx_with_gas(1, 21_000))
            .await
            .expect("21k tx should be accepted into mempool");
        let deferred_hash = provider
            .send_raw_transaction(&make_signed_legacy_tx_with_gas(1, 30_000))
            .await
            .expect("30k tx should be accepted into mempool");

        assert_eq!(mempool.read().await.len(), 2);

        let first_block = low_gas_dev
            .produce_block_from_mempool(100)
            .await
            .expect("low-gas block production should succeed");
        assert_eq!(first_block.tx_count, 1);
        assert_eq!(first_block.gas_used, 21_000);

        let pool = mempool.read().await;
        assert_eq!(pool.len(), 1);
        assert!(!pool.contains(&executed_hash.0));
        assert!(pool.contains(&deferred_hash.0));
        drop(pool);

        let second_block = high_gas_dev
            .produce_block_from_mempool(100)
            .await
            .expect("higher-gas block production should execute the deferred tx");
        assert_eq!(second_block.tx_count, 1);
        assert_eq!(second_block.gas_used, 30_000);

        let pool = mempool.read().await;
        assert_eq!(pool.len(), 0);
        assert!(!pool.contains(&executed_hash.0));
        assert!(!pool.contains(&deferred_hash.0));
    }

    #[tokio::test]
    async fn provider_rejects_wrong_chain_id_without_mempool_insert() {
        let chain_index = Arc::new(PersistentChainIndex::in_memory().expect("in-memory index"));
        let mempool = new_shared_mempool::<TxContext>();
        let provider = ChainStateProvider::with_mempool(
            Arc::clone(&chain_index),
            test_provider_config(),
            Arc::new(NoopAccountCodes),
            mempool.clone(),
        );

        let wrong_chain_tx = make_signed_legacy_tx(2);
        let err = provider
            .send_raw_transaction(&wrong_chain_tx)
            .await
            .expect_err("wrong-chain tx must be rejected");
        let msg = err.to_string();
        assert!(
            msg.contains("expected 1") && msg.contains("got Some(2)"),
            "unexpected error message: {msg}"
        );
        assert_eq!(mempool.read().await.len(), 0);
    }

    #[tokio::test]
    async fn provider_returns_full_transactions_for_indexed_block() {
        let chain_index = Arc::new(PersistentChainIndex::in_memory().expect("in-memory index"));
        let mempool = new_shared_mempool::<TxContext>();
        let provider = ChainStateProvider::with_mempool(
            Arc::clone(&chain_index),
            test_provider_config(),
            Arc::new(NoopAccountCodes),
            mempool.clone(),
        );
        let subscriptions = Arc::new(SubscriptionManager::new());

        let dev = DevConsensus::with_rpc_and_mempool(
            MockStf,
            MockStorage::new(),
            MockCodes,
            DevConfig::default(),
            Arc::clone(&chain_index),
            subscriptions,
            mempool.clone(),
        );

        let raw_tx = make_signed_legacy_tx(1);
        let tx_hash = provider
            .send_raw_transaction(&raw_tx)
            .await
            .expect("raw tx should be accepted into mempool");
        dev.produce_block_from_mempool(100)
            .await
            .expect("block production should succeed");

        timeout(Duration::from_secs(2), async {
            loop {
                let maybe = provider
                    .get_block_by_number(1, true)
                    .await
                    .expect("query should succeed");
                if maybe.is_some() {
                    break;
                }
                sleep(Duration::from_millis(20)).await;
            }
        })
        .await
        .expect("block should become visible");

        let block = provider
            .get_block_by_number(1, true)
            .await
            .expect("query should succeed")
            .expect("block should be indexed");
        match block.transactions {
            Some(BlockTransactions::Full(txs)) => {
                assert_eq!(txs.len(), 1);
                assert_eq!(txs[0].hash, tx_hash);
                assert_eq!(txs[0].block_number, Some(U64::from(1u64)));
                assert_eq!(txs[0].gas_price, Some(U256::from(1u64)));
                assert_eq!(txs[0].nonce, U64::ZERO);
                assert_eq!(txs[0].tx_type, U64::ZERO);
                assert_ne!(txs[0].r, U256::ZERO);
                assert_ne!(txs[0].s, U256::ZERO);
            }
            _ => panic!("expected full transactions in block"),
        }
    }
}