Merge pull request #4208 from ProvableHQ/feat/sync-log-fork-source

vicsn · web-flow · commit aa0fcf35506c · 2026-04-09T11:14:06.000+02:00
[Feature] Log where a fork originated (Router or Gateway)
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/node/bft/examples/simple_node.rs b/node/bft/examples/simple_node.rs
@@ -28,6 +28,7 @@ use snarkos_node_bft::{
 };
 use snarkos_node_bft_ledger_service::TranslucentLedgerService;
 use snarkos_node_bft_storage_service::BFTMemoryService;
+use snarkos_node_network::ConnectionMode;
 use snarkos_node_sync::BlockSync;
 use snarkos_utilities::{NodeDataDir, SimpleStoppable};
 
@@ -149,7 +150,7 @@ pub async fn start_bft(
     // Initialize the consensus receiver handler.
     consensus_handler(consensus_receiver);
     // Initialize the BFT instance.
-    let block_sync = Arc::new(BlockSync::new(ledger.clone()));
+    let block_sync = Arc::new(BlockSync::new(ledger.clone(), ConnectionMode::Gateway));
     let mut bft = BFT::<CurrentNetwork>::new(
         account,
         storage,
@@ -199,7 +200,7 @@ pub async fn start_primary(
     let trusted_validators = trusted_validators(node_id, num_nodes, peers);
     let trusted_peers_only = false;
     // Initialize the primary instance.
-    let block_sync = Arc::new(BlockSync::new(ledger.clone()));
+    let block_sync = Arc::new(BlockSync::new(ledger.clone(), ConnectionMode::Gateway));
     let primary = Primary::<CurrentNetwork>::new(
         account,
         storage,
diff --git a/node/bft/src/bft.rs b/node/bft/src/bft.rs
@@ -890,6 +890,7 @@ mod tests {
     use snarkos_account::Account;
     use snarkos_node_bft_ledger_service::{LedgerService, MockLedgerService};
     use snarkos_node_bft_storage_service::BFTMemoryService;
+    use snarkos_node_network::ConnectionMode;
     use snarkos_node_sync::BlockSync;
     use snarkos_utilities::NodeDataDir;
 
@@ -948,7 +949,7 @@ mod tests {
         ledger: Arc<MockLedgerService<CurrentNetwork>>,
     ) -> anyhow::Result<BFT<CurrentNetwork>> {
         // Create the block synchronization logic.
-        let block_sync = Arc::new(BlockSync::new(ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(ledger.clone(), ConnectionMode::Gateway));
         // Initialize the BFT.
         BFT::new(
             account.clone(),
diff --git a/node/bft/src/primary.rs b/node/bft/src/primary.rs
@@ -45,6 +45,8 @@ use crate::{
 use snarkos_account::Account;
 use snarkos_node_bft_events::PrimaryPing;
 use snarkos_node_bft_ledger_service::LedgerService;
+#[cfg(test)]
+use snarkos_node_network::ConnectionMode;
 use snarkos_node_network::PeerPoolHandling;
 use snarkos_node_sync::{BlockSync, DUMMY_SELF_IP, Ping};
 use snarkos_utilities::{CallbackHandle, NodeDataDir};
@@ -2082,7 +2084,7 @@ mod tests {
 
         // Initialize the primary.
         let account = accounts[account_index].1.clone();
-        let block_sync = Arc::new(BlockSync::new(ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(ledger.clone(), ConnectionMode::Gateway));
         let primary =
             Primary::new(account, storage, ledger, block_sync, None, &[], false, NodeDataDir::new_test(None), None)
                 .unwrap();
diff --git a/node/bft/src/sync/mod.rs b/node/bft/src/sync/mod.rs
@@ -1099,6 +1099,7 @@ mod tests {
     use crate::{BFT, helpers::now, ledger_service::CoreLedgerService, storage_service::BFTMemoryService};
 
     use snarkos_account::Account;
+    use snarkos_node_network::ConnectionMode;
     use snarkos_node_sync::BlockSync;
     use snarkos_utilities::{NodeDataDir, SimpleStoppable};
 
@@ -1356,7 +1357,7 @@ mod tests {
         )
         .unwrap();
 
-        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone(), ConnectionMode::Gateway));
         let sync = Sync::new(gateway.clone(), syncing_storage.clone(), syncing_ledger.clone(), block_sync.clone());
 
         let syncing_bft = BFT::new(
@@ -1425,7 +1426,7 @@ mod tests {
         )
         .unwrap();
 
-        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone(), ConnectionMode::Gateway));
         let sync = Sync::new(gateway.clone(), syncing_storage.clone(), syncing_ledger.clone(), block_sync.clone());
 
         let syncing_bft = BFT::new(
@@ -1516,7 +1517,7 @@ mod tests {
         )
         .unwrap();
 
-        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone(), ConnectionMode::Gateway));
         let sync = Sync::new(gateway.clone(), syncing_storage.clone(), syncing_ledger.clone(), block_sync.clone());
 
         let syncing_bft = BFT::new(
@@ -1601,7 +1602,7 @@ mod tests {
         )
         .unwrap();
 
-        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone(), ConnectionMode::Gateway));
         let sync = Sync::new(gateway.clone(), syncing_storage.clone(), syncing_ledger.clone(), block_sync.clone());
 
         let syncing_bft = BFT::new(
@@ -1683,7 +1684,7 @@ mod tests {
         )
         .unwrap();
 
-        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone()));
+        let block_sync = Arc::new(BlockSync::new(syncing_ledger.clone(), ConnectionMode::Gateway));
         let sync = Sync::new(gateway.clone(), syncing_storage.clone(), syncing_ledger.clone(), block_sync.clone());
 
         let syncing_bft = BFT::new(
diff --git a/node/bft/tests/common/primary.rs b/node/bft/tests/common/primary.rs
@@ -28,7 +28,7 @@ use snarkos_node_bft::{
     helpers::{PrimarySender, Storage, init_primary_channels},
 };
 use snarkos_node_bft_storage_service::BFTMemoryService;
-use snarkos_node_network::PeerPoolHandling;
+use snarkos_node_network::{ConnectionMode, PeerPoolHandling};
 use snarkos_node_sync::BlockSync;
 use snarkos_utilities::{NodeDataDir, SimpleStoppable};
 
@@ -171,7 +171,7 @@ impl TestNetwork {
             )
             .unwrap();
             // Initialize the block synchronization logic.
-            let block_sync = Arc::new(BlockSync::new(ledger.clone()));
+            let block_sync = Arc::new(BlockSync::new(ledger.clone(), ConnectionMode::Gateway));
             let (primary, bft) = if config.bft {
                 let bft = BFT::<CurrentNetwork>::new(
                     account,
diff --git a/node/network/src/peer.rs b/node/network/src/peer.rs
@@ -17,7 +17,7 @@ use crate::NodeType;
 use snarkvm::prelude::{Address, Network};
 use tracing::*;
 
-use std::{net::SocketAddr, time::Instant};
+use std::{fmt, net::SocketAddr, time::Instant};
 
 /// A peer of any connection status.
 #[derive(Clone, Debug)]
@@ -89,6 +89,15 @@ pub enum ConnectionMode {
     Router,
 }
 
+impl fmt::Display for ConnectionMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            ConnectionMode::Gateway => write!(f, "Gateway"),
+            ConnectionMode::Router => write!(f, "Router"),
+        }
+    }
+}
+
 impl<N: Network> Peer<N> {
     /// Create a candidate peer.
     pub const fn new_candidate(listener_addr: SocketAddr, trusted: bool) -> Self {
diff --git a/node/src/client/mod.rs b/node/src/client/mod.rs
@@ -22,7 +22,7 @@ use crate::{
 };
 
 use snarkos_account::Account;
-use snarkos_node_network::NodeType;
+use snarkos_node_network::{ConnectionMode, NodeType};
 use snarkos_node_rest::Rest;
 use snarkos_node_router::{
     Heartbeat,
@@ -172,7 +172,7 @@ impl<N: Network, C: ConsensusStorage<N>> Client<N, C> {
         .await?;
 
         // Initialize the sync module.
-        let sync = Arc::new(BlockSync::new(ledger_service.clone()));
+        let sync = Arc::new(BlockSync::new(ledger_service.clone(), ConnectionMode::Router));
 
         // Set up the ping logic.
         let locators = sync.get_block_locators()?;
diff --git a/node/src/prover/mod.rs b/node/src/prover/mod.rs
@@ -22,7 +22,7 @@ use crate::{
 };
 
 use snarkos_account::Account;
-use snarkos_node_network::{NodeType, PeerPoolHandling};
+use snarkos_node_network::{ConnectionMode, NodeType, PeerPoolHandling};
 use snarkos_node_router::{
     Heartbeat,
     Inbound,
@@ -125,7 +125,7 @@ impl<N: Network, C: ConsensusStorage<N>> Prover<N, C> {
         .await?;
 
         // Initialize the sync module.
-        let sync = BlockSync::new(ledger_service.clone());
+        let sync = BlockSync::new(ledger_service.clone(), ConnectionMode::Router);
 
         // Set up the ping logic.
         let ping = Arc::new(Ping::new_nosync(router.clone()));
diff --git a/node/src/validator/mod.rs b/node/src/validator/mod.rs
@@ -21,7 +21,7 @@ use snarkos_account::Account;
 use snarkos_node_bft::{ledger_service::CoreLedgerService, spawn_blocking};
 use snarkos_node_cdn::CdnBlockSync;
 use snarkos_node_consensus::Consensus;
-use snarkos_node_network::{NodeType, PeerPoolHandling};
+use snarkos_node_network::{ConnectionMode, NodeType, PeerPoolHandling};
 use snarkos_node_rest::Rest;
 use snarkos_node_router::{
     Heartbeat,
@@ -121,7 +121,7 @@ impl<N: Network, C: ConsensusStorage<N>> Validator<N, C> {
         .await?;
 
         // Initialize the block synchronization logic.
-        let sync = Arc::new(BlockSync::new(ledger_service.clone()));
+        let sync = Arc::new(BlockSync::new(ledger_service.clone(), ConnectionMode::Gateway));
         let locators = sync.get_block_locators()?;
         let ping = Arc::new(Ping::new(router.clone(), locators));
 
diff --git a/node/sync/Cargo.toml b/node/sync/Cargo.toml
@@ -70,6 +70,9 @@ features = [ "ledger-write" ]
 workspace = true
 optional = true
 
+[dependencies.snarkos-node-network]
+workspace = true
+
 [dependencies.snarkos-node-router]
 workspace = true
 
diff --git a/node/sync/src/block_sync.rs b/node/sync/src/block_sync.rs
@@ -18,6 +18,7 @@ use crate::{
     locators::BlockLocators,
 };
 use snarkos_node_bft_ledger_service::{BeginLedgerUpdateError, LedgerService};
+use snarkos_node_network::ConnectionMode;
 use snarkos_node_router::messages::DataBlocks;
 use snarkos_node_sync_communication_service::CommunicationService;
 use snarkos_node_sync_locators::{CHECKPOINT_INTERVAL, NUM_RECENT_BLOCKS};
@@ -193,6 +194,9 @@ pub struct BlockSync<N: Network> {
     /// The ledger.
     ledger: Arc<dyn LedgerService<N>>,
 
+    /// The connection mode of this node (Gateway for validators, Router for clients/provers).
+    connection_mode: ConnectionMode,
+
     /// The map of peer IP to their block locators.
     /// The block locators are consistent with the ledger and every other peer's block locators.
     locators: RwLock<HashMap<SocketAddr, BlockLocators<N>>>,
@@ -232,12 +236,13 @@ pub struct BlockSync<N: Network> {
 
 impl<N: Network> BlockSync<N> {
     /// Initializes a new block sync module.
-    pub fn new(ledger: Arc<dyn LedgerService<N>>) -> Self {
+    pub fn new(ledger: Arc<dyn LedgerService<N>>, connection_mode: ConnectionMode) -> Self {
         // Make sync state aware of the blocks that already exist on disk at startup.
         let sync_state = SyncState::new_with_height(ledger.latest_block_height());
 
         Self {
             ledger,
+            connection_mode,
             sync_state: RwLock::new(sync_state),
             peer_notify: Default::default(),
             response_notify: Default::default(),
@@ -780,6 +785,7 @@ impl<N: Network> BlockSync<N> {
     /// This function does **not** check
     /// that the block locators are consistent with the peer's previous block locators or other peers' block locators.
     pub fn update_peer_locators(&self, peer_ip: SocketAddr, locators: &BlockLocators<N>) -> Result<()> {
+        let connection_mode = self.connection_mode;
         // -- First, update the locators entry for the given peer IP. --
         // We perform this update atomically, and drop the lock as soon as we are done with the update.
         match self.locators.write().entry(peer_ip) {
@@ -813,7 +819,9 @@ impl<N: Network> BlockSync<N> {
                     match ledger_hash == hash {
                         true => ancestor = height,
                         false => {
-                            warn!("Detected fork between this node and peer \"{peer_ip}\" at height {height}");
+                            warn!(
+                                "[{connection_mode}] Detected fork between this node and peer \"{peer_ip}\" at height {height}"
+                            );
                             break;
                         }
                     }
@@ -841,7 +849,7 @@ impl<N: Network> BlockSync<N> {
                             true => ancestor = height,
                             false => {
                                 debug!(
-                                    "Detected fork between peers \"{other_ip}\" and \"{peer_ip}\" at height {height}"
+                                    "[{connection_mode}] Detected fork between peers \"{other_ip}\" and \"{peer_ip}\" at height {height}"
                                 );
                                 break;
                             }
@@ -1660,7 +1668,7 @@ mod tests {
 
     /// Returns the sync pool, with the ledger initialized to the given height.
     fn sample_sync_at_height(height: u32) -> BlockSync<CurrentNetwork> {
-        BlockSync::<CurrentNetwork>::new(Arc::new(sample_ledger_service(height)))
+        BlockSync::<CurrentNetwork>::new(Arc::new(sample_ledger_service(height)), ConnectionMode::Router)
     }
 
     /// Returns a vector of randomly sampled block heights in [0, max_height].
@@ -1686,6 +1694,7 @@ mod tests {
             peer_notify: Notify::new(),
             response_notify: Default::default(),
             ledger: Arc::new(sample_ledger_service(height)),
+            connection_mode: sync.connection_mode,
             locators: RwLock::new(sync.locators.read().clone()),
             common_ancestors: RwLock::new(sync.common_ancestors.read().clone()),
             requests: RwLock::new(sync.requests.read().clone()),
