feat(p2p): preconfirmed p2p event consumer

Adds a task that reads preconfirmed P2P events and updates a watched
value based on those events. For now, since the event itself is a
placeholder, the watched value is a simple counter.

Adds the receiver part of the watched value to the `RpcContext`.

Author: sistemd

crates/pathfinder/src/bin/pathfinder/main.rs

@@ -17,10 +17,12 @@ use pathfinder_ethereum::EthereumClient;
 use pathfinder_gas_price::{L1GasPriceConfig, L1GasPriceProvider};
 #[cfg(feature = "p2p")]
 use pathfinder_lib::consensus::ConsensusTaskHandles;
+#[cfg(feature = "p2p")]
+use pathfinder_lib::preconfirmed::PreconfirmedTaskHandles;
 use pathfinder_lib::state::{sync_gas_prices, L1GasPriceSyncConfig, SyncContext};
 use pathfinder_lib::ConsensusChannels;
 #[cfg(feature = "p2p")]
-use pathfinder_lib::{config, consensus, monitoring, p2p_network, state};
+use pathfinder_lib::{config, consensus, monitoring, p2p_network, preconfirmed, state};
 #[cfg(not(feature = "p2p"))]
 use pathfinder_lib::{config, monitoring, p2p_network, state};
 use pathfinder_rpc::context::{EthContractAddresses, WebsocketContext};
@@ -323,7 +325,7 @@ Hint: This is usually caused by exceeding the file descriptor limit of your syst
     )
     .await;
 
-    let (preconfirmed_p2p_handle, _preconfirmed_p2p_client_and_event_rx) =
+    let (preconfirmed_p2p_handle, preconfirmed_p2p_client_and_event_rx) =
         if config.preconfirmed_p2p.enable {
             p2p_network::preconfirmed::start(
                 chain_id,
@@ -437,13 +439,41 @@ Hint: This is usually caused by exceeding the file descriptor limit of your syst
         )
     };
 
-    let context = if let Some(consensus_info_watch) = consensus_channels
+    #[cfg(feature = "p2p")]
+    let (preconfirmed_p2p_event_processing_handle, preconfirmed_watch_rx) = {
+        let handles = if config.preconfirmed_p2p.enable {
+            match preconfirmed_p2p_client_and_event_rx {
+                Some((event_rx, _)) => preconfirmed::start(event_rx),
+                None => PreconfirmedTaskHandles::pending(),
+            }
+        } else {
+            PreconfirmedTaskHandles::pending()
+        };
+        (
+            handles.preconfirmed_p2p_event_processing_handle,
+            handles.preconfirmed_watch,
+        )
+    };
+
+    #[cfg(not(feature = "p2p"))]
+    let (preconfirmed_p2p_event_processing_handle, preconfirmed_watch_rx) = {
+        let _ = preconfirmed_p2p_client_and_event_rx;
+        (
+            tokio::task::spawn(std::future::pending::<anyhow::Result<()>>()),
+            None,
+        )
+    };
+
+    let context = match consensus_channels
         .as_ref()
         .map(|cc| cc.consensus_info_watch.clone())
     {
-        context.with_consensus_info_watch(consensus_info_watch)
-    } else {
-        context
+        Some(consensus_info_watch) => context.with_consensus_info_watch(consensus_info_watch),
+        None => context,
+    };
+    let context = match preconfirmed_watch_rx {
+        Some(preconfirmed_watch) => context.with_preconfirmed_watch(preconfirmed_watch),
+        None => context,
     };
 
     let default_version = match config.rpc_root_version {
@@ -528,6 +558,7 @@ Hint: This is usually caused by exceeding the file descriptor limit of your syst
         result = consensus_p2p_handle => handle_critical_task_result("Consensus P2P network", result),
         result = preconfirmed_p2p_handle => handle_critical_task_result("Preconfirmed P2P network", result),
         result = consensus_p2p_event_processing_handle => handle_critical_task_result("Consensus P2P event processing", result),
+        result = preconfirmed_p2p_event_processing_handle => handle_critical_task_result("Preconfirmed P2P event processing", result),
         result = consensus_engine_handle => handle_critical_task_result("Consensus engine", result),
         result = http_client_refresh_handle => handle_critical_task_result("HTTP client refresh", result),
         _ = term_signal.recv() => {

crates/pathfinder/src/lib.rs

@@ -7,6 +7,8 @@ pub mod consensus;
 pub mod devnet;
 pub mod monitoring;
 pub mod p2p_network;
+#[cfg(feature = "p2p")]
+pub mod preconfirmed;
 pub mod state;
 pub mod sync;
 pub enum SyncMessageToConsensus {

crates/pathfinder/src/preconfirmed.rs

@@ -0,0 +1,43 @@
+use p2p::preconfirmed::Event;
+use tokio::sync::{mpsc, watch};
+
+pub type PreconfirmedP2PEventProcessingTaskHandle = tokio::task::JoinHandle<anyhow::Result<()>>;
+
+pub struct PreconfirmedTaskHandles {
+    pub preconfirmed_p2p_event_processing_handle: PreconfirmedP2PEventProcessingTaskHandle,
+    // Placeholder watched type, should be the preconfirmed block.
+    pub preconfirmed_watch: Option<watch::Receiver<u32>>,
+}
+
+impl PreconfirmedTaskHandles {
+    pub fn pending() -> Self {
+        Self {
+            preconfirmed_p2p_event_processing_handle: tokio::task::spawn(std::future::pending()),
+            preconfirmed_watch: None,
+        }
+    }
+}
+
+pub fn start(p2p_event_rx: mpsc::UnboundedReceiver<Event>) -> PreconfirmedTaskHandles {
+    let (watch_tx, watch_rx) = watch::channel(0);
+    let jh = util::task::spawn(process_preconfirmed_p2p_events(p2p_event_rx, watch_tx));
+    PreconfirmedTaskHandles {
+        preconfirmed_p2p_event_processing_handle: jh,
+        preconfirmed_watch: Some(watch_rx),
+    }
+}
+
+async fn process_preconfirmed_p2p_events(
+    mut p2p_event_rx: mpsc::UnboundedReceiver<Event>,
+    preconfirmed_watch_tx: watch::Sender<u32>,
+) -> anyhow::Result<()> {
+    while let Some(event) = p2p_event_rx.recv().await {
+        match event.kind {
+            p2p::preconfirmed::EventKind::PreconfirmedTransactionsPlaceholder => {
+                preconfirmed_watch_tx.send_modify(|x| *x += 1)
+            }
+        }
+    }
+
+    Ok(())
+}

crates/rpc/src/context.rs

@@ -99,6 +99,7 @@ pub struct RpcContext {
     pub config: RpcConfig,
     pub native_class_cache: Option<NativeClassCache>,
     pub consensus_info_watch: Option<watch::Receiver<consensus_info::ConsensusInfo>>,
+    pub preconfirmed_watch: Option<watch::Receiver<u32>>,
 }
 
 impl RpcContext {
@@ -144,6 +145,7 @@ impl RpcContext {
             config,
             native_class_cache,
             consensus_info_watch: None,
+            preconfirmed_watch: None,
         }
     }
 
@@ -180,6 +182,13 @@ impl RpcContext {
         }
     }
 
+    pub fn with_preconfirmed_watch(self, preconfirmed_watch: watch::Receiver<u32>) -> Self {
+        Self {
+            preconfirmed_watch: Some(preconfirmed_watch),
+            ..self
+        }
+    }
+
     #[cfg(test)]
     pub fn with_notifications(self, notifications: Notifications) -> Self {
         Self {