arb_rpc/
nitro_execution_handler.rs

1//! Implementation of the `nitroexecution` RPC handler.
2//!
3//! Receives messages from the Nitro consensus layer, produces blocks,
4//! and maintains the mapping between message indices and block numbers.
5
6use std::sync::{Arc, OnceLock};
7
8use alloy_consensus::BlockHeader;
9use alloy_primitives::B256;
10use alloy_rpc_types_eth::BlockNumberOrTag;
11use base64::{
12    Engine as _, alphabet,
13    engine::{DecodePaddingMode, GeneralPurpose, GeneralPurposeConfig},
14};
15use jsonrpsee::core::RpcResult;
16use parking_lot::RwLock;
17use reth_metrics::{
18    Metrics,
19    metrics::{self, Gauge},
20};
21use reth_provider::{BlockNumReader, BlockReaderIdExt, HeaderProvider};
22use tracing::{debug, info, warn};
23
24use crate::{
25    block_producer::{BlockProducer, BlockProductionInput},
26    error::{RpcError, RpcResult as ArbRpcResult},
27    nitro_execution::{
28        NitroExecutionApiServer, RpcConsensusSyncData, RpcFinalityData, RpcMaintenanceStatus,
29        RpcMessageResult, RpcMessageWithMetadata, RpcMessageWithMetadataAndBlockInfo,
30    },
31};
32
33/// State shared between the RPC handler and the node.
34#[derive(Debug, Default)]
35pub struct NitroExecutionState {
36    /// Whether the node is synced with consensus.
37    pub synced: bool,
38    /// Maximum message count from consensus.
39    pub max_message_count: u64,
40}
41
42/// Prometheus metrics for the nitroexecution RPC handler.
43#[derive(Metrics)]
44#[metrics(scope = "arb_sync")]
45struct NitroExecutionMetrics {
46    /// Messages the consensus layer is ahead of the execution head.
47    messages_behind: Gauge,
48}
49
50/// Handler for the `nitroexecution` RPC namespace.
51///
52/// Receives L1 incoming messages from the consensus layer and produces blocks.
53/// Delegates actual block production to the `BlockProducer` implementation.
54pub struct NitroExecutionHandler<Provider, BP> {
55    provider: Provider,
56    block_producer: Arc<BP>,
57    state: Arc<RwLock<NitroExecutionState>>,
58    /// Genesis block number (0 for Arbitrum Sepolia, 22207817 for Arbitrum One).
59    genesis_block_num: u64,
60    metrics: NitroExecutionMetrics,
61}
62
63impl<Provider, BP> NitroExecutionHandler<Provider, BP> {
64    /// Create a new handler with a block producer.
65    pub fn new(provider: Provider, block_producer: Arc<BP>, genesis_block_num: u64) -> Self {
66        Self {
67            provider,
68            block_producer,
69            state: Arc::new(RwLock::new(NitroExecutionState::default())),
70            genesis_block_num,
71            metrics: NitroExecutionMetrics::default(),
72        }
73    }
74
75    /// Convert a message index to a block number.
76    fn message_index_to_block_number(&self, msg_idx: u64) -> u64 {
77        self.genesis_block_num + msg_idx
78    }
79
80    /// Convert a block number to a message index.
81    fn block_number_to_message_index(&self, block_num: u64) -> Option<u64> {
82        if block_num < self.genesis_block_num {
83            return None;
84        }
85        Some(block_num - self.genesis_block_num)
86    }
87}
88
89impl<Provider, BP> NitroExecutionHandler<Provider, BP>
90where
91    Provider: BlockReaderIdExt + HeaderProvider,
92{
93    /// Look up a sealed header by block number.
94    fn get_header(
95        &self,
96        block_num: u64,
97    ) -> ArbRpcResult<
98        Option<reth_primitives_traits::SealedHeader<<Provider as HeaderProvider>::Header>>,
99    > {
100        Ok(self
101            .provider
102            .sealed_header_by_number_or_tag(BlockNumberOrTag::Number(block_num))?)
103    }
104
105    /// Extract send root from a header's extra_data.
106    fn send_root_from_header(header: &impl BlockHeader) -> B256 {
107        let extra = header.extra_data();
108        if extra.len() >= 32 {
109            B256::from_slice(&extra[..32])
110        } else {
111            B256::ZERO
112        }
113    }
114}
115
116impl<Provider, BP> NitroExecutionHandler<Provider, BP>
117where
118    Provider: BlockNumReader,
119{
120    /// Refresh the `messages_behind` gauge from the current head and target.
121    fn update_messages_behind(&self) {
122        let best = match self.provider.best_block_number() {
123            Ok(best) => best,
124            Err(err) => {
125                warn!(target: "nitroexecution", %err, "failed to read best block number; skipping messages_behind update");
126                return;
127            }
128        };
129        let Some(head_msg_idx) = self.block_number_to_message_index(best) else {
130            debug!(target: "nitroexecution", best, genesis = self.genesis_block_num, "head below genesis; skipping messages_behind update");
131            return;
132        };
133        let max_message_count = self.state.read().max_message_count;
134        // Processed = head_msg_idx + 1, so behind = count - (idx + 1).
135        let processed = head_msg_idx.saturating_add(1);
136        self.metrics
137            .messages_behind
138            .set(max_message_count.saturating_sub(processed) as f64);
139    }
140}
141
142/// Decode the l2_msg field from the RPC message.
143///
144/// JSON encoding always base64-encodes byte fields. The base64 output
145/// can start with "0x" as valid base64 characters, so always decode as base64.
146fn decode_l2_msg(l2_msg: &Option<String>) -> ArbRpcResult<Vec<u8>> {
147    match l2_msg {
148        Some(s) if !s.is_empty() => base64_decode(s),
149        _ => Ok(vec![]),
150    }
151}
152
153const STANDARD_ALPHABET: &[u8; 64] =
154    b"ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/";
155
156fn base64_engine() -> &'static GeneralPurpose {
157    static ENGINE: OnceLock<GeneralPurpose> = OnceLock::new();
158    ENGINE.get_or_init(|| {
159        let cfg = GeneralPurposeConfig::new()
160            .with_decode_padding_mode(DecodePaddingMode::Indifferent)
161            .with_decode_allow_trailing_bits(true);
162        GeneralPurpose::new(&alphabet::STANDARD, cfg)
163    })
164}
165
166fn base64_decode(input: &str) -> ArbRpcResult<Vec<u8>> {
167    let stripped = input.trim_end_matches('=');
168    // A length-1 mod 4 tail carries no meaningful bits; the orphan symbol
169    // is still alphabet-validated, then discarded.
170    let body_len = stripped.len() & !3;
171    let tail = &stripped.as_bytes()[body_len..];
172    let body = if tail.len() == 1 {
173        let b = tail[0];
174        if !STANDARD_ALPHABET.contains(&b) {
175            return Err(RpcError::base64_decode(format!(
176                "invalid base64 character: {}",
177                b as char
178            )));
179        }
180        &stripped[..body_len]
181    } else {
182        stripped
183    };
184    base64_engine()
185        .decode(body)
186        .map_err(|e| RpcError::base64_decode(format!("invalid base64: {e}")))
187}
188
189#[async_trait::async_trait]
190impl<Provider, BP> NitroExecutionApiServer for NitroExecutionHandler<Provider, BP>
191where
192    Provider: BlockNumReader + BlockReaderIdExt + HeaderProvider + 'static,
193    BP: BlockProducer,
194{
195    async fn digest_message(
196        &self,
197        msg_idx: u64,
198        message: RpcMessageWithMetadata,
199        _message_for_prefetch: Option<RpcMessageWithMetadata>,
200    ) -> RpcResult<RpcMessageResult> {
201        let block_num = self.message_index_to_block_number(msg_idx);
202        let kind = message.message.header.kind;
203        info!(target: "nitroexecution", msg_idx, block_num, kind, "digestMessage called");
204
205        // Handle init message (Kind=11) — cache params, return genesis block.
206        // The Init message does NOT produce a block. Its params are applied
207        // during the first real block's execution.
208        if kind == 11 {
209            let l2_msg = decode_l2_msg(&message.message.l2_msg)?;
210            self.block_producer
211                .cache_init_message(&l2_msg)
212                .map_err(RpcError::from)?;
213
214            let genesis_header = self
215                .get_header(self.genesis_block_num)?
216                .ok_or_else(|| RpcError::not_found("Genesis block not found for Init message"))?;
217            let send_root = Self::send_root_from_header(genesis_header.header());
218            info!(target: "nitroexecution", "Init message cached, returning genesis block");
219            return Ok(RpcMessageResult {
220                block_hash: genesis_header.hash(),
221                send_root,
222            });
223        }
224
225        // Check if we already have this block (idempotent).
226        if let Some(header) = self.get_header(block_num)? {
227            let send_root = Self::send_root_from_header(header.header());
228            debug!(target: "nitroexecution", block_num, ?send_root, "Block already exists");
229            return Ok(RpcMessageResult {
230                block_hash: header.hash(),
231                send_root,
232            });
233        }
234
235        let l2_msg = decode_l2_msg(&message.message.l2_msg)?;
236
237        // Build batch data stats if present
238        let batch_data_stats = message
239            .message
240            .batch_data_tokens
241            .as_ref()
242            .map(|s| (s.length, s.nonzeros));
243
244        // Build the block production input
245        let input = BlockProductionInput {
246            kind,
247            sender: message.message.header.sender,
248            l1_block_number: message.message.header.block_number,
249            l1_timestamp: message.message.header.timestamp,
250            request_id: message.message.header.request_id,
251            l1_base_fee: message.message.header.base_fee_l1,
252            l2_msg,
253            delayed_messages_read: message.delayed_messages_read,
254            batch_gas_cost: message.message.batch_gas_cost,
255            batch_data_stats,
256        };
257
258        let result = self
259            .block_producer
260            .produce_block(msg_idx, input)
261            .await
262            .map_err(RpcError::from)?;
263
264        self.update_messages_behind();
265
266        Ok(RpcMessageResult {
267            block_hash: result.block_hash,
268            send_root: result.send_root,
269        })
270    }
271
272    async fn reorg(
273        &self,
274        msg_idx_of_first_msg_to_add: u64,
275        new_messages: Vec<RpcMessageWithMetadataAndBlockInfo>,
276        _old_messages: Vec<RpcMessageWithMetadata>,
277    ) -> RpcResult<Vec<RpcMessageResult>> {
278        info!(
279            target: "nitroexecution",
280            msg_idx_of_first_msg_to_add,
281            new_msgs = new_messages.len(),
282            "reorg"
283        );
284
285        // Roll back to the last kept block = (first divergent msg) - 1.
286        // Message i corresponds to block (genesis + i), so reset head to
287        // genesis + (msg_idx_of_first_msg_to_add - 1). If the target is
288        // before genesis, reset to genesis.
289        let target_block = msg_idx_of_first_msg_to_add
290            .saturating_sub(1)
291            .saturating_add(self.genesis_block_num);
292
293        self.block_producer
294            .reset_to_block(target_block)
295            .await
296            .map_err(RpcError::from)?;
297
298        // Replay new messages on top of the rolled-back head.
299        let mut results = Vec::with_capacity(new_messages.len());
300        for (i, wrapped) in new_messages.into_iter().enumerate() {
301            let msg_idx = msg_idx_of_first_msg_to_add + i as u64;
302            let meta = wrapped.message;
303            let l2_msg = decode_l2_msg(&meta.message.l2_msg)?;
304            let batch_data_stats = meta
305                .message
306                .batch_data_tokens
307                .as_ref()
308                .map(|s| (s.length, s.nonzeros));
309            let input = BlockProductionInput {
310                kind: meta.message.header.kind,
311                sender: meta.message.header.sender,
312                l1_block_number: meta.message.header.block_number,
313                l1_timestamp: meta.message.header.timestamp,
314                request_id: meta.message.header.request_id,
315                l1_base_fee: meta.message.header.base_fee_l1,
316                l2_msg,
317                delayed_messages_read: meta.delayed_messages_read,
318                batch_gas_cost: meta.message.batch_gas_cost,
319                batch_data_stats,
320            };
321            let produced = self
322                .block_producer
323                .produce_block(msg_idx, input)
324                .await
325                .map_err(RpcError::from)?;
326            results.push(RpcMessageResult {
327                block_hash: produced.block_hash,
328                send_root: produced.send_root,
329            });
330        }
331        Ok(results)
332    }
333
334    async fn head_message_index(&self) -> RpcResult<u64> {
335        let best = self.provider.best_block_number().map_err(RpcError::from)?;
336
337        let msg_idx = self.block_number_to_message_index(best).unwrap_or(0);
338        debug!(target: "nitroexecution", best, msg_idx, "headMessageIndex");
339        Ok(msg_idx)
340    }
341
342    async fn result_at_message_index(&self, msg_idx: u64) -> RpcResult<RpcMessageResult> {
343        let block_num = self.message_index_to_block_number(msg_idx);
344
345        let header = self
346            .get_header(block_num)?
347            .ok_or_else(|| RpcError::not_found(format!("Block {block_num} not found")))?;
348
349        let send_root = Self::send_root_from_header(header.header());
350
351        Ok(RpcMessageResult {
352            block_hash: header.hash(),
353            send_root,
354        })
355    }
356
357    fn set_finality_data(
358        &self,
359        safe: Option<RpcFinalityData>,
360        finalized: Option<RpcFinalityData>,
361        validated: Option<RpcFinalityData>,
362    ) -> RpcResult<()> {
363        debug!(target: "nitroexecution", ?safe, ?finalized, ?validated, "setFinalityData");
364        self.block_producer
365            .set_finality(
366                safe.map(|f| f.block_hash),
367                finalized.map(|f| f.block_hash),
368                validated.map(|f| f.block_hash),
369            )
370            .map_err(RpcError::from)?;
371        Ok(())
372    }
373
374    fn set_consensus_sync_data(&self, sync_data: RpcConsensusSyncData) -> RpcResult<()> {
375        {
376            let mut state = self.state.write();
377            state.synced = sync_data.synced;
378            state.max_message_count = sync_data.max_message_count;
379        }
380        debug!(target: "nitroexecution", synced = sync_data.synced, max = sync_data.max_message_count, "setConsensusSyncData");
381        self.update_messages_behind();
382        Ok(())
383    }
384
385    fn mark_feed_start(&self, to: u64) -> RpcResult<()> {
386        debug!(target: "nitroexecution", to, "markFeedStart");
387        Ok(())
388    }
389
390    async fn trigger_maintenance(&self) -> RpcResult<()> {
391        Ok(())
392    }
393
394    async fn should_trigger_maintenance(&self) -> RpcResult<bool> {
395        Ok(false)
396    }
397
398    async fn maintenance_status(&self) -> RpcResult<RpcMaintenanceStatus> {
399        Ok(RpcMaintenanceStatus { is_running: false })
400    }
401
402    async fn arbos_version_for_message_index(&self, msg_idx: u64) -> RpcResult<u64> {
403        let block_num = self.message_index_to_block_number(msg_idx);
404
405        let header = self
406            .get_header(block_num)?
407            .ok_or_else(|| RpcError::not_found(format!("Block {block_num} not found")))?;
408
409        let mix = header.header().mix_hash().unwrap_or_default();
410        let arbos_version = u64::from_be_bytes(mix.0[16..24].try_into().unwrap_or_default());
411
412        Ok(arbos_version)
413    }
414}
415
416#[cfg(test)]
417mod tests {
418    use base64::engine::general_purpose::STANDARD as B64;
419
420    use super::*;
421
422    #[test]
423    fn decode_empty_option_is_ok() {
424        assert_eq!(decode_l2_msg(&None).unwrap(), Vec::<u8>::new());
425        assert_eq!(
426            decode_l2_msg(&Some(String::new())).unwrap(),
427            Vec::<u8>::new()
428        );
429    }
430
431    #[test]
432    fn decode_standard_padded() {
433        let encoded = B64.encode(b"Hello, world!");
434        let out = base64_decode(&encoded).unwrap();
435        assert_eq!(out, b"Hello, world!");
436    }
437
438    #[test]
439    fn decode_accepts_unpadded() {
440        let encoded = B64.encode(b"Hello");
441        let stripped = encoded.trim_end_matches('=').to_string();
442        assert_eq!(base64_decode(&stripped).unwrap(), b"Hello");
443    }
444
445    #[test]
446    fn decode_accepts_extra_padding() {
447        assert_eq!(base64_decode("SGVsbG8==").unwrap(), b"Hello");
448        assert_eq!(base64_decode("SGVsbG8====").unwrap(), b"Hello");
449    }
450
451    #[test]
452    fn decode_rejects_invalid_character() {
453        assert!(base64_decode("SG!X").is_err());
454        assert!(base64_decode("a b").is_err());
455        assert!(base64_decode("hello world").is_err());
456    }
457
458    #[test]
459    fn decode_rejects_padding_in_body() {
460        assert!(base64_decode("=SGVs").is_err());
461        assert!(base64_decode("SGVs=bG8").is_err());
462    }
463
464    #[test]
465    fn decode_large_payload_matches_roundtrip() {
466        let bytes: Vec<u8> = (0..32 * 1024).map(|i| (i * 7 + 3) as u8).collect();
467        let encoded = B64.encode(&bytes);
468        assert_eq!(base64_decode(&encoded).unwrap(), bytes);
469    }
470
471    #[test]
472    fn decode_preserves_lenient_padding_tail() {
473        assert_eq!(base64_decode("S").unwrap(), Vec::<u8>::new());
474        assert_eq!(base64_decode("SG").unwrap(), vec![b'H']);
475        assert_eq!(base64_decode("SGV").unwrap(), vec![b'H', b'e']);
476    }
477
478    #[test]
479    fn decode_length_one_tail_validates_alphabet() {
480        assert!(base64_decode("ABCD!").is_err());
481    }
482
483    #[test]
484    fn decode_all_alphabet_characters() {
485        let out = base64_decode("ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/")
486            .unwrap();
487        assert_eq!(out.len(), 48);
488    }
489}
arb_rpc/nitro_execution_handler.rs

arb_rpc/
nitro_execution_handler.rs
