[Perf] Dedicated sync streams

cli/src/commands/start.rs

@@ -155,6 +155,11 @@ pub struct Start {
     #[clap(long, requires = "validator")]
     pub bft: Option<SocketAddr>,
 
+    /// Set the IP address and port used for providing block synchronization.
+    /// The default is 0.0.0.0:6130 if not specified.
+    #[clap(long = "sync-listener")]
+    pub sync_listener: Option<SocketAddr>,
+
     /// Specify the host:port address pairs of the peer(s) to connect to (as a comma-separated list).
     ///
     /// These peers will be set as "trusted", which means the node will not disconnect from them when performing peer rotation.
@@ -507,6 +512,13 @@ impl Start {
             self.node = Some(address);
         }
 
+        if self.sync_listener.is_none() {
+            let port = get_devnet_sync_address_for_node(dev).port();
+            let address = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, port));
+            debug!("Setting sync listener address to {address} due to dev={dev}");
+            self.sync_listener = Some(address);
+        }
+
         // If the `norest` flag is not set and the REST IP is not already specified set the REST IP to `3030 + dev`.
         if !self.norest && self.rest.is_none() {
             let port = DEFAULT_REST_PORT + dev;
@@ -714,6 +726,9 @@ impl Start {
         // Parse the node IP or use the default IP/port.
         let node_ip = self.node.unwrap_or(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, DEFAULT_NODE_PORT)));
 
+        // Parse the sync listener.
+        let sync_listener = self.sync_listener.unwrap_or(SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::UNSPECIFIED, 6130)));
+
         // Parse the REST IP.
         let rest_ip = match self.norest {
             true => None,
@@ -846,7 +861,7 @@ impl Start {
 
         // Initialize the node.
         let node = match node_type {
-            NodeType::Validator => Node::new_validator(node_ip, self.bft, rest_ip, self.rest_rps, account, &trusted_peers, &trusted_validators, genesis, cdn, storage_mode, node_data_dir, self.trusted_peers_only, self.auto_db_checkpoints.clone(), dev_txs, self.dev, signal_handler.clone()).await,
+            NodeType::Validator => Node::new_validator(node_ip, sync_listener, self.bft, rest_ip, self.rest_rps, account, &trusted_peers, &trusted_validators, genesis, cdn, storage_mode, node_data_dir, self.trusted_peers_only, self.auto_db_checkpoints.clone(), dev_txs, self.dev, signal_handler.clone()).await,
             NodeType::Prover => Node::new_prover(node_ip, account, &trusted_peers, genesis, node_data_dir, self.trusted_peers_only, self.dev, signal_handler.clone()).await,
             NodeType::Client => Node::new_client(node_ip, rest_ip, self.rest_rps, account, &trusted_peers, genesis, cdn, storage_mode, node_data_dir, self.trusted_peers_only, self.auto_db_checkpoints.clone(), self.dev, signal_handler.clone()).await,
             NodeType::BootstrapClient => Node::new_bootstrap_client(node_ip, account, *genesis.header(), self.dev).await,

cli/src/helpers/dev.rs

@@ -13,7 +13,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use snarkos_node::{bft::MEMORY_POOL_PORT, router::DEFAULT_NODE_PORT};
+use snarkos_node::{bft::MEMORY_POOL_PORT, network::DEFAULT_SYNC_PORT, router::DEFAULT_NODE_PORT};
 
 use snarkvm::{console::network::Network, prelude::PrivateKey};
 
@@ -58,3 +58,8 @@ pub fn get_devnet_gateway_address_for_validator(dev: u16) -> SocketAddr {
 pub fn get_devnet_router_address_for_node(dev: u16) -> SocketAddr {
     SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, DEFAULT_NODE_PORT + dev))
 }
+
+/// Returns the router address a particular devnet validator will list on.
+pub fn get_devnet_sync_address_for_node(dev: u16) -> SocketAddr {
+    SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, DEFAULT_SYNC_PORT + dev))
+}

node/bft/events/Cargo.toml

@@ -42,6 +42,9 @@ features = [ "serde", "rayon" ]
 [dependencies.serde]
 workspace = true
 
+[dependencies.snarkos-node-network]
+workspace = true
+
 [dependencies.snarkos-node-sync-locators]
 workspace = true
 

node/bft/events/src/lib.rs

@@ -24,18 +24,6 @@ pub use batch_propose::BatchPropose;
 mod batch_signature;
 pub use batch_signature::BatchSignature;
 
-mod block_request;
-pub use block_request::BlockRequest;
-
-mod block_response;
-pub use block_response::{BlockResponse, DataBlocks};
-
-mod certificate_request;
-pub use certificate_request::CertificateRequest;
-
-mod certificate_response;
-pub use certificate_response::CertificateResponse;
-
 mod challenge_request;
 pub use challenge_request::ChallengeRequest;
 
@@ -69,17 +57,22 @@ pub use worker_ping::WorkerPing;
 #[cfg(any(test, feature = "test-helpers"))]
 pub mod committee_prop_tests;
 
+pub use snarkos_node_network::{
+    BlockRequest,
+    BlockResponse,
+    CertificateRequest,
+    CertificateResponse,
+    SyncResponse,
+    SyncToken,
+};
+
 use snarkos_node_sync_locators::BlockLocators;
 use snarkvm::{
     console::prelude::{FromBytes, Network, Read, ToBytes, Write, error, io_error},
-    ledger::{
-        block::Block,
-        narwhal::{BatchCertificate, BatchHeader, Data, Transmission, TransmissionID},
-    },
+    ledger::narwhal::{BatchCertificate, BatchHeader, Data, Transmission, TransmissionID},
     prelude::{Address, Field, Signature},
 };
 
-use anyhow::{Result, bail, ensure};
 use indexmap::{IndexMap, IndexSet};
 use serde::{Deserialize, Serialize};
 pub use std::io::{self, Result as IoResult};
@@ -90,6 +83,52 @@ pub trait EventTrait: ToBytes + FromBytes {
     fn name(&self) -> Cow<'static, str>;
 }
 
+// TODO: remove once the compatibility layer for Gateway-based sync is gone
+impl EventTrait for BlockRequest {
+    /// Returns the event name.
+    #[inline]
+    fn name(&self) -> Cow<'static, str> {
+        let start = self.start_height;
+        let end = self.end_height;
+        match start + 1 == end {
+            true => format!("BlockRequest {start}"),
+            false => format!("BlockRequest {start}..{end}"),
+        }
+        .into()
+    }
+}
+
+// TODO: remove once the compatibility layer for Gateway-based sync is gone
+impl<N: Network> EventTrait for BlockResponse<N> {
+    /// Returns the event name.
+    #[inline]
+    fn name(&self) -> Cow<'static, str> {
+        let start = self.request.start_height;
+        let end = self.request.end_height;
+        match start + 1 == end {
+            true => format!("BlockResponse {start}"),
+            false => format!("BlockResponse {start}..{end}"),
+        }
+        .into()
+    }
+}
+
+impl<N: Network> EventTrait for CertificateRequest<N> {
+    /// Returns the event name.
+    #[inline]
+    fn name(&self) -> Cow<'static, str> {
+        "CertificateRequest".into()
+    }
+}
+
+impl<N: Network> EventTrait for CertificateResponse<N> {
+    /// Returns the event name.
+    #[inline]
+    fn name(&self) -> Cow<'static, str> {
+        "CertificateResponse".into()
+    }
+}
+
 #[derive(Clone, Debug, PartialEq, Eq)]
 // TODO (howardwu): For mainnet - Remove this clippy lint. The CertificateResponse should not
 //  be a large enum variant, after removing the versioning.
@@ -111,6 +150,8 @@ pub enum Event<N: Network> {
     ValidatorsRequest(ValidatorsRequest),
     ValidatorsResponse(ValidatorsResponse<N>),
     WorkerPing(WorkerPing<N>),
+    SyncRequest(BlockRequest),
+    SyncResponse(SyncResponse),
 }
 
 impl<N: Network> From<DisconnectReason> for Event<N> {
@@ -143,6 +184,8 @@ impl<N: Network> Event<N> {
             Self::ValidatorsRequest(event) => event.name(),
             Self::ValidatorsResponse(event) => event.name(),
             Self::WorkerPing(event) => event.name(),
+            Self::SyncRequest(event) => event.name(),
+            Self::SyncResponse(event) => event.to_string().into(),
         }
     }
 
@@ -166,6 +209,8 @@ impl<N: Network> Event<N> {
             Self::ValidatorsRequest(..) => 13,
             Self::ValidatorsResponse(..) => 14,
             Self::WorkerPing(..) => 15,
+            Self::SyncRequest(..) => 16,
+            Self::SyncResponse(..) => 17,
         }
     }
 }
@@ -191,6 +236,8 @@ impl<N: Network> ToBytes for Event<N> {
             Self::ValidatorsRequest(event) => event.write_le(writer),
             Self::ValidatorsResponse(event) => event.write_le(writer),
             Self::WorkerPing(event) => event.write_le(writer),
+            Self::SyncRequest(event) => event.write_le(writer),
+            Self::SyncResponse(event) => event.write_le(writer),
         }
     }
 }
@@ -218,7 +265,9 @@ impl<N: Network> FromBytes for Event<N> {
             13 => Self::ValidatorsRequest(ValidatorsRequest::read_le(&mut reader)?),
             14 => Self::ValidatorsResponse(ValidatorsResponse::read_le(&mut reader)?),
             15 => Self::WorkerPing(WorkerPing::read_le(&mut reader)?),
-            16.. => return Err(error(format!("Unknown event ID {id}"))),
+            16 => Self::SyncRequest(BlockRequest::read_le(&mut reader)?),
+            17 => Self::SyncResponse(SyncResponse::read_le(&mut reader)?),
+            18.. => return Err(error(format!("Unknown event ID {id}"))),
         };
 
         // Ensure that there are no "dangling" bytes.

node/bft/ledger-service/src/ledger.rs

@@ -126,7 +126,7 @@ impl<'a, N: Network, C: ConsensusStorage<N>> LedgerUpdateService<N> for LedgerUp
             metrics::update_block_metrics(block);
         }
 
-        tracing::info!("Advanced to block {} at round {} - {}\n", block.height(), block.round(), block.hash());
+        tracing::info!("Advanced to block {} at round {} - {}", block.height(), block.round(), block.hash());
         Ok(())
     }
 }

node/bft/src/bft.rs

@@ -86,6 +86,7 @@ impl<N: Network> BFT<N> {
         storage: Storage<N>,
         ledger: Arc<dyn LedgerService<N>>,
         block_sync: Arc<BlockSync<N>>,
+        sync_listener: SocketAddr,
         ip: Option<SocketAddr>,
         trusted_validators: &[SocketAddr],
         trusted_peers_only: bool,
@@ -98,6 +99,7 @@ impl<N: Network> BFT<N> {
                 storage,
                 ledger,
                 block_sync,
+                sync_listener,
                 ip,
                 trusted_validators,
                 trusted_peers_only,