Merge pull request #572 from SorellaLabs/jonas/feat/uniswap-v4-generics

feat(op-anstrom): uniswap v4 generic types

Author: mempirate

Cargo.lock

@@ -335,7 +335,7 @@ where
         init_telemetry(signer_addr, grace_shutdown)
     });
 
-    let uniswap_pool_manager = configure_uniswap_manager::<_, DEFAULT_TICKS>(
+    let uniswap_pool_manager = configure_uniswap_manager::<_, EthPrimitives, DEFAULT_TICKS>(
         querying_provider.clone(),
         eth_handle.subscribe_cannon_state_notifications().await,
         uniswap_registry,

bin/angstrom/src/components.rs

@@ -35,11 +35,10 @@ use angstrom_types::{
     reth_db_wrapper::RethDbWrapper,
     submission::SubmissionHandler
 };
-use consensus::{AngstromValidator, ConsensusManager, ManagerNetworkDeps};
+use consensus::AngstromValidator;
 use futures::Stream;
 use matching_engine::{MatchingManager, manager::MatcherCommand};
 use order_pool::{PoolConfig, PoolManagerUpdate, order_storage::OrderStorage};
-use parking_lot::RwLock;
 use reth::{
     api::NodeAddOns,
     builder::FullNodeComponents,
@@ -296,7 +295,7 @@ where
         init_telemetry(signer_addr, grace_shutdown)
     });
 
-    let uniswap_pool_manager = configure_uniswap_manager::<_, DEFAULT_TICKS>(
+    let uniswap_pool_manager = configure_uniswap_manager::<_, OpPrimitives, DEFAULT_TICKS>(
         querying_provider.clone(),
         eth_handle.subscribe_cannon_state_notifications().await,
         uniswap_registry,

bin/op-angstrom/src/components.rs

@@ -23,6 +23,8 @@ itertools.workspace = true
 malachite.workspace = true
 rand.workspace = true
 rayon = "1"
+reth-node-builder.workspace = true
+reth-primitives.workspace = true
 reth-provider.workspace = true
 serde.workspace = true
 telemetry-recorder.workspace = true

crates/uniswap-v4/Cargo.toml

@@ -17,6 +17,7 @@ use angstrom_types::{
 };
 use futures::Stream;
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
+use reth_node_builder::NodePrimitives;
 use reth_provider::{
     CanonStateNotifications, DatabaseProviderFactory, ReceiptProvider, StateProvider,
     TryIntoHistoricalStateProvider
@@ -119,16 +120,20 @@ where
         .collect::<Vec<_>>()
 }
 
-pub async fn configure_uniswap_manager<BlockSync: BlockSyncConsumer, const TICKS: u16>(
+pub async fn configure_uniswap_manager<
+    BlockSync: BlockSyncConsumer,
+    N: NodePrimitives,
+    const TICKS: u16
+>(
     provider: Arc<impl Provider + 'static>,
-    state_notification: CanonStateNotifications,
+    state_notification: CanonStateNotifications<N>,
     uniswap_pool_registry: UniswapPoolRegistry,
     current_block: BlockNumber,
     block_sync: BlockSync,
     pool_manager_address: Address,
     update_stream: Pin<Box<dyn Stream<Item = EthEvent> + Send + Sync>>
 ) -> UniswapPoolManager<
-    CanonicalStateAdapter<impl Provider + 'static>,
+    CanonicalStateAdapter<impl Provider + 'static, N>,
     impl Provider + 'static,
     BlockSync,
     TICKS
@@ -139,8 +144,11 @@ pub async fn configure_uniswap_manager<BlockSync: BlockSyncConsumer, const TICKS
         pool_manager_address
     );
 
-    let notifier =
-        Arc::new(CanonicalStateAdapter::new(state_notification, provider.clone(), current_block));
+    let notifier = Arc::new(CanonicalStateAdapter::<_, N>::new(
+        state_notification,
+        provider.clone(),
+        current_block
+    ));
 
     UniswapPoolManager::new(factory, current_block, notifier, block_sync, update_stream).await
 }

crates/uniswap-v4/src/lib.rs

@@ -11,25 +11,29 @@ use alloy::{
     rpc::types::{Filter, FilterBlockOption}
 };
 use futures_util::StreamExt;
+use reth_node_builder::NodePrimitives;
+use reth_primitives::EthPrimitives;
 use reth_provider::CanonStateNotification;
 use tokio::sync::broadcast;
 
 use super::PoolMangerBlocks;
 use crate::uniswap::{pool_manager::PoolManagerError, pool_providers::PoolManagerProvider};
 
-pub struct CanonicalStateAdapter<P>
+pub struct CanonicalStateAdapter<P, N = EthPrimitives>
 where
-    P: Provider + 'static
+    P: Provider + 'static,
+    N: NodePrimitives
 {
-    canon_state_notifications: broadcast::Receiver<CanonStateNotification>,
+    canon_state_notifications: broadcast::Receiver<CanonStateNotification<N>>,
     last_logs:                 Arc<RwLock<Vec<Log>>>,
     last_block_number:         Arc<AtomicU64>,
     node_provider:             Arc<P>
 }
 
-impl<P> Clone for CanonicalStateAdapter<P>
+impl<P, N> Clone for CanonicalStateAdapter<P, N>
 where
-    P: Provider + 'static
+    P: Provider + 'static,
+    N: NodePrimitives
 {
     fn clone(&self) -> Self {
         Self {
@@ -41,12 +45,13 @@ where
     }
 }
 
-impl<P> CanonicalStateAdapter<P>
+impl<P, N> CanonicalStateAdapter<P, N>
 where
-    P: Provider + 'static
+    P: Provider + 'static,
+    N: NodePrimitives
 {
     pub fn new(
-        canon_state_notifications: broadcast::Receiver<CanonStateNotification>,
+        canon_state_notifications: broadcast::Receiver<CanonStateNotification<N>>,
         node_provider: Arc<P>,
         block_number: u64
     ) -> Self {
@@ -59,9 +64,10 @@ where
     }
 }
 
-impl<P> PoolManagerProvider for CanonicalStateAdapter<P>
+impl<P, N> PoolManagerProvider for CanonicalStateAdapter<P, N>
 where
-    P: Provider + 'static
+    P: Provider + 'static,
+    N: NodePrimitives
 {
     fn provider(&self) -> Arc<impl Provider> {
         self.node_provider.clone()
@@ -81,11 +87,12 @@ where
 
                                 let logs: Vec<Log> = new
                                     .execution_outcome()
-                                    .logs(block.number)
+                                    .logs(block.number())
                                     .map_or_else(Vec::new, |logs| logs.cloned().collect());
                                 *last_log_write = logs;
-                                this.last_block_number.store(block.number, Ordering::SeqCst);
-                                tracing::info!(?block.number,"updated number");
+                                this.last_block_number
+                                    .store(block.number(), Ordering::SeqCst);
+                                tracing::info!(number = ?block.number(), "updated number");
                                 Some(Some(PoolMangerBlocks::NewBlock(block.number())))
                             }
                             CanonStateNotification::Reorg { old, new } => {
@@ -122,9 +129,10 @@ where
                                 }
 
                                 *last_log_write = logs;
-                                this.last_block_number.store(block.number, Ordering::SeqCst);
-                                tracing::info!(?block.number,"updated number");
-                                Some(Some(PoolMangerBlocks::Reorg(block.number, range)))
+                                this.last_block_number
+                                    .store(block.number(), Ordering::SeqCst);
+                                tracing::info!(number = ?block.number(), "updated number");
+                                Some(Some(PoolMangerBlocks::Reorg(block.number(), range)))
                             }
                         };
                         Some((block, notifications))
@@ -152,9 +160,10 @@ where
     }
 }
 
-impl<P> CanonicalStateAdapter<P>
+impl<P, N> CanonicalStateAdapter<P, N>
 where
-    P: Provider + 'static
+    P: Provider + 'static,
+    N: NodePrimitives
 {
     fn validate_filter(&self, filter: &Filter) -> Result<(), PoolManagerError> {
         let last_block = self.last_block_number.load(Ordering::SeqCst);