feat: add census with peer scoring to downloader

Author: ogenev

Cargo.lock

@@ -123,7 +123,7 @@ impl QueryTrace {
     }
 
     /// Returns milliseconds since the time provided.
-    fn timestamp_millis_u64(since: u64) -> u64 {
+    pub fn timestamp_millis_u64(since: u64) -> u64 {
         // Convert `since` (milliseconds) to a `SystemTime`
         let since_time = UNIX_EPOCH + Duration::from_millis(since);
 

ethportal-api/src/types/query_trace.rs

@@ -11,7 +11,7 @@ use thiserror::Error;
 use tokio::task::JoinHandle;
 use tracing::{error, info, Instrument};
 
-use crate::cli::BridgeConfig;
+use crate::cli::ClientType;
 
 mod network;
 mod peer;
@@ -51,11 +51,30 @@ impl Census {
     const SUPPORTED_SUBNETWORKS: [Subnetwork; 3] =
         [Subnetwork::Beacon, Subnetwork::History, Subnetwork::State];
 
-    pub fn new(client: HttpClient, bridge_config: &BridgeConfig) -> Self {
+    pub fn new(
+        client: HttpClient,
+        enr_offer_limit: usize,
+        filter_clients: Vec<ClientType>,
+    ) -> Self {
         Self {
-            history: Network::new(client.clone(), Subnetwork::History, bridge_config),
-            state: Network::new(client.clone(), Subnetwork::State, bridge_config),
-            beacon: Network::new(client.clone(), Subnetwork::Beacon, bridge_config),
+            history: Network::new(
+                client.clone(),
+                Subnetwork::History,
+                enr_offer_limit,
+                filter_clients.clone(),
+            ),
+            state: Network::new(
+                client.clone(),
+                Subnetwork::State,
+                enr_offer_limit,
+                filter_clients.clone(),
+            ),
+            beacon: Network::new(
+                client.clone(),
+                Subnetwork::Beacon,
+                enr_offer_limit,
+                filter_clients,
+            ),
             initialized: false,
         }
     }

portal-bridge/src/census/mod.rs

@@ -20,10 +20,7 @@ use super::{
     peers::Peers,
     scoring::{AdditiveWeight, PeerSelector},
 };
-use crate::{
-    census::CensusError,
-    cli::{BridgeConfig, ClientType},
-};
+use crate::{census::CensusError, cli::ClientType};
 
 /// The result of the liveness check.
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -78,7 +75,12 @@ pub(super) struct Network {
 }
 
 impl Network {
-    pub fn new(client: HttpClient, subnetwork: Subnetwork, bridge_config: &BridgeConfig) -> Self {
+    pub fn new(
+        client: HttpClient,
+        subnetwork: Subnetwork,
+        enr_offer_limit: usize,
+        filter_clients: Vec<ClientType>,
+    ) -> Self {
         if !matches!(
             subnetwork,
             Subnetwork::History | Subnetwork::Beacon | Subnetwork::State
@@ -89,11 +91,11 @@ impl Network {
         Self {
             peers: Peers::new(PeerSelector::new(
                 AdditiveWeight::default(),
-                bridge_config.enr_offer_limit,
+                enr_offer_limit,
             )),
             client,
             subnetwork,
-            filter_clients: bridge_config.filter_clients.to_vec(),
+            filter_clients,
         }
     }
 

portal-bridge/src/census/network.rs

@@ -49,7 +49,11 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         .contains(&Subnetwork::State)
     {
         // Create and initialize the census to acquire critical view of network before gossiping
-        let mut census = Census::new(portal_client.clone(), &bridge_config);
+        let mut census = Census::new(
+            portal_client.clone(),
+            bridge_config.enr_offer_limit,
+            bridge_config.filter_clients,
+        );
         census_handle = Some(census.init([Subnetwork::State]).await?);
 
         let state_bridge = StateBridge::new(

portal-bridge/src/main.rs

@@ -26,6 +26,7 @@ tree_hash.workspace = true
 trin-storage.workspace = true
 trin-validation.workspace = true
 utp-rs.workspace = true
+portal-bridge.workspace = true
 futures = "0.3.31"
 
 [dev-dependencies]

trin-history/Cargo.toml

@@ -7,34 +7,49 @@ use std::fs::File;
 use std::{
     io::{self, BufRead},
     path::Path,
+    time::Duration,
 };
 
 use anyhow::anyhow;
 use ethportal_api::{
-    utils::bytes::hex_decode, BlockBodyKey, BlockReceiptsKey, ContentValue, HistoryContentKey,
-    HistoryContentValue,
+    jsonrpsee::http_client::{HttpClient, HttpClientBuilder},
+    types::{cli::DEFAULT_WEB3_HTTP_ADDRESS, network::Subnetwork, query_trace::QueryTrace},
+    utils::bytes::hex_decode,
+    BlockBodyKey, BlockReceiptsKey, ContentValue, HistoryContentKey, HistoryContentValue,
 };
 use futures::{channel::oneshot, future::join_all};
-use portalnet::overlay::command::OverlayCommand;
+use portal_bridge::census::Census;
+use portalnet::overlay::{command::OverlayCommand, config::FindContentConfig};
 use tokio::sync::mpsc::UnboundedSender;
 use tracing::{error, info, warn};
 
 /// The number of blocks to download in a single batch.
-const BATCH_SIZE: usize = 40;
+const BATCH_SIZE: usize = 3;
 /// The path to the CSV file with block numbers and block hashes.
 const CSV_PATH: &str = "ethereum_blocks_14000000_merge.csv";
 
 #[derive(Clone)]
 pub struct Downloader {
+    pub census: Census,
     pub overlay_tx: UnboundedSender<OverlayCommand<HistoryContentKey>>,
 }
 
 impl Downloader {
     pub fn new(overlay_tx: UnboundedSender<OverlayCommand<HistoryContentKey>>) -> Self {
-        Self { overlay_tx }
+        let http_client: HttpClient = HttpClientBuilder::default()
+            // increase default timeout to allow for trace_gossip requests that can take a long
+            // time
+            .request_timeout(Duration::from_secs(120))
+            .build(DEFAULT_WEB3_HTTP_ADDRESS)
+            .map_err(|e| e.to_string())
+            .expect("Failed to build http client");
+
+        // BUild hhtp client binded to the current node web3rpc
+        let census = Census::new(http_client, 0, vec![]);
+        Self { overlay_tx, census }
     }
 
-    pub async fn start(self) -> io::Result<()> {
+    pub async fn start(mut self) -> io::Result<()> {
         // set the csv path to a file in the root trin-history directory
         info!("Opening CSV file");
         let csv_path = Path::new(CSV_PATH);
@@ -47,6 +62,14 @@ impl Downloader {
         // skip the header of the csv file
         let lines = &lines[1..];
         let blocks: Vec<(u64, String)> = lines.iter().map(|line| parse_line(line)).collect();
+        // Initialize the census with the history subnetwork
+        let _ = Some(
+            self.census
+                .init([Subnetwork::History])
+                .await
+                .expect("Failed to initialize Census"),
+        );
+
         info!("Processing blocks");
         let batches = blocks.chunks(BATCH_SIZE);
 
@@ -91,7 +114,10 @@ impl Downloader {
         let overlay_command = OverlayCommand::FindContentQuery {
             target: content_key.clone(),
             callback: tx,
-            config: Default::default(),
+            config: FindContentConfig {
+                is_trace: true,
+                ..Default::default()
+            },
         };
 
         if let Err(err) = self.overlay_tx.send(overlay_command) {
@@ -104,7 +130,14 @@ impl Downloader {
             Ok(result) => match result {
                 Ok(result) => {
                     HistoryContentValue::decode(&content_key, &result.0)?;
-                    info!(block_number = block_number, "Downloaded content for block");
+                    let duration_ms = QueryTrace::timestamp_millis_u64(
+                        result.2.expect("QueryTrace not found").started_at_ms,
+                    );
+                    info!(
+                        block_number = block_number,
+                        query_duration = duration_ms,
+                        "Downloaded content for block"
+                    );
                     Ok(())
                 }
                 Err(err) => {