arb_rpc/

arbdebug.rs

//! `arbdebug_*` namespace — historical pricing + retryable queue
//! introspection. Samples ArbOS state at each block in the requested
//! range.

use std::sync::Arc;

use alloy_consensus::BlockHeader;
use alloy_primitives::{Address, B256, StorageKey, U256};
use alloy_rpc_types_eth::BlockNumberOrTag;
use arb_storage::{
    ARBOS_STATE_ADDRESS,
    layout::{
        L1_PRICING_SUBSPACE, L2_PRICING_SUBSPACE, RETRYABLES_SUBSPACE, ROOT_STORAGE_KEY,
        derive_subspace_key, map_slot, subspace_slot,
    },
};
use arbos::{
    l1_pricing::{
        AMORTIZED_COST_CAP_BIPS_OFFSET as L1_AMORTIZED_COST_CAP_BIPS_OFFSET,
        EQUILIBRATION_UNITS_OFFSET as L1_EQUILIBRATION_UNITS_OFFSET,
        FUNDS_DUE_FOR_REWARDS_OFFSET as L1_FUNDS_DUE_FOR_REWARDS_OFFSET,
        INERTIA_OFFSET as L1_INERTIA_OFFSET,
        L1_FEES_AVAILABLE_OFFSET as L1_L1_FEES_AVAILABLE_OFFSET,
        LAST_SURPLUS_OFFSET as L1_LAST_SURPLUS_OFFSET,
        LAST_UPDATE_TIME_OFFSET as L1_LAST_UPDATE_TIME_OFFSET,
        PAY_REWARDS_TO_OFFSET as L1_PAY_REWARDS_TO_OFFSET,
        PER_BATCH_GAS_COST_OFFSET as L1_PER_BATCH_GAS_COST_OFFSET,
        PER_UNIT_REWARD_OFFSET as L1_PER_UNIT_REWARD_OFFSET,
        PRICE_PER_UNIT_OFFSET as L1_PRICE_PER_UNIT_OFFSET,
        UNITS_SINCE_OFFSET as L1_UNITS_SINCE_UPDATE_OFFSET,
    },
    l2_pricing::{
        BACKLOG_TOLERANCE_OFFSET as L2_BACKLOG_TOLERANCE_OFFSET,
        BASE_FEE_WEI_OFFSET as L2_BASE_FEE_OFFSET, GAS_BACKLOG_OFFSET as L2_GAS_BACKLOG_OFFSET,
        MIN_BASE_FEE_WEI_OFFSET as L2_MIN_BASE_FEE_OFFSET,
        PER_BLOCK_GAS_LIMIT_OFFSET as L2_PER_BLOCK_GAS_LIMIT_OFFSET,
        PRICING_INERTIA_OFFSET as L2_PRICING_INERTIA_OFFSET,
        SPEED_LIMIT_PER_SECOND_OFFSET as L2_SPEED_LIMIT_OFFSET,
    },
    retryables::{TIMEOUT_OFFSET, TIMEOUT_QUEUE_KEY},
};
use jsonrpsee::{
    core::RpcResult,
    proc_macros::rpc,
    types::{ErrorObject, error::INTERNAL_ERROR_CODE},
};
use reth_provider::{BlockReaderIdExt, ReceiptProvider, StateProviderFactory};
use serde::{Deserialize, Serialize};

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PricingModelHistory {
    pub start: u64,
    pub end: u64,
    pub step: u64,
    pub timestamp: Vec<u64>,
    pub base_fee: Vec<U256>,
    pub gas_backlog: Vec<u64>,
    pub gas_used: Vec<u64>,
    pub min_base_fee: U256,
    pub speed_limit: u64,
    pub per_block_gas_limit: u64,
    pub per_tx_gas_limit: u64,
    pub pricing_inertia: u64,
    pub backlog_tolerance: u64,
    pub l1_base_fee_estimate: Vec<U256>,
    pub l1_last_surplus: Vec<U256>,
    pub l1_funds_due: Vec<U256>,
    pub l1_funds_due_for_rewards: Vec<U256>,
    pub l1_units_since_update: Vec<u64>,
    pub l1_last_update_time: Vec<u64>,
    pub l1_equilibration_units: U256,
    pub l1_per_batch_cost: i64,
    pub l1_amortized_cost_cap_bips: u64,
    pub l1_pricing_inertia: u64,
    pub l1_per_unit_reward: u64,
    pub l1_pay_reward_to: Address,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TimeoutQueueHistory {
    pub start: u64,
    pub end: u64,
    pub step: u64,
    pub timestamp: Vec<u64>,
    pub size: Vec<u64>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TimeoutQueue {
    pub block_number: u64,
    pub tickets: Vec<B256>,
    pub timeouts: Vec<u64>,
}

#[rpc(server, namespace = "arbdebug")]
pub trait ArbDebugApi {
    #[method(name = "pricingModel")]
    async fn pricing_model(&self, start: u64, end: u64) -> RpcResult<PricingModelHistory>;

    #[method(name = "timeoutQueueHistory")]
    async fn timeout_queue_history(&self, start: u64, end: u64) -> RpcResult<TimeoutQueueHistory>;

    #[method(name = "timeoutQueue")]
    async fn timeout_queue(&self, block_num: u64) -> RpcResult<TimeoutQueue>;
}

#[derive(Debug, Clone)]
pub struct ArbDebugConfig {
    /// Max samples per query. Zero disables arbdebug.
    pub block_range_bound: u64,
    /// Max tickets returned from `timeoutQueue`.
    pub timeout_queue_bound: u64,
}

impl Default for ArbDebugConfig {
    fn default() -> Self {
        Self {
            block_range_bound: 256,
            timeout_queue_bound: 256,
        }
    }
}

pub struct ArbDebugHandler<Provider> {
    provider: Provider,
    config: Arc<ArbDebugConfig>,
}

impl<Provider: std::fmt::Debug> std::fmt::Debug for ArbDebugHandler<Provider> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("ArbDebugHandler")
            .field("config", &self.config)
            .finish_non_exhaustive()
    }
}

impl<Provider: Clone> Clone for ArbDebugHandler<Provider> {
    fn clone(&self) -> Self {
        Self {
            provider: self.provider.clone(),
            config: self.config.clone(),
        }
    }
}

impl<Provider> ArbDebugHandler<Provider> {
    pub fn new(provider: Provider, config: ArbDebugConfig) -> Self {
        Self {
            provider,
            config: Arc::new(config),
        }
    }
}

fn internal_err(msg: impl std::fmt::Display) -> ErrorObject<'static> {
    ErrorObject::owned(INTERNAL_ERROR_CODE, msg.to_string(), None::<()>)
}

/// Sample step-size: if the range exceeds `bound` blocks, step > 1 so the
/// total number of samples stays ≤ bound.
fn compute_step(start: u64, end: u64, bound: u64) -> (u64, u64, u64) {
    let span = end.saturating_sub(start).saturating_add(1);
    if span == 0 || bound == 0 {
        return (start, 1, 0);
    }
    let step = if span > bound {
        span.div_ceil(bound)
    } else {
        1
    };
    let samples = span.div_ceil(step).min(bound);
    let first = end.saturating_sub(step.saturating_mul(samples.saturating_sub(1)));
    (first, step, samples)
}

impl<Provider> ArbDebugHandler<Provider>
where
    Provider: StateProviderFactory + BlockReaderIdExt + ReceiptProvider + Clone + 'static,
{
    /// Total gas consumed in the given block, summed from the last
    /// receipt's `cumulative_gas_used`.
    fn block_gas_used(&self, block: u64) -> Result<u64, ErrorObject<'static>> {
        use alloy_consensus::TxReceipt;
        let receipts = self
            .provider
            .receipts_by_block(alloy_eips::BlockHashOrNumber::Number(block))
            .map_err(internal_err)?
            .unwrap_or_default();
        Ok(receipts
            .last()
            .map(|r| r.cumulative_gas_used())
            .unwrap_or(0))
    }

    fn check_enabled(&self) -> Result<(), ErrorObject<'static>> {
        if self.config.block_range_bound == 0 {
            return Err(internal_err("arbdebug disabled (block_range_bound = 0)"));
        }
        Ok(())
    }

    fn validate_range(&self, start: u64, end: u64) -> Result<(), ErrorObject<'static>> {
        if start > end {
            return Err(internal_err(format!(
                "invalid range: start {start} > end {end}"
            )));
        }
        Ok(())
    }

    fn header_timestamp(&self, block: u64) -> Result<u64, ErrorObject<'static>> {
        let header = self
            .provider
            .sealed_header_by_number_or_tag(BlockNumberOrTag::Number(block))
            .map_err(internal_err)?
            .ok_or_else(|| internal_err(format!("block {block} not found")))?;
        Ok(header.timestamp())
    }

    fn read_slot(&self, block: u64, slot: U256) -> Result<U256, ErrorObject<'static>> {
        let state = self
            .provider
            .state_by_block_id(BlockNumberOrTag::Number(block).into())
            .map_err(internal_err)?;
        let k = StorageKey::from(B256::from(slot.to_be_bytes::<32>()));
        Ok(state
            .storage(ARBOS_STATE_ADDRESS, k)
            .map_err(internal_err)?
            .unwrap_or(U256::ZERO))
    }

    fn read_l1_field(&self, block: u64, offset: u64) -> Result<U256, ErrorObject<'static>> {
        self.read_slot(block, subspace_slot(L1_PRICING_SUBSPACE, offset))
    }

    fn read_l2_field(&self, block: u64, offset: u64) -> Result<U256, ErrorObject<'static>> {
        self.read_slot(block, subspace_slot(L2_PRICING_SUBSPACE, offset))
    }

    /// Storage key for the retryable timeout queue's body.
    fn retryable_queue_storage_key() -> B256 {
        let retryables = derive_subspace_key(ROOT_STORAGE_KEY, RETRYABLES_SUBSPACE);
        derive_subspace_key(retryables.as_slice(), TIMEOUT_QUEUE_KEY)
    }

    /// Size of the timeout queue at the given block.
    fn queue_size_at(&self, block: u64) -> Result<u64, ErrorObject<'static>> {
        // Queue layout: the queue's own storage has offset-0 = next_put,
        // offset-1 = next_get. `size = next_put - next_get`.
        let qk = Self::retryable_queue_storage_key();
        let put = self
            .read_slot(block, map_slot(qk.as_slice(), 0))?
            .try_into()
            .unwrap_or(0u64);
        let get = self
            .read_slot(block, map_slot(qk.as_slice(), 1))?
            .try_into()
            .unwrap_or(0u64);
        Ok(put.saturating_sub(get))
    }

    /// Enumerate `(ticket_id, timeout)` for every pending retryable
    /// in the timeout queue at the given block, up to `max_entries`.
    /// Reads are against the historic state for `block` via the
    /// StateProvider — no ArbosState instantiation required.
    fn queue_snapshot_at(
        &self,
        block: u64,
        max_entries: usize,
    ) -> Result<Vec<(B256, u64)>, ErrorObject<'static>> {
        let qk = Self::retryable_queue_storage_key();
        let put: u64 = self
            .read_slot(block, map_slot(qk.as_slice(), 0))?
            .try_into()
            .unwrap_or(0);
        let get: u64 = self
            .read_slot(block, map_slot(qk.as_slice(), 1))?
            .try_into()
            .unwrap_or(0);
        let retryables_key = derive_subspace_key(ROOT_STORAGE_KEY, RETRYABLES_SUBSPACE);
        let mut out = Vec::new();
        for idx in get..put {
            if out.len() >= max_entries {
                break;
            }
            let ticket_slot = map_slot(qk.as_slice(), idx);
            let ticket_word = self.read_slot(block, ticket_slot)?;
            let id = B256::from(ticket_word.to_be_bytes::<32>());
            if id == B256::ZERO {
                continue;
            }
            // Per-retryable storage is keyed by ticket_id as a subspace
            // of the retryables subspace. Read the timeout field.
            let ret_key = derive_subspace_key(retryables_key.as_slice(), id.as_slice());
            let timeout: u64 = self
                .read_slot(block, map_slot(ret_key.as_slice(), TIMEOUT_OFFSET))?
                .try_into()
                .unwrap_or(0);
            if timeout == 0 {
                continue;
            }
            out.push((id, timeout));
        }
        Ok(out)
    }
}

#[async_trait::async_trait]
impl<Provider> ArbDebugApiServer for ArbDebugHandler<Provider>
where
    Provider:
        StateProviderFactory + BlockReaderIdExt + ReceiptProvider + Clone + Send + Sync + 'static,
{
    async fn pricing_model(&self, start: u64, end: u64) -> RpcResult<PricingModelHistory> {
        self.check_enabled()?;
        self.validate_range(start, end)?;
        let (first, step, samples) = compute_step(start, end, self.config.block_range_bound);

        let mut timestamp = Vec::with_capacity(samples as usize);
        let mut base_fee = Vec::with_capacity(samples as usize);
        let mut gas_backlog = Vec::with_capacity(samples as usize);
        let mut gas_used = Vec::with_capacity(samples as usize);
        let mut l1_base_fee_estimate = Vec::with_capacity(samples as usize);
        let mut l1_last_surplus = Vec::with_capacity(samples as usize);
        let mut l1_funds_due = Vec::with_capacity(samples as usize);
        let mut l1_funds_due_for_rewards = Vec::with_capacity(samples as usize);
        let mut l1_units_since_update = Vec::with_capacity(samples as usize);
        let mut l1_last_update_time = Vec::with_capacity(samples as usize);

        for i in 0..samples {
            let b = first + step * i;
            timestamp.push(self.header_timestamp(b)?);
            base_fee.push(self.read_l2_field(b, L2_BASE_FEE_OFFSET)?);
            gas_backlog.push(
                self.read_l2_field(b, L2_GAS_BACKLOG_OFFSET)?
                    .try_into()
                    .unwrap_or(0u64),
            );
            gas_used.push(self.block_gas_used(b)?);
            l1_base_fee_estimate.push(self.read_l1_field(b, L1_PRICE_PER_UNIT_OFFSET)?);
            l1_last_surplus.push(self.read_l1_field(b, L1_LAST_SURPLUS_OFFSET)?);
            l1_funds_due.push(self.read_l1_field(b, L1_L1_FEES_AVAILABLE_OFFSET)?);
            l1_funds_due_for_rewards.push(self.read_l1_field(b, L1_FUNDS_DUE_FOR_REWARDS_OFFSET)?);
            l1_units_since_update.push(
                self.read_l1_field(b, L1_UNITS_SINCE_UPDATE_OFFSET)?
                    .try_into()
                    .unwrap_or(0u64),
            );
            l1_last_update_time.push(
                self.read_l1_field(b, L1_LAST_UPDATE_TIME_OFFSET)?
                    .try_into()
                    .unwrap_or(0u64),
            );
        }

        // Scalar fields — read once at `end`.
        let min_base_fee = self.read_l2_field(end, L2_MIN_BASE_FEE_OFFSET)?;
        let speed_limit = self
            .read_l2_field(end, L2_SPEED_LIMIT_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let per_block_gas_limit = self
            .read_l2_field(end, L2_PER_BLOCK_GAS_LIMIT_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let pricing_inertia = self
            .read_l2_field(end, L2_PRICING_INERTIA_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let backlog_tolerance = self
            .read_l2_field(end, L2_BACKLOG_TOLERANCE_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let l1_equilibration_units = self.read_l1_field(end, L1_EQUILIBRATION_UNITS_OFFSET)?;
        let l1_per_batch_cost: i64 = self
            .read_l1_field(end, L1_PER_BATCH_GAS_COST_OFFSET)?
            .try_into()
            .unwrap_or(0i64);
        let l1_amortized_cost_cap_bips = self
            .read_l1_field(end, L1_AMORTIZED_COST_CAP_BIPS_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let l1_pricing_inertia = self
            .read_l1_field(end, L1_INERTIA_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let l1_per_unit_reward = self
            .read_l1_field(end, L1_PER_UNIT_REWARD_OFFSET)?
            .try_into()
            .unwrap_or(0u64);
        let l1_pay_reward_to = {
            let word = self.read_l1_field(end, L1_PAY_REWARDS_TO_OFFSET)?;
            Address::from_slice(&word.to_be_bytes::<32>()[12..])
        };

        Ok(PricingModelHistory {
            start,
            end,
            step,
            timestamp,
            base_fee,
            gas_backlog,
            gas_used,
            min_base_fee,
            speed_limit,
            per_block_gas_limit,
            per_tx_gas_limit: 0,
            pricing_inertia,
            backlog_tolerance,
            l1_base_fee_estimate,
            l1_last_surplus,
            l1_funds_due,
            l1_funds_due_for_rewards,
            l1_units_since_update,
            l1_last_update_time,
            l1_equilibration_units,
            l1_per_batch_cost,
            l1_amortized_cost_cap_bips,
            l1_pricing_inertia,
            l1_per_unit_reward,
            l1_pay_reward_to,
        })
    }

    async fn timeout_queue_history(&self, start: u64, end: u64) -> RpcResult<TimeoutQueueHistory> {
        self.check_enabled()?;
        self.validate_range(start, end)?;
        let (first, step, samples) = compute_step(start, end, self.config.block_range_bound);

        let mut timestamp = Vec::with_capacity(samples as usize);
        let mut size = Vec::with_capacity(samples as usize);
        for i in 0..samples {
            let b = first + step * i;
            timestamp.push(self.header_timestamp(b)?);
            size.push(self.queue_size_at(b)?);
        }
        Ok(TimeoutQueueHistory {
            start,
            end,
            step,
            timestamp,
            size,
        })
    }

    async fn timeout_queue(&self, block_num: u64) -> RpcResult<TimeoutQueue> {
        self.check_enabled()?;
        let entries =
            self.queue_snapshot_at(block_num, self.config.timeout_queue_bound as usize)?;
        let (tickets, timeouts): (Vec<B256>, Vec<u64>) = entries.into_iter().unzip();
        Ok(TimeoutQueue {
            block_number: block_num,
            tickets,
            timeouts,
        })
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn compute_step_span_fits_bound() {
        let (first, step, samples) = compute_step(100, 109, 256);
        assert_eq!(first, 100);
        assert_eq!(step, 1);
        assert_eq!(samples, 10);
    }

    #[test]
    fn compute_step_span_exceeds_bound() {
        let (first, step, samples) = compute_step(0, 9999, 100);
        assert!(samples <= 100);
        assert!(step >= 100);
        // Should anchor last sample at `end`.
        assert_eq!(first + step * (samples - 1), 9999);
    }

    #[test]
    fn compute_step_single_block() {
        let (first, step, samples) = compute_step(42, 42, 256);
        assert_eq!(first, 42);
        assert_eq!(step, 1);
        assert_eq!(samples, 1);
    }

    #[test]
    fn compute_step_zero_bound() {
        let (_, step, samples) = compute_step(0, 10, 0);
        assert_eq!(step, 1);
        assert_eq!(samples, 0);
    }
}