feat(full-node): watch L2 chain tip

Author: kariy

Cargo.lock

@@ -29,10 +29,6 @@ pub struct FullNodeArgs {
     #[arg(value_name = "PATH")]
     pub db_dir: PathBuf,
 
-    #[arg(long = "eth.rpc")]
-    #[arg(value_name = "PATH")]
-    pub eth_rpc_url: String,
-
     #[arg(long)]
     pub network: Network,
 
@@ -105,7 +101,6 @@ impl FullNodeArgs {
             rpc,
             metrics,
             network: self.network,
-            eth_rpc_url: self.eth_rpc_url.clone(),
             gateway_api_key: self.gateway_api_key.clone(),
         })
     }

crates/cli/src/full.rs

@@ -42,8 +42,6 @@ strum.workspace = true
 strum_macros.workspace = true
 
 tokio = { workspace = true, features = [ "time" ] }
-katana-starknet.workspace = true
-alloy-provider = { workspace = true, default-features = false, features = [ "reqwest", "reqwest-rustls-tls" ] }
 
 [features]
 cartridge = [ "katana-rpc-api/cartridge", "katana-rpc-server/cartridge" ]

crates/node/Cargo.toml

@@ -3,7 +3,6 @@
 use std::future::IntoFuture;
 use std::sync::Arc;
 
-use alloy_provider::RootProvider;
 use anyhow::Result;
 use http::header::CONTENT_TYPE;
 use http::Method;
@@ -65,7 +64,6 @@ pub struct Config {
     pub rpc: RpcConfig,
     pub metrics: Option<MetricsConfig>,
     pub gateway_api_key: Option<String>,
-    pub eth_rpc_url: String,
     pub network: Network,
 }
 
@@ -78,7 +76,7 @@ pub struct Node {
     pub pipeline: Pipeline<DbProvider>,
     pub rpc_server: RpcServer,
     pub gateway_client: SequencerGateway,
-    pub chain_tip_watcher: ChainTipWatcher<RootProvider>,
+    pub chain_tip_watcher: ChainTipWatcher<SequencerGateway>,
 }
 
 impl Node {
@@ -129,18 +127,9 @@ impl Node {
         pipeline.add_stage(Classes::new(provider.clone(), gateway_client.clone(), 8));
         pipeline.add_stage(StateTrie::new(provider.clone()));
 
-        // --
+        // -- build chain tip watcher using gateway client
 
-        let core_contract = match config.network {
-            Network::Mainnet => {
-                katana_starknet::StarknetCore::new_http_mainnet(&config.eth_rpc_url)?
-            }
-            Network::Sepolia => {
-                katana_starknet::StarknetCore::new_http_sepolia(&config.eth_rpc_url)?
-            }
-        };
-
-        let chain_tip_watcher = ChainTipWatcher::new(core_contract);
+        let chain_tip_watcher = ChainTipWatcher::new(gateway_client.clone());
 
         let preconf_factory = PreconfStateFactory::new(
             provider.clone(),

crates/node/src/full/mod.rs

@@ -1,30 +1,44 @@
 use std::future::IntoFuture;
 use std::time::Duration;
 
-use alloy_provider::Provider;
 use anyhow::Result;
 use futures::future::BoxFuture;
+use katana_gateway_types::BlockId;
 use katana_primitives::block::BlockNumber;
-use katana_starknet::StarknetCore;
 use tokio::sync::watch;
 use tracing::{error, info};
 
 pub type TipWatcherFut = BoxFuture<'static, Result<()>>;
 
+/// A trait for abstracting the source of the latest block number.
+///
+/// This allows the chain tip watcher to work with different sources such as:
+/// - Starknet Core Contract on L1 (Ethereum) - tracks the settled/proven tip
+/// - Starknet RPC endpoints - tracks the latest L2 tip
+/// - Feeder Gateway - tracks the latest L2 tip from the sequencer
+pub trait ChainTipProvider: Send + Sync {
+    /// Retrieves the latest block number from the source.
+    ///
+    /// # Returns
+    ///
+    /// Returns the latest block number.
+    fn latest_number(&self) -> BoxFuture<'_, Result<BlockNumber>>;
+}
+
 pub struct ChainTipWatcher<P> {
-    /// The Starknet Core Contract client for fetching the latest block.
-    core_contract: StarknetCore<P>,
+    /// The block number provider for fetching the latest block.
+    tip_provider: P,
     /// Interval for checking the new tip.
     watch_interval: Duration,
     /// Watch channel for notifying subscribers of the latest tip.
     tip_sender: watch::Sender<BlockNumber>,
 }
 
-impl<P: alloy_provider::Provider> ChainTipWatcher<P> {
-    pub fn new(core_contract: StarknetCore<P>) -> Self {
+impl<P: ChainTipProvider> ChainTipWatcher<P> {
+    pub fn new(provider: P) -> Self {
         let (tip_tx, _) = watch::channel(0);
         let watch_interval = Duration::from_secs(30);
-        Self { core_contract, watch_interval, tip_sender: tip_tx }
+        Self { tip_provider: provider, watch_interval, tip_sender: tip_tx }
     }
 
     /// Set the watch interval for checking new tips.
@@ -47,7 +61,7 @@ impl<P: alloy_provider::Provider> ChainTipWatcher<P> {
         let mut prev_tip: BlockNumber = 0;
 
         loop {
-            let block_number = self.core_contract.state_block_number().await? as BlockNumber;
+            let block_number = self.tip_provider.latest_number().await?;
 
             if prev_tip != block_number {
                 info!(block = %block_number, "New tip found.");
@@ -64,7 +78,7 @@ impl<P: alloy_provider::Provider> ChainTipWatcher<P> {
     }
 }
 
-impl<P: Provider + 'static> IntoFuture for ChainTipWatcher<P> {
+impl<P: ChainTipProvider + 'static> IntoFuture for ChainTipWatcher<P> {
     type Output = Result<()>;
     type IntoFuture = TipWatcherFut;
 
@@ -77,10 +91,10 @@ impl<P: Provider + 'static> IntoFuture for ChainTipWatcher<P> {
     }
 }
 
-impl<P: std::fmt::Debug> std::fmt::Debug for ChainTipWatcher<P> {
+impl<P> std::fmt::Debug for ChainTipWatcher<P> {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("ChainTipWatcher")
-            .field("core_contract", &self.core_contract)
+            .field("provider", &"ChainTipProvider")
             .field("subscribers", &self.tip_sender.receiver_count())
             .field("watch_interval", &self.watch_interval)
             .finish()
@@ -109,3 +123,126 @@ impl std::fmt::Debug for TipSubscription {
         f.debug_struct("TipSubscription").field("current_tip", &self.tip()).finish()
     }
 }
+
+/// Implementation of [`ChainTipProvider`] for the feeder gateway client.
+///
+/// This fetches the latest L2 block number directly from the Starknet feeder gateway,
+/// which may be ahead of the L1 settlement.
+impl ChainTipProvider for katana_gateway_client::Client {
+    fn latest_number(&self) -> BoxFuture<'_, Result<BlockNumber>> {
+        Box::pin(async move {
+            let block = self.get_block(BlockId::Latest).await?;
+            block.block_number.ok_or_else(|| anyhow::anyhow!("Block number not available"))
+        })
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use std::sync::atomic::{AtomicU64, Ordering};
+    use std::sync::Arc;
+
+    use super::*;
+
+    /// Mock provider that returns a sequence of block numbers from an atomic counter.
+    #[derive(Clone)]
+    struct MockProvider {
+        counter: Arc<AtomicU64>,
+    }
+
+    impl MockProvider {
+        fn new(initial: BlockNumber) -> Self {
+            Self { counter: Arc::new(AtomicU64::new(initial)) }
+        }
+
+        fn set(&self, value: BlockNumber) {
+            self.counter.store(value, Ordering::SeqCst);
+        }
+
+        fn increment(&self) {
+            self.counter.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    impl ChainTipProvider for MockProvider {
+        fn latest_number(&self) -> BoxFuture<'_, Result<BlockNumber>> {
+            let value = self.counter.load(Ordering::SeqCst);
+            Box::pin(async move { Ok(value) })
+        }
+    }
+
+    #[tokio::test]
+    async fn tip_updates_are_broadcast_to_subscribers() {
+        let provider = MockProvider::new(100);
+        let watcher = ChainTipWatcher::new(provider.clone()).interval(Duration::from_millis(10));
+
+        let mut sub1 = watcher.subscribe();
+        let sub2 = watcher.subscribe();
+
+        // Initial value should be 0 (default)
+        assert_eq!(sub1.tip(), 0);
+        assert_eq!(sub2.tip(), 0);
+
+        // Spawn the watcher task
+        let handle = tokio::spawn(async move { watcher.run().await });
+
+        // Wait for first update (block 100)
+        let tip = sub1.changed().await.unwrap();
+        assert_eq!(tip, 100);
+        assert_eq!(sub2.tip(), 100);
+
+        // Update provider and wait for new tip
+        provider.set(150);
+        tokio::time::sleep(Duration::from_millis(20)).await;
+
+        let tip = sub1.changed().await.unwrap();
+        assert_eq!(tip, 150);
+        assert_eq!(sub2.tip(), 150);
+
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn duplicate_tips_are_not_rebroadcast() {
+        let provider = MockProvider::new(100);
+        let watcher = ChainTipWatcher::new(provider.clone()).interval(Duration::from_millis(10));
+
+        let mut sub = watcher.subscribe();
+
+        let handle = tokio::spawn(async move { watcher.run().await });
+
+        // Wait for first update
+        let tip = sub.changed().await.unwrap();
+        assert_eq!(tip, 100);
+
+        // Keep the same tip for multiple intervals
+        tokio::time::sleep(Duration::from_millis(50)).await;
+
+        // Should timeout as no new update is expected
+        let result = tokio::time::timeout(Duration::from_millis(30), sub.changed()).await;
+        assert!(result.is_err(), "Should not receive duplicate tip updates");
+
+        handle.abort();
+    }
+
+    #[tokio::test]
+    async fn monotonically_increasing_tips() {
+        let provider = MockProvider::new(1);
+        let watcher = ChainTipWatcher::new(provider.clone()).interval(Duration::from_millis(10));
+
+        let mut sub = watcher.subscribe();
+
+        let handle = tokio::spawn(async move { watcher.run().await });
+
+        // Verify multiple sequential updates
+        for expected in 1..=5 {
+            let tip = sub.changed().await.unwrap();
+            assert_eq!(tip, expected);
+
+            provider.increment(); // Increment the chain tip
+            tokio::time::sleep(Duration::from_millis(15)).await;
+        }
+
+        handle.abort();
+    }
+}