Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

enha: capture error metrics in rpc_subscriptions_active and remove the method wrapper #1942

Merged
merged 3 commits into from
Jan 7, 2025
Merged

enha: capture error metrics in rpc_subscriptions_active and remove the method wrapper #1942

merged 3 commits into from
Jan 7, 2025

Conversation

carneiro-cw
Copy link
Contributor

@carneiro-cw carneiro-cw commented Jan 7, 2025 •
edited by github-actions bot
Loading

PR Type

Enhancement

Description

  • Add major_error_code attribute to ErrorCode derive macro

  • Implement new error code system with major codes

  • Remove method wrapper and related metrics

  • Update RPC methods to use new error system

Changes walkthrough 📝

Relevant files
Enhancement
lib.rs
Enhance ErrorCode derive macro with major error codes       

crates/stratus_macros/src/lib.rs

  • Added major_error_code attribute to ErrorCode derive macro
  • Implemented major error code in error code generation
  • Updated reverse mapping to return Option<&'static str>
  • Modified test case to include major_error_code
    		• +32/-13 
    stratus_error.rs
    Implement new error code system with major codes                 

    src/eth/primitives/stratus_error.rs

  • Added major_error_code attribute to error enums
  • Removed ErrorCode implementation for StratusError
  • Implemented new ErrorCode trait for StratusError
    		• +30/-17 
    mod.rs
    Remove rpc_method_wrapper module import                                   

    src/eth/rpc/mod.rs

    • Removed import of rpc_method_wrapper module
    		 +0/-1     
    rpc_method_wrapper.rs
    Remove rpc_method_wrapper.rs file                                               

    src/eth/rpc/rpc_method_wrapper.rs

    • Entire file removed
    		 +0/-39   
    rpc_middleware.rs
    Update RPC middleware for new error code system                   

    src/eth/rpc/rpc_middleware.rs

  • Updated error code handling in RPC response
  • Added import for ErrorCode trait
    		• +2/-1     
    rpc_server.rs
    Update RPC server for new error handling system                   

    src/eth/rpc/rpc_server.rs

  • Removed method wrapper and related imports
  • Updated RPC method registrations
  • Modified RPC methods to use new error system
    		• +29/-45 
    metrics_definitions.rs
    Remove rpc_error_response metric                                                 

    src/infra/metrics/metrics_definitions.rs

    • Removed rpc_error_response metric
    		 +1/-4     

    💡 PR-Agent usage: Comment /help "your question" on any pull request to receive relevant information

    @github-actions GitHub Actions
    Copy link

    github-actions bot commented Jan 7, 2025

    PR Reviewer Guide 🔍

    Here are some key observations to aid the review process:

    ⏱️ Estimated effort to review: 4 🔵🔵🔵🔵⚪
    🧪 No relevant tests
    🔒 No security concerns identified
    ⚡ Recommended focus areas for review

    Error Handling

    The removal of the metrics wrapper and changes to error handling might affect how errors are logged and tracked. Ensure that error logging and metrics are still properly implemented throughout the RPC methods.

        module.register_method("web3_clientVersion", web3_client_version)?;

    // gas
    module.register_method("eth_gasPrice", eth_gas_price)?;

    // stratus importing helpers
    module.register_blocking_method("stratus_getBlockAndReceipts", stratus_get_block_and_receipts)?;

    // block
    module.register_blocking_method("eth_blockNumber", eth_block_number)?;
    module.register_blocking_method("eth_getBlockByNumber", eth_get_block_by_number)?;
    module.register_blocking_method("eth_getBlockByHash", eth_get_block_by_hash)?;
    module.register_method("eth_getUncleByBlockHashAndIndex", eth_get_uncle_by_block_hash_and_index)?;

    // transactions
    module.register_blocking_method("eth_getTransactionByHash", eth_get_transaction_by_hash)?;
    module.register_blocking_method("eth_getTransactionReceipt", eth_get_transaction_receipt)?;
    module.register_blocking_method("eth_estimateGas", eth_estimate_gas)?;
    module.register_blocking_method("eth_call", eth_call)?;
    module.register_blocking_method("eth_sendRawTransaction", eth_send_raw_transaction)?;

    // logs
    module.register_blocking_method("eth_getLogs", eth_get_logs)?;

    // account
    module.register_method("eth_accounts", eth_accounts)?;
    module.register_blocking_method("eth_getTransactionCount", eth_get_transaction_count)?;
    module.register_blocking_method("eth_getBalance", eth_get_balance)?;
    module.register_blocking_method("eth_getCode", eth_get_code)?;

    // storage
    module.register_blocking_method("eth_getStorageAt", eth_get_storage_at)?;

    // subscriptions
    module.register_subscription("eth_subscribe", "eth_subscription", "eth_unsubscribe", eth_subscribe)?;

    Ok(module)
}

// -----------------------------------------------------------------------------
// Debug
// -----------------------------------------------------------------------------

#[cfg(feature = "dev")]
fn evm_mine(_params: Params<'_>, ctx: Arc<RpcContext>, _: Extensions) -> Result<JsonValue, StratusError> {
    ctx.miner.mine_local_and_commit()?;
    Ok(to_json_value(true))
}

#[cfg(feature = "dev")]
fn evm_set_next_block_timestamp(params: Params<'_>, ctx: Arc<RpcContext>, _: Extensions) -> Result<JsonValue, StratusError> {
    use crate::eth::primitives::UnixTime;
    use crate::log_and_err;

    let (_, timestamp) = next_rpc_param::<UnixTime>(params.sequence())?;
    let latest = ctx.storage.read_block(BlockFilter::Latest)?;
    match latest {
        Some(block) => UnixTime::set_offset(block.header.timestamp, timestamp)?,
        None => return log_and_err!("reading latest block returned None")?,
    }
    Ok(to_json_value(timestamp))
}

// -----------------------------------------------------------------------------
// Status - Health checks
// -----------------------------------------------------------------------------

async fn stratus_health(_: Params<'_>, context: Arc<RpcContext>, _: Extensions) -> Result<JsonValue, StratusError> {
    if GlobalState::is_shutdown() {
        tracing::warn!("liveness check failed because of shutdown");
        return Err(StateError::StratusShutdown.into());
    }

    let should_serve = match GlobalState::get_node_mode() {
        NodeMode::Leader | NodeMode::FakeLeader => true,
        NodeMode::Follower => match context.consensus() {
            Some(consensus) => consensus.should_serve().await,
            None => false,
        },
    };

    if not(should_serve) {
        tracing::warn!("readiness check failed because consensus is not ready");
        metrics::set_consensus_is_ready(0_u64);
        return Err(StateError::StratusNotReady.into());
    }

    metrics::set_consensus_is_ready(1_u64);
    Ok(json!(true))
}

// -----------------------------------------------------------------------------
// Stratus - Admin
// -----------------------------------------------------------------------------

#[cfg(feature = "dev")]
fn stratus_reset(_: Params<'_>, ctx: Arc<RpcContext>, _: Extensions) -> Result<JsonValue, StratusError> {
    ctx.storage.reset_to_genesis()?;
    Ok(to_json_value(true))
}

static MODE_CHANGE_SEMAPHORE: Lazy<Semaphore> = Lazy::new(|| Semaphore::new(1));

async fn stratus_change_to_leader(_: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    ext.authentication().auth_admin()?;
    let permit = MODE_CHANGE_SEMAPHORE.try_acquire();
    let _permit: SemaphorePermit = match permit {
        Ok(permit) => permit,
        Err(_) => return Err(StateError::ModeChangeInProgress.into()),
    };

    const LEADER_MINER_INTERVAL: Duration = Duration::from_secs(1);
    tracing::info!("starting process to change node to leader");

    if GlobalState::get_node_mode() == NodeMode::Leader {
        tracing::info!("node is already in leader mode, no changes made");
        return Ok(json!(false));
    }

    if GlobalState::is_transactions_enabled() {
        tracing::error!("transactions are currently enabled, cannot change node mode");
        return Err(StateError::TransactionsEnabled.into());
    }

    tracing::info!("shutting down importer");
    let shutdown_importer_result = stratus_shutdown_importer(Params::new(None), &ctx, &ext);
    match shutdown_importer_result {
        Ok(_) => tracing::info!("importer shutdown successfully"),
        Err(StratusError::Importer(ImporterError::AlreadyShutdown)) => {
            tracing::warn!("importer is already shutdown, continuing");
        }
        Err(e) => {
            tracing::error!(reason = ?e, "failed to shutdown importer");
            return Err(e);
        }
    }

    tracing::info!("wait for importer to shutdown");
    GlobalState::wait_for_importer_to_finish().await;

    let change_miner_mode_result = change_miner_mode(MinerMode::Interval(LEADER_MINER_INTERVAL), &ctx).await;
    if let Err(e) = change_miner_mode_result {
        tracing::error!(reason = ?e, "failed to change miner mode");
        return Err(e);
    }
    tracing::info!("miner mode changed to interval(1s) successfully");

    GlobalState::set_node_mode(NodeMode::Leader);
    tracing::info!("node mode changed to leader successfully");

    Ok(json!(true))
}

async fn stratus_change_to_follower(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    ext.authentication().auth_admin()?;
    let permit = MODE_CHANGE_SEMAPHORE.try_acquire();
    let _permit: SemaphorePermit = match permit {
        Ok(permit) => permit,
        Err(_) => return Err(StateError::ModeChangeInProgress.into()),
    };

    tracing::info!("starting process to change node to follower");

    if GlobalState::get_node_mode() == NodeMode::Follower {
        tracing::info!("node is already in follower mode, no changes made");
        return Ok(json!(false));
    }

    if GlobalState::is_transactions_enabled() {
        tracing::error!("transactions are currently enabled, cannot change node mode");
        return Err(StateError::TransactionsEnabled.into());
    }

    let pending_txs = ctx.storage.pending_transactions();
    if not(pending_txs.is_empty()) {
        tracing::error!(pending_txs = ?pending_txs.len(), "cannot change to follower mode with pending transactions");
        return Err(StorageError::PendingTransactionsExist {
            pending_txs: pending_txs.len(),
        }
        .into());
    }

    let change_miner_mode_result = change_miner_mode(MinerMode::External, &ctx).await;
    if let Err(e) = change_miner_mode_result {
        tracing::error!(reason = ?e, "failed to change miner mode");
        return Err(e);
    }
    tracing::info!("miner mode changed to external successfully");

    GlobalState::set_node_mode(NodeMode::Follower);

    tracing::info!("initializing importer");
    let init_importer_result = stratus_init_importer(params, Arc::clone(&ctx), ext).await;
    match init_importer_result {
        Ok(_) => tracing::info!("importer initialized successfully"),
        Err(StratusError::Importer(ImporterError::AlreadyRunning)) => {
            tracing::warn!("importer is already running, continuing");
        }
        Err(e) => {
            tracing::error!(reason = ?e, "failed to initialize importer, reverting node mode to leader");
            GlobalState::set_node_mode(NodeMode::Leader);
            return Err(e);
        }
    }
    tracing::info!("node mode changed to follower successfully");

    Ok(json!(true))
}

async fn stratus_init_importer(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    ext.authentication().auth_admin()?;
    let (params, external_rpc) = next_rpc_param::<String>(params.sequence())?;
    let (params, external_rpc_ws) = next_rpc_param::<String>(params)?;
    let (params, raw_external_rpc_timeout) = next_rpc_param::<String>(params)?;
    let (_, raw_sync_interval) = next_rpc_param::<String>(params)?;

    let external_rpc_timeout = parse_duration(&raw_external_rpc_timeout).map_err(|e| {
        tracing::error!(reason = ?e, "failed to parse external_rpc_timeout");
        ImporterError::ConfigParseError
    })?;

    let sync_interval = parse_duration(&raw_sync_interval).map_err(|e| {
        tracing::error!(reason = ?e, "failed to parse sync_interval");
        ImporterError::ConfigParseError
    })?;

    let importer_config = ImporterConfig {
        external_rpc,
        external_rpc_ws: Some(external_rpc_ws),
        external_rpc_timeout,
        sync_interval,
    };

    importer_config.init_follower_importer(ctx).await
}

fn stratus_shutdown_importer(_: Params<'_>, ctx: &RpcContext, ext: &Extensions) -> Result<JsonValue, StratusError> {
    ext.authentication().auth_admin()?;
    if GlobalState::get_node_mode() != NodeMode::Follower {
        tracing::error!("node is currently not a follower");
        return Err(StateError::StratusNotFollower.into());
    }

    if GlobalState::is_importer_shutdown() && ctx.consensus().is_none() {
        tracing::error!("importer is already shut down");
        return Err(ImporterError::AlreadyShutdown.into());
    }

    ctx.set_consensus(None);

    const TASK_NAME: &str = "rpc-server::importer-shutdown";
    GlobalState::shutdown_importer_from(TASK_NAME, "received importer shutdown request");

    Ok(json!(true))
}

async fn stratus_change_miner_mode(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    ext.authentication().auth_admin()?;
    let (_, mode_str) = next_rpc_param::<String>(params.sequence())?;

    let mode = MinerMode::from_str(&mode_str).map_err(|e| {
        tracing::error!(reason = ?e, "failed to parse miner mode");
        RpcError::MinerModeParamInvalid
    })?;

    change_miner_mode(mode, &ctx).await
}

/// Tries changing miner mode, returns `Ok(true)` if changed, and `Ok(false)` if no changing was necessary.
///
/// This function also enables the miner after changing it.
async fn change_miner_mode(new_mode: MinerMode, ctx: &RpcContext) -> Result<JsonValue, StratusError> {
    if GlobalState::is_transactions_enabled() {
        tracing::error!("cannot change miner mode while transactions are enabled");
        return Err(StateError::TransactionsEnabled.into());
    }

    let previous_mode = ctx.miner.mode();

    if previous_mode == new_mode {
        tracing::warn!(?new_mode, current = ?new_mode, "miner mode already set, skipping");
        return Ok(json!(false));
    }

    if not(ctx.miner.is_paused()) && previous_mode.is_interval() {
        return log_and_err!("can't change miner mode from Interval without pausing it first").map_err(Into::into);
    }

    match new_mode {
        MinerMode::External => {
            tracing::info!("changing miner mode to External");

            let pending_txs = ctx.storage.pending_transactions();
            if not(pending_txs.is_empty()) {
                tracing::error!(pending_txs = ?pending_txs.len(), "cannot change miner mode to External with pending transactions");
                return Err(StorageError::PendingTransactionsExist {
                    pending_txs: pending_txs.len(),
                }
                .into());
            }

            ctx.miner.switch_to_external_mode().await;
        }
        MinerMode::Interval(duration) => {
            tracing::info!(duration = ?duration, "changing miner mode to Interval");

            if ctx.consensus().is_some() {
                tracing::error!("cannot change miner mode to Interval with consensus set");
                return Err(ConsensusError::Set.into());
            }

            ctx.miner.start_interval_mining(duration).await;
        }
        MinerMode::Automine => {
            return log_and_err!("Miner mode change to 'automine' is unsupported.").map_err(Into::into);
        }
    }

    Ok(json!(true))
}

fn stratus_enable_unknown_clients(_: Params<'_>, _: &RpcContext, ext: &Extensions) -> Result<bool, StratusError> {
    ext.authentication().auth_admin()?;
    GlobalState::set_unknown_client_enabled(true);
    Ok(GlobalState::is_unknown_client_enabled())
}

fn stratus_disable_unknown_clients(_: Params<'_>, _: &RpcContext, ext: &Extensions) -> Result<bool, StratusError> {
    ext.authentication().auth_admin()?;
    GlobalState::set_unknown_client_enabled(false);
    Ok(GlobalState::is_unknown_client_enabled())
}

fn stratus_enable_transactions(_: Params<'_>, _: &RpcContext, ext: &Extensions) -> Result<bool, StratusError> {
    ext.authentication().auth_admin()?;
    GlobalState::set_transactions_enabled(true);
    Ok(GlobalState::is_transactions_enabled())
}

fn stratus_disable_transactions(_: Params<'_>, _: &RpcContext, ext: &Extensions) -> Result<bool, StratusError> {
    ext.authentication().auth_admin()?;
    GlobalState::set_transactions_enabled(false);
    Ok(GlobalState::is_transactions_enabled())
}

fn stratus_enable_miner(_: Params<'_>, ctx: &RpcContext, ext: &Extensions) -> Result<bool, StratusError> {
    ext.authentication().auth_admin()?;
    ctx.miner.unpause();
    Ok(true)
}

fn stratus_disable_miner(_: Params<'_>, ctx: &RpcContext, ext: &Extensions) -> Result<bool, StratusError> {
    ext.authentication().auth_admin()?;
    ctx.miner.pause();
    Ok(false)
}

/// Returns the count of executed transactions waiting to enter the next block.
fn stratus_pending_transactions_count(_: Params<'_>, ctx: &RpcContext, _: &Extensions) -> usize {
    ctx.storage.pending_transactions().len()
}

// -----------------------------------------------------------------------------
// Stratus - State
// -----------------------------------------------------------------------------

fn stratus_version(_: Params<'_>, _: &RpcContext, _: &Extensions) -> Result<JsonValue, StratusError> {
    Ok(build_info::as_json())
}

fn stratus_config(_: Params<'_>, ctx: &RpcContext, _: &Extensions) -> Result<JsonValue, StratusError> {
    Ok(ctx.app_config.clone())
}

fn stratus_state(_: Params<'_>, ctx: &RpcContext, _: &Extensions) -> Result<JsonValue, StratusError> {
    Ok(GlobalState::get_global_state_as_json(ctx))
}

async fn stratus_get_subscriptions(_: Params<'_>, ctx: Arc<RpcContext>, _: Extensions) -> Result<JsonValue, StratusError> {
    // NOTE: this is a workaround for holding only one lock at a time
    let pending_txs = serde_json::to_value(ctx.subs.new_heads.read().await.values().collect_vec()).expect_infallible();
    let new_heads = serde_json::to_value(ctx.subs.pending_txs.read().await.values().collect_vec()).expect_infallible();
    let logs = serde_json::to_value(ctx.subs.logs.read().await.values().flat_map(HashMap::values).collect_vec()).expect_infallible();

    let response = json!({
        "newPendingTransactions": pending_txs,
        "newHeads": new_heads,
        "logs": logs,
    });
    Ok(response)
}

// -----------------------------------------------------------------------------
// Blockchain
// -----------------------------------------------------------------------------

async fn net_listening(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    let net_listening = stratus_health(params, ctx, ext).await;

    tracing::info!(net_listening = ?net_listening, "network listening status");

    net_listening
}

fn net_version(_: Params<'_>, ctx: &RpcContext, _: &Extensions) -> String {
    ctx.chain_id.to_string()
}

fn eth_chain_id(_: Params<'_>, ctx: &RpcContext, _: &Extensions) -> String {
    hex_num(ctx.chain_id)
}

fn web3_client_version(_: Params<'_>, ctx: &RpcContext, _: &Extensions) -> String {
    ctx.client_version.to_owned()
}

// -----------------------------------------------------------------------------
// Gas
// -----------------------------------------------------------------------------

fn eth_gas_price(_: Params<'_>, _: &RpcContext, _: &Extensions) -> String {
    hex_zero()
}

// -----------------------------------------------------------------------------
// Block
// -----------------------------------------------------------------------------

fn eth_block_number(_params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_blockNumber", block_number = field::Empty).entered();

    // execute
    let block_number = ctx.storage.read_mined_block_number()?;
    Span::with(|s| s.rec_str("block_number", &block_number));

    Ok(to_json_value(block_number))
}

fn stratus_get_block_and_receipts(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::stratus_getBlockAndReceipts").entered();

    // parse params
    let (_, filter) = next_rpc_param::<BlockFilter>(params.sequence())?;

    // track
    tracing::info!(%filter, "reading block and receipts");

    let Some(block) = ctx.storage.read_block(filter)? else {
        tracing::info!(%filter, "block not found");
        return Ok(JsonValue::Null);
    };

    tracing::info!(%filter, "block with transactions found");
    let receipts = block.transactions.iter().cloned().map(EthersReceipt::from).collect::<Vec<_>>();

    Ok(json!({
        "block": block.to_json_rpc_with_full_transactions(),
        "receipts": receipts,
    }))
}

fn eth_get_block_by_hash(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    eth_get_block_by_selector::<'h'>(params, ctx, ext)
}

fn eth_get_block_by_number(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    eth_get_block_by_selector::<'n'>(params, ctx, ext)
}

#[inline(always)]
fn eth_get_block_by_selector<const KIND: char>(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = if KIND == 'h' {
        info_span!(
            "rpc::eth_getBlockByHash",
            filter = field::Empty,
            block_number = field::Empty,
            found = field::Empty
        )
        .entered()
    } else {
        info_span!(
            "rpc::eth_getBlockByNumber",
            filter = field::Empty,
            block_number = field::Empty,
            found = field::Empty
        )
        .entered()
    };

    // parse params
    let (params, filter) = next_rpc_param::<BlockFilter>(params.sequence())?;
    let (_, full_transactions) = next_rpc_param::<bool>(params)?;

    // track
    Span::with(|s| s.rec_str("filter", &filter));
    tracing::info!(%filter, %full_transactions, "reading block");

    // execute
    let block = ctx.storage.read_block(filter)?;
    Span::with(|s| {
        s.record("found", block.is_some());
        if let Some(ref block) = block {
            s.rec_str("block_number", &block.number());
        }
    });
    match (block, full_transactions) {
        (Some(block), true) => {
            tracing::info!(%filter, "block with full transactions found");
            Ok(block.to_json_rpc_with_full_transactions())
        }
        (Some(block), false) => {
            tracing::info!(%filter, "block with only hashes found");
            Ok(block.to_json_rpc_with_transactions_hashes())
        }
        (None, _) => {
            tracing::info!(%filter, "block not found");
            Ok(JsonValue::Null)
        }
    }
}

fn eth_get_uncle_by_block_hash_and_index(_: Params<'_>, _: &RpcContext, _: &Extensions) -> Result<JsonValue, StratusError> {
    Ok(JsonValue::Null)
}

// -----------------------------------------------------------------------------
// Transaction
// -----------------------------------------------------------------------------

fn eth_get_transaction_by_hash(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_getTransactionByHash", tx_hash = field::Empty, found = field::Empty).entered();

    // parse params
    let (_, tx_hash) = next_rpc_param::<Hash>(params.sequence())?;

    // track
    Span::with(|s| s.rec_str("tx_hash", &tx_hash));
    tracing::info!(%tx_hash, "reading transaction");

    // execute
    let tx = ctx.storage.read_transaction(tx_hash)?;
    Span::with(|s| {
        s.record("found", tx.is_some());
    });

    match tx {
        Some(tx) => {
            tracing::info!(%tx_hash, "transaction found");
            Ok(tx.to_json_rpc_transaction())
        }
        None => {
            tracing::info!(%tx_hash, "transaction not found");
            Ok(JsonValue::Null)
        }
    }
}

fn eth_get_transaction_receipt(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_getTransactionReceipt", tx_hash = field::Empty, found = field::Empty).entered();

    // parse params
    let (_, tx_hash) = next_rpc_param::<Hash>(params.sequence())?;

    // track
    Span::with(|s| s.rec_str("tx_hash", &tx_hash));
    tracing::info!(%tx_hash, "reading transaction receipt");

    // execute
    let tx = ctx.storage.read_transaction(tx_hash)?;
    Span::with(|s| {
        s.record("found", tx.is_some());
    });

    match tx {
        Some(tx) => {
            tracing::info!(%tx_hash, "transaction receipt found");
            Ok(tx.to_json_rpc_receipt())
        }
        None => {
            tracing::info!(%tx_hash, "transaction receipt not found");
            Ok(JsonValue::Null)
        }
    }
}

fn eth_estimate_gas(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_estimateGas", tx_from = field::Empty, tx_to = field::Empty).entered();

    // parse params
    let (_, call) = next_rpc_param::<CallInput>(params.sequence())?;

    // track
    Span::with(|s| {
        s.rec_opt("tx_from", &call.from);
        s.rec_opt("tx_to", &call.to);
    });
    tracing::info!("executing eth_estimateGas");

    // execute
    match ctx.executor.execute_local_call(call, PointInTime::Mined) {
        // result is success
        Ok(result) if result.is_success() => {
            tracing::info!(tx_output = %result.output, "executed eth_estimateGas with success");
            let overestimated_gas = (result.gas.as_u64()) as f64 * 1.1;
            Ok(hex_num(overestimated_gas as u64))
        }

        // result is failure
        Ok(result) => {
            tracing::warn!(tx_output = %result.output, "executed eth_estimateGas with failure");
            Err(TransactionError::Reverted { output: result.output }.into())
        }

        // internal error
        Err(e) => {
            tracing::warn!(reason = ?e, "failed to execute eth_estimateGas");
            Err(e)
        }
    }
}

fn eth_call(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_call", tx_from = field::Empty, tx_to = field::Empty, filter = field::Empty).entered();

    // parse params
    let (params, call) = next_rpc_param::<CallInput>(params.sequence())?;
    let (_, filter) = next_rpc_param_or_default::<BlockFilter>(params)?;

    // track
    Span::with(|s| {
        s.rec_opt("tx_from", &call.from);
        s.rec_opt("tx_to", &call.to);
        s.rec_str("filter", &filter);
    });
    tracing::info!(%filter, "executing eth_call");

    // execute
    let point_in_time = ctx.storage.translate_to_point_in_time(filter)?;
    match ctx.executor.execute_local_call(call, point_in_time) {
        // result is success
        Ok(result) if result.is_success() => {
            tracing::info!(tx_output = %result.output, "executed eth_call with success");
            Ok(hex_data(result.output))
        }

        // result is failure
        Ok(result) => {
            tracing::warn!(tx_output = %result.output, "executed eth_call with failure");
            Err(TransactionError::Reverted { output: result.output }.into())
        }

        // internal error
        Err(e) => {
            tracing::warn!(reason = ?e, "failed to execute eth_call");
            Err(e)
        }
    }
}

fn eth_send_raw_transaction(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!(
        "rpc::eth_sendRawTransaction",
        tx_hash = field::Empty,
        tx_from = field::Empty,
        tx_to = field::Empty,
        tx_nonce = field::Empty
    )
    .entered();

    // parse params
    let (_, tx_data) = next_rpc_param::<Bytes>(params.sequence())?;
    let tx = parse_rpc_rlp::<TransactionInput>(&tx_data)?;
    let tx_hash = tx.hash;

    // track
    Span::with(|s| {
        s.rec_str("tx_hash", &tx_hash);
        s.rec_str("tx_from", &tx.signer);
        s.rec_opt("tx_to", &tx.to);
        s.rec_str("tx_nonce", &tx.nonce);
    });

    if not(GlobalState::is_transactions_enabled()) {
        tracing::warn!(%tx_hash, "failed to execute eth_sendRawTransaction because transactions are disabled");
        return Err(StateError::TransactionsDisabled.into());
    }

    // execute locally or forward to leader
    match GlobalState::get_node_mode() {
        NodeMode::Leader | NodeMode::FakeLeader => match ctx.executor.execute_local_transaction(tx) {
            Ok(_) => Ok(hex_data(tx_hash)),
            Err(e) => {
                tracing::warn!(reason = ?e, "failed to execute eth_sendRawTransaction");
                Err(e)
            }
        },
        NodeMode::Follower => match ctx.consensus() {
            Some(consensus) => match Handle::current().block_on(consensus.forward_to_leader(tx_hash, tx_data, ext.rpc_client())) {
                Ok(hash) => Ok(hex_data(hash)),
                Err(e) => Err(e),
            },
            None => {
                tracing::error!("unable to forward transaction because consensus is temporarily unavailable for follower node");
                Err(ConsensusError::Unavailable.into())
            }
        },
    }
}

// -----------------------------------------------------------------------------
// Logs
// -----------------------------------------------------------------------------

fn eth_get_logs(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<JsonValue, StratusError> {
    const MAX_BLOCK_RANGE: u64 = 5_000;

    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!(
        "rpc::eth_getLogs",
        filter = field::Empty,
        filter_from = field::Empty,
        filter_to = field::Empty,
        filter_range = field::Empty
    )
    .entered();

    // parse params
    let (_, filter_input) = next_rpc_param_or_default::<LogFilterInput>(params.sequence())?;
    let mut filter = filter_input.parse(&ctx.storage)?;

    // for this operation, the filter always need the end block specified to calculate the difference
    let to_block = match filter.to_block {
        Some(block) => block,
        None => {
            let block = ctx.storage.read_mined_block_number()?;
            filter.to_block = Some(block);
            block
        }
    };
    let blocks_in_range = filter.from_block.count_to(to_block);

    // track
    Span::with(|s| {
        s.rec_str("filter", &to_json_string(&filter));
        s.rec_str("filter_from", &filter.from_block);
        s.rec_str("filter_to", &to_block);
        s.rec_str("filter_range", &blocks_in_range);
    });
    tracing::info!(?filter, "reading logs");

    // check range
    if blocks_in_range > MAX_BLOCK_RANGE {
        return Err(RpcError::BlockRangeInvalid {
            actual: blocks_in_range,
            max: MAX_BLOCK_RANGE,
        }
        .into());
    }

    // execute
    let logs = ctx.storage.read_logs(&filter)?;
    Ok(JsonValue::Array(logs.into_iter().map(|x| x.to_json_rpc_log()).collect()))
}

// -----------------------------------------------------------------------------
// Account
// -----------------------------------------------------------------------------

fn eth_accounts(_: Params<'_>, _ctx: &RpcContext, _: &Extensions) -> Result<JsonValue, StratusError> {
    Ok(json!([]))
}

fn eth_get_transaction_count(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_getTransactionCount", address = field::Empty, filter = field::Empty).entered();

    // pare params
    let (params, address) = next_rpc_param::<Address>(params.sequence())?;
    let (_, filter) = next_rpc_param_or_default::<BlockFilter>(params)?;

    // track
    Span::with(|s| {
        s.rec_str("address", &address);
        s.rec_str("address", &filter);
    });
    tracing::info!(%address, %filter, "reading account nonce");

    let point_in_time = ctx.storage.translate_to_point_in_time(filter)?;
    let account = ctx.storage.read_account(address, point_in_time)?;
    Ok(hex_num(account.nonce))
}

fn eth_get_balance(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_getBalance", address = field::Empty, filter = field::Empty).entered();

    // parse params
    let (params, address) = next_rpc_param::<Address>(params.sequence())?;
    let (_, filter) = next_rpc_param_or_default::<BlockFilter>(params)?;

    // track
    Span::with(|s| {
        s.rec_str("address", &address);
        s.rec_str("filter", &filter);
    });
    tracing::info!(%address, %filter, "reading account native balance");

    // execute
    let point_in_time = ctx.storage.translate_to_point_in_time(filter)?;
    let account = ctx.storage.read_account(address, point_in_time)?;
    Ok(hex_num(account.balance))
}

fn eth_get_code(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_getCode", address = field::Empty, filter = field::Empty).entered();

    // parse params
    let (params, address) = next_rpc_param::<Address>(params.sequence())?;
    let (_, filter) = next_rpc_param_or_default::<BlockFilter>(params)?;

    // track
    Span::with(|s| {
        s.rec_str("address", &address);
        s.rec_str("filter", &filter);
    });

    // execute
    let point_in_time = ctx.storage.translate_to_point_in_time(filter)?;
    let account = ctx.storage.read_account(address, point_in_time)?;

    Ok(account.bytecode.map(hex_data).unwrap_or_else(hex_null))
}

// -----------------------------------------------------------------------------
// Subscriptions
// -----------------------------------------------------------------------------

async fn eth_subscribe(params: Params<'_>, pending: PendingSubscriptionSink, ctx: Arc<RpcContext>, ext: Extensions) -> impl IntoSubscriptionCloseResponse {
    // `middleware_enter` created to be used as a parent by `method_span`
    let middleware_enter = ext.enter_middleware_span();
    let method_span = info_span!("rpc::eth_subscribe", subscription = field::Empty);
    drop(middleware_enter);

    async move {
        // parse params
        let client = ext.rpc_client();
        let (params, event) = match next_rpc_param::<String>(params.sequence()) {
            Ok((params, event)) => (params, event),
            Err(e) => {
                pending.reject(StratusError::from(e)).await;
                return Ok(());
            }
        };

        // check subscription limits
        if let Err(e) = ctx.subs.check_client_subscriptions(ctx.rpc_server.rpc_max_subscriptions, client).await {
            pending.reject(e).await;
            return Ok(());
        }

        // track
        Span::with(|s| s.rec_str("subscription", &event));
        tracing::info!(%event, "subscribing to rpc event");

        // execute
        match event.deref() {
            "newPendingTransactions" => {
                ctx.subs.add_new_pending_txs_subscription(client, pending.accept().await?).await;
            }

            "newHeads" => {
                ctx.subs.add_new_heads_subscription(client, pending.accept().await?).await;
            }

            "logs" => {
                let (_, filter) = next_rpc_param_or_default::<LogFilterInput>(params)?;
                let filter = filter.parse(&ctx.storage)?;
                ctx.subs.add_logs_subscription(client, filter, pending.accept().await?).await;
            }

            // unsupported
            event => {
                pending
                    .reject(StratusError::from(RpcError::SubscriptionInvalid { event: event.to_string() }))
                    .await;
            }
        }

        Ok(())
    }
    .instrument(method_span)
    .await
}

// -----------------------------------------------------------------------------
// Storage
// -----------------------------------------------------------------------------

fn eth_get_storage_at(params: Params<'_>, ctx: Arc<RpcContext>, ext: Extensions) -> Result<String, StratusError> {
    // enter span
    let _middleware_enter = ext.enter_middleware_span();
    let _method_enter = info_span!("rpc::eth_getStorageAt", address = field::Empty, index = field::Empty).entered();
    Error Code System

    The new error code system with major codes has been implemented. Verify that all error codes are correctly assigned and that the new system doesn't introduce any conflicts or inconsistencies.

    use crate::ext::to_json_value;

pub trait ErrorCode {
    fn error_code(&self) -> i32;
    fn str_repr_from_err_code(code: i32) -> Option<&'static str>;
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 1000]
pub enum RpcError {
    #[error("Block filter does not point to a valid block.")]
    #[error_code = 1]
    BlockFilterInvalid { filter: BlockFilter },

    #[error("Denied because will fetch data from {actual} blocks, but the max allowed is {max}.")]
    #[error_code = 2]
    BlockRangeInvalid { actual: u64, max: u64 },

    #[error("Denied because client did not identify itself.")]
    #[error_code = 3]
    ClientMissing,

    #[error("Failed to decode {rust_type} parameter.")]
    #[error_code = 4]
    ParameterInvalid { rust_type: &'static str, decode_error: String },

    #[error("Expected {rust_type} parameter, but received nothing.")]
    #[error_code = 5]
    ParameterMissing { rust_type: &'static str },

    #[error("Invalid subscription event: {event}")]
    #[error_code = 6]
    SubscriptionInvalid { event: String },

    #[error("Denied because reached maximum subscription limit of {max}.")]
    #[error_code = 7]
    SubscriptionLimit { max: u32 },

    #[error("Failed to decode transaction RLP data.")]
    #[error_code = 8]
    TransactionInvalid { decode_error: String },

    #[error("Miner mode param is invalid.")]
    #[error_code = 9]
    MinerModeParamInvalid,
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 2000]
pub enum TransactionError {
    #[error("Account at {address} is not a contract.")]
    #[error_code = 1]
    AccountNotContract { address: Address },

    #[error("Transaction nonce {transaction} does not match account nonce {account}.")]
    #[error_code = 2]
    Nonce { transaction: Nonce, account: Nonce },

    #[error("Failed to executed transaction in EVM: {0:?}.")]
    #[error_code = 3]
    EvmFailed(String), // TODO: split this in multiple errors

    #[error("Failed to execute transaction in leader: {0:?}.")]
    #[error_code = 4]
    LeaderFailed(ErrorObjectOwned),

    #[error("Failed to forward transaction to leader node.")]
    #[error_code = 5]
    ForwardToLeaderFailed,

    #[error("Transaction reverted during execution.")]
    #[error_code = 6]
    Reverted { output: Bytes },

    #[error("Transaction from zero address is not allowed.")]
    #[error_code = 7]
    FromZeroAddress,
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 3000]
pub enum StorageError {
    #[error("Block conflict: {number} already exists in the permanent storage.")]
    #[error_code = 1]
    BlockConflict { number: BlockNumber },

    #[error("Mined number conflict between new block number ({new}) and mined block number ({mined}).")]
    #[error_code = 2]
    MinedNumberConflict { new: BlockNumber, mined: BlockNumber },

    #[error("Transaction execution conflicts: {0:?}.")]
    #[error_code = 3]
    TransactionConflict(Box<ExecutionConflicts>),

    #[error("Transaction input does not match block header")]
    #[error_code = 4]
    EvmInputMismatch { expected: Box<EvmInput>, actual: Box<EvmInput> },

    #[error("Pending number conflict between new block number ({new}) and pending block number ({pending}).")]
    #[error_code = 5]
    PendingNumberConflict { new: BlockNumber, pending: BlockNumber },

    #[error("There are ({pending_txs}) pending transactions.")]
    #[error_code = 6]
    PendingTransactionsExist { pending_txs: usize },

    #[error("Rocksdb returned an error: {err}")]
    #[error_code = 7]
    RocksError { err: anyhow::Error },

    #[error("Block not found using filter: {filter}")]
    #[error_code = 8]
    BlockNotFound { filter: BlockFilter },

    #[error("Unexpected storage error: {msg}")]
    #[error_code = 9]
    Unexpected { msg: String },
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 4000]
pub enum ImporterError {
    #[error("Importer is already running.")]
    #[error_code = 1]
    AlreadyRunning,

    #[error("Importer is already shutdown.")]
    #[error_code = 2]
    AlreadyShutdown,

    #[error("Failed to parse importer configuration.")]
    #[error_code = 3]
    ConfigParseError,

    #[error("Failed to initialize importer.")]
    #[error_code = 4]
    InitError,
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 5000]
pub enum ConsensusError {
    #[error("Consensus is temporarily unavailable for follower node.")]
    #[error_code = 1]
    Unavailable,

    #[error("Consensus is set.")]
    #[error_code = 2]
    Set,

    #[error("Failed to update consensus: Consensus is not set.")]
    #[error_code = 3]
    NotSet,
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 6000]
pub enum UnexpectedError {
    #[error("Unexpected channel {channel} closed.")]
    #[error_code = 1]
    ChannelClosed { channel: &'static str },

    #[error("Unexpected error: {0:?}.")]
    #[error_code = 2]
    Unexpected(anyhow::Error),
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr, ErrorCode)]
#[major_error_code = 7000]
pub enum StateError {
    #[error("Stratus is not ready to start servicing requests.")]
    #[error_code = 1]
    StratusNotReady,

    #[error("Stratus is shutting down.")]
    #[error_code = 2]
    StratusShutdown,

    #[error("Stratus node is not a follower.")]
    #[error_code = 3]
    StratusNotFollower,

    #[error("Incorrect password, cancelling operation.")]
    #[error_code = 4]
    InvalidPassword,

    #[error("Stratus node is already in the process of changing mode.")]
    #[error_code = 5]
    ModeChangeInProgress,

    #[error("Transaction processing is temporarily disabled.")]
    #[error_code = 6]
    TransactionsDisabled,

    #[error("Can't change miner mode while transactions are enabled.")]
    #[error_code = 7]
    TransactionsEnabled,
}

#[derive(Debug, thiserror::Error, strum::EnumProperty, strum::IntoStaticStr)]
pub enum StratusError {
    #[error(transparent)]
    RPC(#[from] RpcError),

    #[error(transparent)]
    Transaction(#[from] TransactionError),

    #[error(transparent)]
    Storage(#[from] StorageError),

    #[error(transparent)]
    Importer(#[from] ImporterError),

    #[error(transparent)]
    Consensus(#[from] ConsensusError),

    #[error(transparent)]
    Unexpected(#[from] UnexpectedError),

    #[error(transparent)]
    State(#[from] StateError),
}

impl ErrorCode for StratusError {
    fn error_code(&self) -> i32 {
        match self {
            Self::RPC(err) => err.error_code(),
            Self::Transaction(err) => err.error_code(),
            Self::Storage(err) => err.error_code(),
            Self::Importer(err) => err.error_code(),
            Self::Consensus(err) => err.error_code(),
            Self::Unexpected(err) => err.error_code(),
            Self::State(err) => err.error_code(),
        }
    }

    fn str_repr_from_err_code(code: i32) -> Option<&'static str> {
        let major = code % 1000;
        match major {
            1 => RpcError::str_repr_from_err_code(code),
            2 => TransactionError::str_repr_from_err_code(code),
            3 => StorageError::str_repr_from_err_code(code),
            4 => ImporterError::str_repr_from_err_code(code),
            5 => ConsensusError::str_repr_from_err_code(code),
            6 => UnexpectedError::str_repr_from_err_code(code),
            7 => StateError::str_repr_from_err_code(code),
            _ => None,
        }
    }
}
    Macro Changes

    The ErrorCode derive macro has been modified to include a major_error_code attribute. Ensure that this change is correctly implemented and doesn't break existing code that uses the ErrorCode derive macro.

    #[proc_macro_derive(ErrorCode, attributes(error_code, major_error_code))]
pub fn derive_error_code(input: TokenStream) -> TokenStream {
    let input = parse_macro_input!(input as DeriveInput);
    derive_error_code_impl(input).into()
}

fn derive_error_code_impl(input: DeriveInput) -> proc_macro2::TokenStream {
    let name = &input.ident;

    let Data::Enum(data_enum) = &input.data else {
        panic!("ErrorCode can only be derived for enums");
    };

    let mut match_arms = Vec::new();
    // Get the major error code if specified
    let major_error_code = input
        .attrs
        .iter()
        .find(|attr| attr.path().is_ident("major_error_code"))
        .and_then(|attr| {
            if let Meta::NameValue(meta) = &attr.meta {
                if let Expr::Lit(ExprLit { lit: Lit::Int(lit_int), .. }) = &meta.value {
                    lit_int.base10_parse::<i32>().ok()
                } else {
                    None
                }
            } else {
                None
            }
        })
        .unwrap_or(0);

    @github-actions GitHub Actions
    Copy link

    github-actions bot commented Jan 7, 2025

    PR Code Suggestions ✨

    Explore these optional code suggestions:

    CategorySuggestion                                                                                                                                    Score
    General
    Improve error handling for unknown error codes in RPC responses

    Implement error handling for the case when StratusError::str_repr_from_err_code
    returns None. Consider logging the unknown error code and providing a default error
    message.

    src/eth/rpc/rpc_middleware.rs [249-251]

     let rpc_result = match response_result.get("result") {
     Some(result) => if_else!(result.is_null(), metrics::LABEL_MISSING, metrics::LABEL_PRESENT),
-    None => StratusError::str_repr_from_err_code(error_code).unwrap_or("Unknown"),
+    None => StratusError::str_repr_from_err_code(error_code).unwrap_or_else(|| {
+        tracing::warn!("Unknown error code: {}", error_code);
+        "Unknown"
+    }),
 };
    Suggestion importance[1-10]: 7

    Why: The suggestion improves error handling by logging unknown error codes and providing a default message, enhancing debugging capabilities and system robustness.

    		7

    @carneiro-cw carneiro-cw enabled auto-merge (squash) January 7, 2025 21:25
    @carneiro-cw carneiro-cw merged commit 0c9bd1c into main Jan 7, 2025
    36 checks passed
    @carneiro-cw carneiro-cw deleted the remove_method_wrapper_2 branch January 7, 2025 21:26
    @gabriel-aranha-cw
    Copy link
    Contributor

    Final benchmark:
    Run ID: bench-411347226

    Git Info:

    Configuration:

    • Target Account Strategy: Default

    RPS Stats: Max: 1253.00, Min: 705.00, Avg: 1115.08, StdDev: 63.35
    TPS Stats: Max: 1241.00, Min: 858.00, Avg: 1081.85, StdDev: 67.38

    Plot: View Plot

    Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
    Reviewers
    No reviews
    Assignees
    No one assigned
    Labels
    None yet
    Projects
    None yet
    Milestone
    No milestone
    Development

    Successfully merging this pull request may close these issues.
    2 participants
    @carneiro-cw @gabriel-aranha-cw