feat(sync): dynamically fetch block locators at sync

Author: kaimast

Cargo.lock

@@ -0,0 +1,78 @@
+// Copyright (c) 2019-2025 Provable Inc.
+// This file is part of the snarkOS library.
+
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at:
+
+// http://www.apache.org/licenses/LICENSE-2.0
+
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::{BlockLocators, EventTrait, FromBytes, IoResult, Network, ToBytes};
+
+use std::{
+    borrow::Cow,
+    io::{Read, Write},
+};
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct BlockLocatorsRequest {
+    pub start_height: u32,
+    pub end_height: u32,
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct BlockLocatorsResponse<N: Network> {
+    pub locators: BlockLocators<N>,
+}
+
+impl EventTrait for BlockLocatorsRequest {
+    /// Returns the event name.
+    #[inline]
+    fn name(&self) -> Cow<'static, str> {
+        "BlockLocatorsRequest".into()
+    }
+}
+
+impl FromBytes for BlockLocatorsRequest {
+    fn read_le<R: Read>(mut reader: R) -> IoResult<Self> {
+        let start_height = u32::read_le(&mut reader)?;
+        let end_height = u32::read_le(&mut reader)?;
+
+        Ok(Self { start_height, end_height })
+    }
+}
+
+impl ToBytes for BlockLocatorsRequest {
+    fn write_le<W: Write>(&self, mut writer: W) -> IoResult<()> {
+        self.start_height.write_le(&mut writer)?;
+        self.end_height.write_le(&mut writer)?;
+        Ok(())
+    }
+}
+
+impl<N: Network> EventTrait for BlockLocatorsResponse<N> {
+    /// Returns the event name.
+    #[inline]
+    fn name(&self) -> Cow<'static, str> {
+        "BlockLocatorsResponse".into()
+    }
+}
+
+impl<N: Network> FromBytes for BlockLocatorsResponse<N> {
+    fn read_le<R: Read>(mut reader: R) -> IoResult<Self> {
+        let locators = BlockLocators::read_le(&mut reader)?;
+        Ok(Self { locators })
+    }
+}
+
+impl<N: Network> ToBytes for BlockLocatorsResponse<N> {
+    fn write_le<W: Write>(&self, mut writer: W) -> IoResult<()> {
+        self.locators.write_le(&mut writer)
+    }
+}

node/bft/events/src/block_locators.rs

@@ -66,6 +66,9 @@ pub use validators_response::ValidatorsResponse;
 mod worker_ping;
 pub use worker_ping::WorkerPing;
 
+mod block_locators;
+pub use block_locators::{BlockLocatorsRequest, BlockLocatorsResponse};
+
 use snarkos_node_sync_locators::BlockLocators;
 use snarkvm::{
     console::prelude::{FromBytes, Network, Read, ToBytes, Write, error},
@@ -90,7 +93,6 @@ pub trait EventTrait: ToBytes + FromBytes {
 #[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.
-#[allow(clippy::large_enum_variant)]
 pub enum Event<N: Network> {
     BatchPropose(BatchPropose<N>),
     BatchSignature(BatchSignature<N>),
@@ -108,6 +110,8 @@ pub enum Event<N: Network> {
     ValidatorsRequest(ValidatorsRequest),
     ValidatorsResponse(ValidatorsResponse<N>),
     WorkerPing(WorkerPing<N>),
+    BlockLocatorsRequest(BlockLocatorsRequest),
+    BlockLocatorsResponse(BlockLocatorsResponse<N>),
 }
 
 impl<N: Network> From<DisconnectReason> for Event<N> {
@@ -140,6 +144,8 @@ impl<N: Network> Event<N> {
             Self::ValidatorsRequest(event) => event.name(),
             Self::ValidatorsResponse(event) => event.name(),
             Self::WorkerPing(event) => event.name(),
+            Self::BlockLocatorsRequest(event) => event.name(),
+            Self::BlockLocatorsResponse(event) => event.name(),
         }
     }
 
@@ -163,6 +169,8 @@ impl<N: Network> Event<N> {
             Self::ValidatorsRequest(..) => 13,
             Self::ValidatorsResponse(..) => 14,
             Self::WorkerPing(..) => 15,
+            Self::BlockLocatorsRequest(..) => 16,
+            Self::BlockLocatorsResponse(..) => 17,
         }
     }
 }
@@ -188,6 +196,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::BlockLocatorsRequest(event) => event.write_le(writer),
+            Self::BlockLocatorsResponse(event) => event.write_le(writer),
         }
     }
 }
@@ -215,7 +225,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::BlockLocatorsRequest(BlockLocatorsRequest::read_le(&mut reader)?),
+            17 => Self::BlockLocatorsResponse(BlockLocatorsResponse::read_le(&mut reader)?),
+            18.. => return Err(error(format!("Unknown event ID {id}"))),
         };
 
         // Ensure that there are no "dangling" bytes.

node/bft/events/src/lib.rs

@@ -18,25 +18,21 @@ use super::*;
 #[derive(Clone, Debug, PartialEq, Eq)]
 pub struct PrimaryPing<N: Network> {
     pub version: u32,
-    pub block_locators: BlockLocators<N>,
+    pub current_block_height: u32,
     pub primary_certificate: Data<BatchCertificate<N>>,
 }
 
 impl<N: Network> PrimaryPing<N> {
     /// Initializes a new ping event.
-    pub const fn new(
-        version: u32,
-        block_locators: BlockLocators<N>,
-        primary_certificate: Data<BatchCertificate<N>>,
-    ) -> Self {
-        Self { version, block_locators, primary_certificate }
+    pub const fn new(version: u32, current_block_height: u32, primary_certificate: Data<BatchCertificate<N>>) -> Self {
+        Self { version, current_block_height, primary_certificate }
     }
 }
 
-impl<N: Network> From<(u32, BlockLocators<N>, BatchCertificate<N>)> for PrimaryPing<N> {
+impl<N: Network> From<(u32, u32, BatchCertificate<N>)> for PrimaryPing<N> {
     /// Initializes a new ping event.
-    fn from((version, block_locators, primary_certificate): (u32, BlockLocators<N>, BatchCertificate<N>)) -> Self {
-        Self::new(version, block_locators, Data::Object(primary_certificate))
+    fn from((version, current_block_height, primary_certificate): (u32, u32, BatchCertificate<N>)) -> Self {
+        Self::new(version, current_block_height, Data::Object(primary_certificate))
     }
 }
 
@@ -50,11 +46,8 @@ impl<N: Network> EventTrait for PrimaryPing<N> {
 
 impl<N: Network> ToBytes for PrimaryPing<N> {
     fn write_le<W: Write>(&self, mut writer: W) -> IoResult<()> {
-        // Write the version.
         self.version.write_le(&mut writer)?;
-        // Write the block locators.
-        self.block_locators.write_le(&mut writer)?;
-        // Write the primary certificate.
+        self.current_block_height.write_le(&mut writer)?;
         self.primary_certificate.write_le(&mut writer)?;
 
         Ok(())
@@ -66,19 +59,18 @@ impl<N: Network> FromBytes for PrimaryPing<N> {
         // Read the version.
         let version = u32::read_le(&mut reader)?;
         // Read the block locators.
-        let block_locators = BlockLocators::read_le(&mut reader)?;
+        let current_block_height = u32::read_le(&mut reader)?;
         // Read the primary certificate.
         let primary_certificate = Data::read_le(&mut reader)?;
 
         // Return the ping event.
-        Ok(Self::new(version, block_locators, primary_certificate))
+        Ok(Self::new(version, current_block_height, primary_certificate))
     }
 }
 
 #[cfg(test)]
 pub mod prop_tests {
     use crate::{PrimaryPing, certificate_response::prop_tests::any_batch_certificate};
-    use snarkos_node_sync_locators::{BlockLocators, test_helpers::sample_block_locators};
     use snarkvm::utilities::{FromBytes, ToBytes};
 
     use bytes::{Buf, BufMut, BytesMut};
@@ -87,14 +79,10 @@ pub mod prop_tests {
 
     type CurrentNetwork = snarkvm::prelude::MainnetV0;
 
-    pub fn any_block_locators() -> BoxedStrategy<BlockLocators<CurrentNetwork>> {
-        any::<u32>().prop_map(sample_block_locators).boxed()
-    }
-
     pub fn any_primary_ping() -> BoxedStrategy<PrimaryPing<CurrentNetwork>> {
-        (any::<u32>(), any_block_locators(), any_batch_certificate())
-            .prop_map(|(version, block_locators, batch_certificate)| {
-                PrimaryPing::from((version, block_locators, batch_certificate.clone()))
+        (any::<u32>(), any::<u32>(), any_batch_certificate())
+            .prop_map(|(version, current_block_height, batch_certificate)| {
+                PrimaryPing::from((version, current_block_height, batch_certificate.clone()))
             })
             .boxed()
     }
@@ -105,7 +93,7 @@ pub mod prop_tests {
         primary_ping.write_le(&mut bytes).unwrap();
         let decoded = PrimaryPing::<CurrentNetwork>::read_le(&mut bytes.into_inner().reader()).unwrap();
         assert_eq!(primary_ping.version, decoded.version);
-        assert_eq!(primary_ping.block_locators, decoded.block_locators);
+        assert_eq!(primary_ping.current_block_height, decoded.current_block_height);
         assert_eq!(
             primary_ping.primary_certificate.deserialize_blocking().unwrap(),
             decoded.primary_certificate.deserialize_blocking().unwrap(),

node/bft/events/src/primary_ping.rs

@@ -26,6 +26,8 @@ use crate::{
 };
 use snarkos_account::Account;
 use snarkos_node_bft_events::{
+    BlockLocatorsRequest,
+    BlockLocatorsResponse,
     BlockRequest,
     BlockResponse,
     CertificateRequest,
@@ -62,6 +64,7 @@ use snarkvm::{
     prelude::{Address, Field},
 };
 
+use anyhow::anyhow;
 use colored::Colorize;
 use futures::SinkExt;
 use indexmap::{IndexMap, IndexSet};
@@ -267,6 +270,11 @@ impl<N: Network> CommunicationService for Gateway<N> {
         Event::BlockRequest(BlockRequest { start_height, end_height })
     }
 
+    /// Prepare a block locators requets to be sent
+    fn prepare_block_locators_request(start_height: u32, end_height: u32) -> Self::Message {
+        Event::BlockLocatorsRequest(BlockLocatorsRequest { start_height, end_height })
+    }
+
     /// Sends the given message to specified peer.
     ///
     /// This function returns as soon as the message is queued to be sent,
@@ -730,7 +738,7 @@ impl<N: Network> Gateway<N> {
                 bail!("{CONTEXT} {:?}", disconnect.reason)
             }
             Event::PrimaryPing(ping) => {
-                let PrimaryPing { version, block_locators, primary_certificate } = ping;
+                let PrimaryPing { version, current_block_height, primary_certificate } = ping;
 
                 // Ensure the event version is not outdated.
                 if version < Event::<N>::VERSION {
@@ -740,7 +748,7 @@ impl<N: Network> Gateway<N> {
                 // Update the peer locators. Except for some tests, there is always a sync sender.
                 if let Some(sync_sender) = self.sync_sender.get() {
                     // Check the block locators are valid, and update the validators in the sync module.
-                    if let Err(error) = sync_sender.update_peer_locators(peer_ip, block_locators).await {
+                    if let Err(error) = sync_sender.update_peer_block_height(peer_ip, current_block_height).await {
                         bail!("Validator '{peer_ip}' sent invalid block locators - {error}");
                     }
                 }
@@ -882,6 +890,29 @@ impl<N: Network> Gateway<N> {
                 }
                 Ok(())
             }
+            Event::BlockLocatorsRequest(BlockLocatorsRequest { start_height, end_height }) => {
+                ensure!(start_height < end_height, "Invalid block locators range");
+
+                let Some(sync_sender) = self.sync_sender.get() else {
+                    bail!("Cannot process block locators request: no sync sender");
+                };
+
+                let locators = sync_sender.get_block_locators(start_height, end_height).await?;
+                let event = Event::BlockLocatorsResponse(BlockLocatorsResponse { locators });
+
+                let Some(result) = Transport::send(self, peer_ip, event).await else {
+                    bail!("Failed to send block locator response to peer {peer_ip}");
+                };
+
+                result.await?.map_err(|err| anyhow!("Send failed: {err}"))
+            }
+            Event::BlockLocatorsResponse(BlockLocatorsResponse { locators }) => {
+                if let Some(sync_sender) = self.sync_sender.get() {
+                    sync_sender.update_peer_block_locators(peer_ip, locators).await
+                } else {
+                    bail!("Cannot update peer block locators: no sync sender");
+                }
+            }
         }
     }