fix: audit round 3 (#284)

* fix(GRETH-037): replace type-erased event bus with typed OnceLock

- Replace OnceCell<Box<dyn Any>> with OnceLock<PipeExecLayerEventBus<EthPrimitives>>
- Remove generic parameter from get_pipe_exec_layer_event_bus()
- Add 120s timeout with panic on expiry (GRETH-038 combined fix)
- Add TypeId assertion in tree/mod.rs pipe_run_inner as runtime safety guard
- Use transmute to bridge Receiver<PipeExecLayerEvent<EthPrimitives>> to generic N
- Update all call sites to remove turbofish generic parameter

* Remove the duplicate `new_system_call_txn()` in `metadata_txn.rs`

* add design-intent comments for epoch/execute_height atomic ordering

Author: nekomoto911

Cargo.lock

@@ -558,8 +558,21 @@ where
     }
 
     fn pipe_run_inner(mut self) {
+        // Safety guard: assert N == EthPrimitives at runtime to prevent silent UB
+        // from the transmute below. This is feasible because NodePrimitives has a
+        // 'static bound, and keeps the reth upstream generic signature chain intact.
+        assert_eq!(
+            std::any::TypeId::of::<N>(),
+            std::any::TypeId::of::<reth_ethereum_primitives::EthPrimitives>(),
+            "pipe_run_inner requires N = EthPrimitives"
+        );
         let pipe_event_rx =
-            get_pipe_exec_layer_event_bus::<N>().event_rx.lock().unwrap().take().unwrap();
+            get_pipe_exec_layer_event_bus().event_rx.lock().unwrap().take().unwrap();
+        // Safety: The TypeId assertion above guarantees N == EthPrimitives,
+        // so Receiver<PipeExecLayerEvent<EthPrimitives>> and Receiver<PipeExecLayerEvent<N>>
+        // are the same type at runtime.
+        let pipe_event_rx: std::sync::mpsc::Receiver<PipeExecLayerEvent<N>> =
+            unsafe { std::mem::transmute(pipe_event_rx) };
         loop {
             match self.try_recv_pipe_exec_event(&pipe_event_rx) {
                 Ok(Some(event)) => self.on_pipe_exec_event(event),

crates/engine/tree/src/tree/mod.rs

@@ -13,8 +13,8 @@ workspace = true
 
 [dependencies]
 reth-primitives.workspace = true
+reth-ethereum-primitives.workspace = true
 reth-chain-state.workspace = true
 alloy-primitives.workspace = true
-once_cell.workspace = true
 tokio.workspace = true
 tracing.workspace = true

crates/pipe-exec-layer-ext-v2/event-bus/Cargo.toml

@@ -1,30 +1,38 @@
 //! Event bus for the pipe execution layer.
 
 use alloy_primitives::TxHash;
-use once_cell::sync::OnceCell;
 use reth_chain_state::ExecutedBlockWithTrieUpdates;
+use reth_ethereum_primitives::EthPrimitives;
 use reth_primitives::NodePrimitives;
-use std::{any::Any, thread::sleep, time::Duration};
+use std::{sync::OnceLock, thread::sleep, time::Duration};
 use tokio::sync::{mpsc::UnboundedReceiver, oneshot};
 use tracing::info;
 
 /// A static instance of `PipeExecLayerEventBus` used for dispatching events.
-pub static PIPE_EXEC_LAYER_EVENT_BUS: OnceCell<Box<dyn Any + Send + Sync>> = OnceCell::new();
+/// Uses typed `OnceLock` instead of `Box<dyn Any>` to eliminate the runtime downcast
+/// and make type mismatches a compile-time error.
+pub static PIPE_EXEC_LAYER_EVENT_BUS: OnceLock<PipeExecLayerEventBus<EthPrimitives>> =
+    OnceLock::new();
 
 /// Get a reference to the global `PipeExecLayerEventBus` instance.
-pub fn get_pipe_exec_layer_event_bus<N: NodePrimitives>() -> &'static PipeExecLayerEventBus<N> {
-    let mut wait_time = 0;
+/// Blocks until the event bus is initialized, with a maximum timeout of 120 seconds.
+pub fn get_pipe_exec_layer_event_bus() -> &'static PipeExecLayerEventBus<EthPrimitives> {
+    const MAX_WAIT_SECS: u64 = 120;
+    let start = std::time::Instant::now();
     loop {
-        let event_bus = PIPE_EXEC_LAYER_EVENT_BUS
-            .get()
-            .map(|ext| ext.downcast_ref::<PipeExecLayerEventBus<N>>().unwrap());
-        if let Some(event_bus) = event_bus {
-            break event_bus;
-        } else if wait_time % 5 == 0 {
+        if let Some(event_bus) = PIPE_EXEC_LAYER_EVENT_BUS.get() {
+            return event_bus;
+        }
+        assert!(
+            start.elapsed().as_secs() < MAX_WAIT_SECS,
+            "PipeExecLayerEventBus not initialized after {}s — \
+             likely a startup ordering bug",
+            MAX_WAIT_SECS
+        );
+        if start.elapsed().as_secs().is_multiple_of(5) {
             info!("Wait PipeExecLayerEventBus ready...");
         }
         sleep(Duration::from_secs(1));
-        wait_time += 1;
     }
 }
 

crates/pipe-exec-layer-ext-v2/event-bus/src/lib.rs

@@ -235,6 +235,21 @@ struct Core<Storage: GravityStorage> {
     make_canonical_barrier: Channel<u64 /* block number */, Instant>,
     discard_txs_tx: UnboundedSender<Vec<TxHash>>,
     cache: PersistBlockCache,
+    // Ordering rationale: `Ordering::Release` (stores) / `Ordering::Acquire` (loads) is
+    // sufficient — `SeqCst` is unnecessary. The `execute_block_barrier` Channel (backed by
+    // `Mutex<Inner>`) enforces strict serial block execution: block N's `process()` cannot
+    // proceed past `wait((epoch, N-1))` until block N-1 has completed and called
+    // `notify((epoch, N-1), ...)`. Both stores sit inside this serialized critical section,
+    // so there is only ever a single writer — no concurrent writer can interleave.
+    //
+    // The only concurrent readers are in the timeout branch below (`self.epoch()` /
+    // `self.execute_height()`), which merely check for stale/duplicate blocks. Each
+    // individual Acquire load sees the latest Release store (no staleness); the two
+    // reads are simply not atomic *with respect to each other*, so at worst a concurrent
+    // update between the two loads causes one extra wait-loop iteration, never an
+    // incorrect discard. Note that even `SeqCst` would not help here — two separate
+    // loads are never atomic as a pair regardless of ordering; only an `AtomicU128`
+    // or a lock could provide an atomic snapshot, but neither is needed.
     epoch: AtomicU64,
     execute_height: AtomicU64,
     metrics: PipeExecLayerMetrics,
@@ -374,7 +389,11 @@ impl<Storage: GravityStorage> Core<Storage> {
                 .await
             {
                 Some(parent) => break parent,
-                // Make sure the ordered blocks are idempotent
+                // Make sure the ordered blocks are idempotent.
+                // NOTE: Each Acquire load below sees the latest Release
+                // store to its respective atomic. The two reads are not mutually
+                // atomic, but a concurrent update between them at worst causes one
+                // extra wait-loop iteration, never an incorrect discard.
                 None => {
                     if block_epoch < self.epoch() || block_number <= self.execute_height() {
                         warn!(target: "PipeExecService.process",
@@ -456,8 +475,12 @@ impl<Storage: GravityStorage> Core<Storage> {
                 new_epoch=?epoch,
                 "new epoch"
             );
+            // SAFETY: Release ordering is sufficient here — see comment on field
+            // declarations.
             assert_eq!(self.epoch.fetch_max(epoch, Ordering::Release), block_epoch);
         }
+        // SAFETY: Release ordering is sufficient — the execute_block_barrier
+        // serializes writers; only the timeout branch reads these concurrently (harmlessly).
         assert_eq!(self.execute_height.fetch_add(1, Ordering::Release), block_number - 1);
         self.execute_block_barrier
             .notify(
@@ -1444,11 +1467,9 @@ where
     };
     tokio::spawn(service.run());
 
-    PIPE_EXEC_LAYER_EVENT_BUS.get_or_init(|| {
-        Box::new(PipeExecLayerEventBus {
-            event_rx: std::sync::Mutex::new(Some(event_rx)),
-            discard_txs: tokio::sync::Mutex::new(Some(discard_txs_rx)),
-        })
+    PIPE_EXEC_LAYER_EVENT_BUS.get_or_init(|| PipeExecLayerEventBus {
+        event_rx: std::sync::Mutex::new(Some(event_rx)),
+        discard_txs: tokio::sync::Mutex::new(Some(discard_txs_rx)),
     });
 
     PipeExecLayerApi {

crates/pipe-exec-layer-ext-v2/execute/src/lib.rs

@@ -1,21 +1,22 @@
 //! Metadata transaction execution
 
 use super::{
+    new_system_call_txn,
     types::{convert_active_validators_to_bcs, onBlockStartCall, NewEpochEvent},
     SYSTEM_CALLER,
 };
 use crate::{onchain_config::BLOCK_ADDR, ExecuteOrderedBlockResult, OrderedBlock};
-use alloy_consensus::{constants::EMPTY_WITHDRAWALS, Header, TxLegacy, EMPTY_OMMER_ROOT_HASH};
+use alloy_consensus::{constants::EMPTY_WITHDRAWALS, Header, EMPTY_OMMER_ROOT_HASH};
 use alloy_eips::{eip4895::Withdrawals, merge::BEACON_NONCE};
-use alloy_primitives::{Address, Bytes, Signature, TxKind, U256};
+use alloy_primitives::{Bytes, U256};
 use alloy_sol_types::{SolCall, SolEvent};
 use gravity_api_types::events::contract_event::GravityEvent;
 use gravity_primitives::get_gravity_config;
 use reth_chainspec::{ChainSpec, EthereumHardforks};
-use reth_ethereum_primitives::{Block, BlockBody, Transaction, TransactionSigned};
-use reth_evm::{Evm, IntoTxEnv};
+use reth_ethereum_primitives::{Block, BlockBody, TransactionSigned};
+use reth_evm::Evm;
 use reth_execution_types::BlockExecutionOutput;
-use reth_primitives::{Receipt, Recovered};
+use reth_primitives::Receipt;
 use reth_provider::BlockExecutionResult;
 use revm::{
     context::TxEnv,
@@ -208,27 +209,6 @@ pub fn transact_system_txn(
     (SystemTxnResult { result: result.result, txn }, result.state)
 }
 
-/// Create a new system call transaction
-fn new_system_call_txn(
-    contract: Address,
-    nonce: u64,
-    gas_price: u128,
-    input: Bytes,
-) -> TransactionSigned {
-    TransactionSigned::new_unhashed(
-        Transaction::Legacy(TxLegacy {
-            chain_id: None,
-            nonce,
-            gas_price,
-            gas_limit: 30_000_000,
-            to: TxKind::Call(contract),
-            value: U256::ZERO,
-            input,
-        }),
-        Signature::new(U256::ZERO, U256::ZERO, false),
-    )
-}
-
 /// Execute a metadata contract call (onBlockStart from Blocker.sol)
 ///
 /// Calls Blocker.onBlockStart(proposerIndex, failedProposerIndices, timestampMicros)

crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/metadata_txn.rs

@@ -102,7 +102,7 @@ use alloy_consensus::{EthereumTxEnvelope, TxEip4844, TxLegacy};
 use alloy_primitives::{Bytes, Signature, U256};
 use reth_ethereum_primitives::{Transaction, TransactionSigned};
 use revm_primitives::TxKind;
-use tracing::{debug, info, warn};
+use tracing::{debug, info};
 
 /// Construct validator transactions envelope (JWK updates and DKG transcripts)
 ///

crates/pipe-exec-layer-ext-v2/execute/src/onchain_config/mod.rs

@@ -6,7 +6,7 @@ use crate::{
     data_source::{source_types, OracleData, OracleDataSource},
     eth_client::EthHttpCli,
 };
-use alloy_primitives::{hex, Address, Bytes, U256};
+use alloy_primitives::{Address, Bytes, U256};
 use alloy_rpc_types::Filter;
 use alloy_sol_macro::sol;
 use alloy_sol_types::SolEvent;
@@ -340,6 +340,7 @@ impl OracleDataSource for BlockchainEventSource {
 mod tests {
     use super::*;
     use crate::data_source::OracleDataSource;
+    use alloy_primitives::hex;
 
     // =========================================================================
     // Fixed Anvil Deployment Addresses

crates/pipe-exec-layer-ext-v2/relayer/src/blockchain_source.rs

@@ -183,7 +183,7 @@ pub async fn maintain_transaction_pool<N, Client, P, St, Tasks>(
         let pool = pool.clone();
         tokio::spawn(async move {
             let mut discard_txs_rx =
-                get_pipe_exec_layer_event_bus::<N>().discard_txs.lock().await.take().unwrap();
+                get_pipe_exec_layer_event_bus().discard_txs.lock().await.take().unwrap();
             while let Some(discard_txs) = discard_txs_rx.recv().await {
                 debug!(target: "txpool", count=%discard_txs.len(), "discarding transactions");
                 pool.remove_transactions(discard_txs);