Merge pull request #132 from anton-rs/rf/chore/get-running

chore: More refactors

Cargo.toml

@@ -37,8 +37,8 @@ all-features = true
 rustdoc-args = ["--cfg", "docsrs"]
 
 [patch.crates-io]
-kona-derive = { git = "https://github.com/anton-rs/kona", branch = "rf/chore/driver-advancing" }
-kona-driver = { git = "https://github.com/anton-rs/kona", branch = "rf/chore/driver-advancing" }
+kona-derive = { git = "https://github.com/anton-rs/kona", branch = "rf/engine-waiting" }
+kona-driver = { git = "https://github.com/anton-rs/kona", branch = "rf/engine-waiting" }
 
 [workspace.dependencies]
 # Workspace

crates/driver/Cargo.toml

@@ -28,6 +28,7 @@ alloy-network.workspace = true
 alloy-transport.workspace = true
 alloy-consensus.workspace = true
 alloy-rpc-types-eth.workspace = true
+alloy-rpc-types-engine = { workspace = true, features = ["jwt", "serde"] }
 alloy-provider = { workspace = true, features = ["ipc", "ws", "reqwest"] }
 alloy-rpc-types = { workspace = true, features = ["ssz"] }
 alloy-primitives = { workspace = true, features = ["map"] }

crates/driver/src/config.rs

@@ -1,5 +1,6 @@
 //! Configuration for the Hilo Driver.
 
+use alloy_rpc_types_engine::JwtSecret;
 use kona_derive::traits::ChainProvider;
 use kona_driver::PipelineCursor;
 use op_alloy_genesis::RollupConfig;
@@ -46,10 +47,30 @@ pub struct Config {
     pub rollup_config: RollupConfig,
     /// The hilo-node RPC server
     pub rpc_url: Option<Url>,
+    /// Engine API JWT Secret.
+    /// This is used to authenticate with the engine API
+    #[serde(deserialize_with = "deserialize_jwt_secret", serialize_with = "as_hex")]
+    pub jwt_secret: JwtSecret,
     /// The cache size for in-memory providers.
     pub cache_size: usize,
 }
 
+fn as_hex<S>(v: &JwtSecret, serializer: S) -> Result<S::Ok, S::Error>
+where
+    S: serde::ser::Serializer,
+{
+    let encoded = alloy_primitives::hex::encode(v.as_bytes());
+    serializer.serialize_str(&encoded)
+}
+
+fn deserialize_jwt_secret<'de, D>(deserializer: D) -> Result<JwtSecret, D::Error>
+where
+    D: serde::de::Deserializer<'de>,
+{
+    let s: &str = serde::de::Deserialize::deserialize(deserializer)?;
+    JwtSecret::from_hex(s).map_err(serde::de::Error::custom)
+}
+
 impl Config {
     /// Construct an [OnlineBlobProviderWithFallback] from the [Config].
     pub async fn blob_provider(
@@ -122,6 +143,11 @@ impl Config {
             .block_info_by_number(l1_origin_number)
             .await
             .map_err(|e| ConfigError::ChainProvider(e.to_string()))?;
-        Ok(PipelineCursor::new(channel_timeout, l1_origin))
+        let mut cursor = PipelineCursor::new(channel_timeout, l1_origin);
+        // TODO: construct a valid tip cursor
+        let tip =
+            kona_driver::TipCursor::new(safe_head_info, Default::default(), Default::default());
+        cursor.advance(l1_origin, tip);
+        Ok(cursor)
     }
 }

crates/driver/src/context/standalone.rs

@@ -199,6 +199,7 @@ impl Context for StandaloneContext {
 
         let header = self.new_block_rx.recv().await?;
         let block_num = header.number;
+        info!("Received new block: {}", block_num);
 
         let entry = self.reorg_cache.entry(block_num).or_default();
         entry.insert(header.hash, header.clone());

crates/driver/src/driver.rs

@@ -4,8 +4,9 @@ use alloy_transport::TransportResult;
 use kona_derive::{errors::PipelineErrorKind, traits::SignalReceiver, types::ResetSignal};
 use kona_driver::{Driver, PipelineCursor, TipCursor};
 use std::sync::Arc;
+// use tokio::sync::watch::{channel, Receiver};
 
-use hilo_engine::{EngineApi, HiloExecutorConstructor};
+use hilo_engine::EngineController;
 use hilo_providers_local::{InMemoryChainProvider, InMemoryL2ChainProvider};
 
 use crate::{
@@ -14,8 +15,7 @@ use crate::{
 };
 
 /// A driver from [kona_driver] that uses hilo-types.
-pub type KonaDriver =
-    Driver<EngineApi, HiloExecutorConstructor, HiloPipeline, HiloDerivationPipeline>;
+pub type KonaDriver = Driver<EngineController, HiloPipeline, HiloDerivationPipeline>;
 
 /// An error that can happen when running the driver.
 #[derive(Debug, thiserror::Error)]
@@ -29,6 +29,9 @@ pub enum DriverError {
     /// Kona's driver unexpectedly errored.
     #[error("kona driver error")]
     DriverErrored,
+    /// Shutdown signal received.
+    #[error("shutdown signal received")]
+    Shutdown,
 }
 
 /// HiloDriver is a wrapper around the `Driver` that
@@ -39,15 +42,13 @@ pub struct HiloDriver<C: Context> {
     pub ctx: C,
     /// The driver config.
     pub cfg: Config,
-    /// A constructor for execution.
-    pub exec: Option<HiloExecutorConstructor>,
 }
 
 impl HiloDriver<StandaloneContext> {
     /// Creates a new [HiloDriver] with a standalone context.
-    pub async fn standalone(cfg: Config, exec: HiloExecutorConstructor) -> TransportResult<Self> {
+    pub async fn standalone(cfg: Config) -> TransportResult<Self> {
         let ctx = StandaloneContext::new(cfg.l1_rpc_url.clone()).await?;
-        Ok(Self::new(cfg, ctx, exec))
+        Ok(Self::new(cfg, ctx))
     }
 }
 
@@ -56,8 +57,8 @@ where
     C: Context,
 {
     /// Constructs a new [HiloDriver].
-    pub fn new(cfg: Config, ctx: C, exec: HiloExecutorConstructor) -> Self {
-        Self { cfg, ctx, exec: Some(exec) }
+    pub fn new(cfg: Config, ctx: C) -> Self {
+        Self { cfg, ctx }
     }
 
     /// Initializes the [HiloPipeline].
@@ -77,7 +78,13 @@ where
     pub async fn init_driver(&mut self) -> Result<KonaDriver, ConfigError> {
         let cursor = self.cfg.tip_cursor().await?;
         let pipeline = self.init_pipeline(cursor.clone()).await?;
-        let exec = self.exec.take().expect("Executor not set");
+        let exec = EngineController::new(
+            self.cfg.l2_engine_url.clone(),
+            self.cfg.jwt_secret,
+            cursor.origin(),
+            cursor.l2_safe_head().block_info.into(),
+            &self.cfg.rollup_config,
+        );
         Ok(Driver::new(cursor, exec, pipeline))
     }
 
@@ -122,6 +129,9 @@ where
         let mut driver = self.init_driver().await?;
         info!("Driver initialized");
 
+        // Wait until the engine is ready
+        driver.wait_for_executor().await;
+
         // Step 3: Start the processing loop
         loop {
             tokio::select! {
@@ -142,6 +152,16 @@ where
         }
     }
 
+    // Exits if a SIGINT signal is received
+    // fn check_shutdown(&self) -> Result<(), DriverError> {
+    //     if *self.shutdown_recv.borrow() {
+    //         tracing::warn!("shutting down");
+    //         std::process::exit(1);
+    //     }
+    //
+    //     Ok(())
+    // }
+
     /// Wait for the L2 genesis' corresponding L1 block to be available in the L1 chain.
     async fn wait_for_l2_genesis_l1_block(&mut self) {
         loop {

crates/engine/Cargo.toml

@@ -28,11 +28,15 @@ alloy-transport-http = { workspace = true, features = ["jwt-auth"] }
 alloy-rpc-types-engine = { workspace = true, features = ["jwt", "serde"] }
 
 # Op Alloy
+op-alloy-genesis.workspace = true
 op-alloy-provider.workspace = true
+op-alloy-protocol.workspace = true
 op-alloy-rpc-types-engine.workspace = true
 
 # Misc
+async-trait.workspace = true
 url.workspace = true
+tokio.workspace = true
 tracing.workspace = true
 tower.workspace = true
 http-body-util.workspace = true