Merge pull request #133 from anton-rs/rf/use-l2-rpc-for-block-ref

chore: use l2 block rpc

crates/driver/src/driver.rs

@@ -80,6 +80,7 @@ where
         let pipeline = self.init_pipeline(cursor.clone()).await?;
         let exec = EngineController::new(
             self.cfg.l2_engine_url.clone(),
+            self.cfg.l2_rpc_url.clone(),
             self.cfg.jwt_secret,
             cursor.origin(),
             cursor.l2_safe_head().block_info.into(),

crates/engine/Cargo.toml

@@ -13,6 +13,9 @@ repository.workspace = true
 rust-version.workspace = true
 
 [dependencies]
+# Hilo
+hilo-providers-alloy.workspace = true
+
 # Kona
 kona-driver.workspace = true
 

crates/engine/src/client.rs

@@ -3,7 +3,7 @@
 use alloy_eips::eip1898::BlockNumberOrTag;
 use alloy_network::AnyNetwork;
 use alloy_primitives::{Bytes, B256};
-use alloy_provider::RootProvider;
+use alloy_provider::{ReqwestProvider, RootProvider};
 use alloy_rpc_client::RpcClient;
 use alloy_rpc_types_engine::{
     ExecutionPayloadV3, ForkchoiceState, ForkchoiceUpdated, JwtSecret, PayloadId, PayloadStatus,
@@ -17,12 +17,16 @@ use alloy_transport_http::{
 };
 use async_trait::async_trait;
 use http_body_util::Full;
-use op_alloy_protocol::L2BlockInfo;
+use op_alloy_genesis::RollupConfig;
+use op_alloy_protocol::{BatchValidationProvider, L2BlockInfo};
 use op_alloy_provider::ext::engine::OpEngineApi;
 use op_alloy_rpc_types_engine::{OpExecutionPayloadEnvelopeV3, OpPayloadAttributes};
+use std::sync::Arc;
 use tower::ServiceBuilder;
 use url::Url;
 
+use hilo_providers_alloy::AlloyL2ChainProvider;
+
 use crate::{Engine, EngineApiError};
 
 /// A Hyper HTTP client with a JWT authentication layer.
@@ -31,24 +35,28 @@ type HyperAuthClient<B = Full<Bytes>> = HyperClient<B, AuthService<Client<HttpCo
 /// An external engine api client
 #[derive(Debug, Clone)]
 pub struct EngineClient {
-    /// The inner provider
-    provider: RootProvider<Http<HyperAuthClient>, AnyNetwork>,
+    /// The L2 engine provider.
+    engine: RootProvider<Http<HyperAuthClient>, AnyNetwork>,
+    /// The L2 chain provider.
+    rpc: AlloyL2ChainProvider,
 }
 
 impl EngineClient {
     /// Creates a new [`EngineClient`] from the provided [Url] and [JwtSecret].
-    pub fn new_http(url: Url, jwt: JwtSecret) -> Self {
+    pub fn new_http(engine: Url, rpc: Url, cfg: Arc<RollupConfig>, jwt: JwtSecret) -> Self {
         let hyper_client = Client::builder(TokioExecutor::new()).build_http::<Full<Bytes>>();
 
         let auth_layer = AuthLayer::new(jwt);
         let service = ServiceBuilder::new().layer(auth_layer).service(hyper_client);
 
         let layer_transport = HyperClient::with_service(service);
-        let http_hyper = Http::with_client(layer_transport, url);
+        let http_hyper = Http::with_client(layer_transport, engine);
         let rpc_client = RpcClient::new(http_hyper, true);
-        let provider = RootProvider::<_, AnyNetwork>::new(rpc_client);
+        let engine = RootProvider::<_, AnyNetwork>::new(rpc_client);
 
-        Self { provider }
+        let rpc = ReqwestProvider::new_http(rpc);
+        let rpc = AlloyL2ChainProvider::new(rpc, cfg);
+        Self { engine, rpc }
     }
 }
 
@@ -60,15 +68,15 @@ impl Engine for EngineClient {
         &self,
         payload_id: PayloadId,
     ) -> Result<OpExecutionPayloadEnvelopeV3, Self::Error> {
-        self.provider.get_payload_v3(payload_id).await.map_err(|_| EngineApiError::PayloadError)
+        self.engine.get_payload_v3(payload_id).await.map_err(|_| EngineApiError::PayloadError)
     }
 
     async fn forkchoice_update(
         &self,
         state: ForkchoiceState,
         attr: Option<OpPayloadAttributes>,
     ) -> Result<ForkchoiceUpdated, Self::Error> {
-        self.provider
+        self.engine
             .fork_choice_updated_v2(state, attr)
             .await
             .map_err(|_| EngineApiError::PayloadError)
@@ -79,24 +87,31 @@ impl Engine for EngineClient {
         payload: ExecutionPayloadV3,
         parent_beacon_block_root: B256,
     ) -> Result<PayloadStatus, Self::Error> {
-        self.provider
+        self.engine
             .new_payload_v3(payload, parent_beacon_block_root)
             .await
             .map_err(|_| EngineApiError::PayloadError)
     }
 
-    async fn l2_block_ref_by_label(&self, _: BlockNumberOrTag) -> Result<L2BlockInfo, Self::Error> {
-        // Convert the payload into an L2 block info.
-        // go impl uses an L2 client and fetches block by number, converting block to payload and
-        // payload to L2 block info.
-        todo!("implement l2_block_ref_by_label for the engine client")
+    async fn l2_block_ref_by_label(
+        &mut self,
+        numtag: BlockNumberOrTag,
+    ) -> Result<L2BlockInfo, Self::Error> {
+        let number = match numtag {
+            BlockNumberOrTag::Number(n) => n,
+            BlockNumberOrTag::Latest => {
+                self.rpc.latest_block_number().await.map_err(|_| EngineApiError::PayloadError)?
+            }
+            _ => return Err(EngineApiError::PayloadError),
+        };
+        self.rpc.l2_block_info_by_number(number).await.map_err(|_| EngineApiError::PayloadError)
     }
 }
 
 impl std::ops::Deref for EngineClient {
     type Target = RootProvider<Http<HyperAuthClient>, AnyNetwork>;
 
     fn deref(&self) -> &Self::Target {
-        &self.provider
+        &self.engine
     }
 }

crates/engine/src/controller.rs

@@ -10,7 +10,7 @@ use kona_driver::Executor;
 use op_alloy_genesis::RollupConfig;
 use op_alloy_protocol::BlockInfo;
 use op_alloy_rpc_types_engine::OpPayloadAttributes;
-use std::time::Duration;
+use std::{sync::Arc, time::Duration};
 use tokio::time::sleep;
 use url::Url;
 
@@ -56,12 +56,18 @@ impl EngineController {
     /// Creates a new engine controller.
     pub fn new(
         l2_engine_url: Url,
+        l2_rpc_url: Url,
         jwt_secret: JwtSecret,
         finalized_head: BlockInfo,
         finalized_epoch: Epoch,
         config: &RollupConfig,
     ) -> Self {
-        let client = EngineClient::new_http(l2_engine_url.clone(), jwt_secret);
+        let client = EngineClient::new_http(
+            l2_engine_url.clone(),
+            l2_rpc_url,
+            Arc::new(config.clone()),
+            jwt_secret,
+        );
         Self {
             blocktime: config.block_time,
             unsafe_head: finalized_head,
@@ -93,15 +99,31 @@ impl Executor for EngineController {
     /// Waits for the engine to be ready.
     async fn wait_until_ready(&mut self) {
         let forkchoice = self.create_forkchoice_state();
-        // Loop until the forkchoice is updated
-        while !self.client.forkchoice_update(forkchoice, None).await.is_ok_and(|u| u.is_valid()) {
+        while self.client.forkchoice_update(forkchoice, None).await.is_err() {
             sleep(Duration::from_secs(1)).await;
         }
     }
 
     /// Updates the safe head.
-    fn update_safe_head(&mut self, _: Sealed<Header>) {
-        todo!()
+    fn update_safe_head(&mut self, header: Sealed<Header>) {
+        if self.safe_head.number < header.number {
+            self.safe_head = BlockInfo {
+                number: header.number,
+                hash: header.hash_slow(),
+                timestamp: header.timestamp,
+                parent_hash: header.parent_hash,
+            };
+            self.safe_epoch = self.safe_head.into();
+        }
+
+        if header.number > self.unsafe_head.number {
+            self.unsafe_head = BlockInfo {
+                number: header.number,
+                hash: header.hash_slow(),
+                timestamp: header.timestamp,
+                parent_hash: header.parent_hash,
+            };
+        }
     }
 
     /// Execute the given payload attributes.

crates/engine/src/traits.rs

@@ -38,7 +38,7 @@ pub trait Engine {
 
     /// Returns the [L2BlockInfo] for the given label.
     async fn l2_block_ref_by_label(
-        &self,
+        &mut self,
         label: BlockNumberOrTag,
     ) -> Result<L2BlockInfo, Self::Error>;
 }