Persist name records for Peer Discovery

Author: boicotinho

Cargo.lock

@@ -19,6 +19,7 @@ ExecStart=/usr/local/bin/monad-node \
     --statesync-sq-thread-cpu 8 \
     --record-metrics-interval-seconds 1 \
     --validators-path /home/monad/monad-bft/config/validators/validators.toml \
+    --persisted-peers-path /home/monad/monad-bft/peers.toml \
     --keystore-password "${KEYSTORE_PASSWORD}"
 EnvironmentFile=-/home/monad/.env
 Environment="RUST_LOG=debug,h2=warn,tower=warn,opentelemetry_sdk=warn,opentelemetry-otlp=warn"

debian/usr/lib/systemd/system/monad-bft.service

@@ -74,6 +74,7 @@ services:
       --node-config /monad/config/node.toml
       --forkpoint-config /monad/config/forkpoint.genesis.toml
       --validators-path /monad/config/validators.toml
+      --persisted-peers-path /monad/peers.toml
       --statesync-ipc-path /monad/statesync.sock
       --wal-path /monad/wal
       --mempool-ipc-path /monad/mempool.sock

docker/single-node/nets/compose.yaml

@@ -95,4 +95,8 @@ pub struct Cli {
 
     #[arg(long)]
     pub manytrace_socket: Option<String>,
+
+    /// Set the path for the file that will persist peer discovery records across restarts
+    #[arg(long)]
+    pub persisted_peers_path: PathBuf,
 }

monad-node/src/cli.rs

@@ -17,6 +17,7 @@ use std::{
     collections::{BTreeMap, BTreeSet, HashMap},
     marker::PhantomData,
     net::{IpAddr, SocketAddr, ToSocketAddrs},
+    path::PathBuf,
     process,
     sync::Arc,
     time::{Duration, Instant},
@@ -171,6 +172,7 @@ async fn run(node_state: NodeState) -> Result<(), ()> {
         locked_epoch_validators.clone(),
         current_epoch,
         current_round,
+        node_state.persisted_peers_path,
     );
 
     let statesync_threshold: usize = node_state.node_config.statesync_threshold.into();
@@ -507,6 +509,7 @@ fn build_raptorcast_router<ST, SCT, M, OM>(
     locked_epoch_validators: Vec<ValidatorSetDataWithEpoch<SCT>>,
     current_epoch: Epoch,
     current_round: Round,
+    persisted_peers_path: PathBuf,
 ) -> MultiRouter<ST, M, OM, MonadEvent<ST, SCT, ExecutionProtocolType>, PeerDiscovery<ST>>
 where
     ST: CertificateSignatureRecoverable,
@@ -657,6 +660,7 @@ where
         enable_publisher: node_config.fullnode_raptorcast.enable_publisher,
         enable_client: node_config.fullnode_raptorcast.enable_client,
         rng: ChaCha8Rng::from_entropy(),
+        persisted_peers_path,
     };
 
     MultiRouter::new(

monad-node/src/main.rs

@@ -72,6 +72,7 @@ pub struct NodeState {
     // should be kept as long as node is alive, tracing listener is stopped when handle is dropped
     #[allow(unused)]
     manytrace_agent: Option<agent::Agent>,
+    pub persisted_peers_path: PathBuf,
 }
 
 impl NodeState {
@@ -95,6 +96,7 @@ impl NodeState {
             record_metrics_interval_seconds,
             pprof,
             manytrace_socket,
+            persisted_peers_path,
         } = Cli::from_arg_matches_mut(&mut cmd.get_matches_mut())?;
 
         let (reload_handle, agent) = NodeState::setup_tracing(manytrace_socket)?;
@@ -200,6 +202,7 @@ impl NodeState {
             pprof,
             reload_handle,
             manytrace_agent: agent,
+            persisted_peers_path,
         })
     }
 

monad-node/src/state.rs

@@ -218,6 +218,7 @@ fn setup_keys_and_swarm_builder(
                         enable_publisher: secondary_raptorcast_enabled,
                         enable_client: secondary_raptorcast_enabled,
                         rng: ChaCha8Rng::seed_from_u64(123456), // fixed seed for reproducibility
+                        persisted_peers_path: Default::default(),
                     },
                     router_scheduler: NoSerRouterConfig::new(all_peers.keys().cloned().collect())
                         .build(),
@@ -396,6 +397,7 @@ fn test_update_name_record() {
             enable_publisher: false,
             enable_client: false,
             rng: ChaCha8Rng::seed_from_u64(123456),
+            persisted_peers_path: Default::default(),
         },
         router_scheduler: NoSerRouterConfig::new(node_ids.iter().cloned().collect()).build(),
         seed: 1,

monad-peer-disc-swarm/tests/peer_discovery.rs

@@ -14,8 +14,10 @@ arrayvec = { workspace = true }
 futures = { workspace = true }
 rand = { workspace = true }
 rand_chacha = { workspace = true }
+serde = { workspace = true }
 tracing = { workspace = true }
 tokio-util = { workspace = true, features = ["time"] }
+toml = { workspace = true }
 
 [dev-dependencies]
 monad-keystore = { workspace = true }
@@ -28,4 +30,3 @@ hex = { workspace = true }
 insta = { workspace = true }
 rstest = { workspace = true }
 test-case = { workspace = true }
-toml = { workspace = true }

monad-peer-discovery/Cargo.toml

@@ -16,6 +16,7 @@
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap, HashSet},
     net::SocketAddrV4,
+    path::PathBuf,
     time::Duration,
 };
 
@@ -165,6 +166,7 @@ pub struct PeerDiscovery<ST: CertificateSignatureRecoverable> {
     // secondary raptorcast setting: enable client mode when self is a full node
     pub enable_client: bool,
     pub rng: ChaCha8Rng,
+    pub persisted_peers_path: PathBuf,
 }
 
 pub struct PeerDiscoveryBuilder<ST: CertificateSignatureRecoverable> {
@@ -186,6 +188,7 @@ pub struct PeerDiscoveryBuilder<ST: CertificateSignatureRecoverable> {
     pub enable_publisher: bool,
     pub enable_client: bool,
     pub rng: ChaCha8Rng,
+    pub persisted_peers_path: PathBuf,
 }
 
 impl<ST: CertificateSignatureRecoverable> PeerDiscoveryAlgoBuilder for PeerDiscoveryBuilder<ST> {
@@ -258,6 +261,7 @@ impl<ST: CertificateSignatureRecoverable> PeerDiscoveryAlgoBuilder for PeerDisco
             enable_publisher: self.enable_publisher,
             enable_client: self.enable_client,
             rng: self.rng,
+            persisted_peers_path: self.persisted_peers_path,
         };
 
         let mut cmds = Vec::new();
@@ -269,6 +273,58 @@ impl<ST: CertificateSignatureRecoverable> PeerDiscoveryAlgoBuilder for PeerDisco
                 }
             });
 
+        // Attempt to load persisted peers from JSON file and restore into pending queue
+        match std::fs::read_to_string(&state.persisted_peers_path) {
+            Ok(contents) => {
+                match toml::from_str::<
+                    BTreeMap<NodeId<CertificateSignaturePubKey<ST>>, MonadNameRecord<ST>>,
+                >(&contents)
+                {
+                    Ok(loaded) => {
+                        debug!(path =? state.persisted_peers_path, loaded_len = loaded.len(), "loaded persisted peers");
+                        for (peer_id, name_record) in loaded.into_iter() {
+                            if peer_id == state.self_id {
+                                continue;
+                            }
+
+                            match name_record.recover_pubkey() {
+                                Ok(recovered) if recovered == peer_id => {}
+                                _ => {
+                                    warn!(
+                                        ?peer_id,
+                                        "skipping persisted peer with invalid signature"
+                                    );
+                                    continue;
+                                }
+                            }
+
+                            match state.insert_peer_to_pending(peer_id, name_record) {
+                                Ok(cmds_from_insert) => {
+                                    cmds.extend(cmds_from_insert);
+                                    debug!(?peer_id, "inserted persisted peer");
+                                }
+                                Err(err) => {
+                                    warn!(
+                                        ?err,
+                                        "skipping persisted peer due to IP/validation error"
+                                    );
+                                    continue;
+                                }
+                            }
+                        }
+                    }
+                    Err(err) => {
+                        warn!(path =? state.persisted_peers_path, ?err, "failed to deserialize persisted peers");
+                    }
+                }
+            }
+            Err(err) => {
+                debug!(path =? state.persisted_peers_path, ?err, "Skipped loading persisted peers");
+            }
+        }
+
+        state.persist_peers(); // Early test before timer cycle
+
         cmds.extend(state.refresh());
 
         (state, cmds)
@@ -608,6 +664,26 @@ impl<ST: CertificateSignatureRecoverable> PeerDiscovery<ST> {
     fn is_pinned_node(&self, peer_id: &NodeId<CertificateSignaturePubKey<ST>>) -> bool {
         self.check_validator_membership(peer_id) || self.pinned_full_nodes.contains(peer_id)
     }
+
+    fn persist_peers(&self) {
+        // Persist both routing_info and pending_queue to TOML file
+        let mut routing_info = self.routing_info.clone();
+        for (peer_id, conn_info) in self.pending_queue.iter() {
+            routing_info.insert(*peer_id, conn_info.name_record.clone());
+        }
+        match toml::to_string(&routing_info) {
+            Ok(serialized) => {
+                if let Err(e) = std::fs::write(&self.persisted_peers_path, serialized) {
+                    warn!(path =? self.persisted_peers_path, ?e, "failed to persist peers");
+                } else {
+                    debug!(path =? self.persisted_peers_path, routing_len =? self.routing_info.len(), "persisted peers to disk (initial)");
+                }
+            }
+            Err(err) => {
+                warn!(path =? self.persisted_peers_path, ?err, "failed to serialize peers");
+            }
+        }
+    }
 }
 
 impl<ST> PeerDiscoveryAlgo for PeerDiscovery<ST>
@@ -1291,6 +1367,8 @@ where
         self.metrics[GAUGE_PEER_DISC_NUM_PEERS] = self.routing_info.len() as u64;
         self.metrics[GAUGE_PEER_DISC_NUM_PENDING_PEERS] = self.pending_queue.len() as u64;
 
+        self.persist_peers();
+
         // reset timer to schedule for the next refresh
         cmds.extend(self.reset_refresh_timer());
 
@@ -1625,6 +1703,7 @@ mod tests {
             enable_publisher: false,
             enable_client: false,
             rng: ChaCha8Rng::seed_from_u64(123456),
+            persisted_peers_path: Default::default(),
         }
     }
 

monad-peer-discovery/src/discovery.rs

@@ -31,6 +31,7 @@ use monad_crypto::{
 use monad_executor::ExecutorMetrics;
 use monad_executor_glue::PeerEntry;
 use monad_types::{Epoch, NodeId, Round};
+use serde::{Deserialize, Serialize};
 use tracing::{debug, warn};
 
 pub mod discovery;
@@ -47,7 +48,8 @@ pub enum PortTag {
     UDP = 1,
 }
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, RlpEncodable, RlpDecodable)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, RlpEncodable, RlpDecodable, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct Port {
     pub tag: u8,
     pub port: u16,
@@ -70,7 +72,8 @@ impl Port {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 struct WireNameRecordV1 {
     pub ip: Ipv4Addr,
     pub port: u16,
@@ -112,7 +115,8 @@ impl WireNameRecordV1 {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 struct PortList<const N: usize>(ArrayVec<Port, N>);
 
 impl<const N: usize> Encodable for PortList<N> {
@@ -182,7 +186,8 @@ impl<const N: usize> PortList<N> {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 struct WireNameRecordV2 {
     pub ip: Ipv4Addr,
     pub ports: PortList<8>,
@@ -256,13 +261,15 @@ impl WireNameRecordV2 {
     }
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 enum VersionedNameRecord {
     V1(WireNameRecordV1),
     V2(WireNameRecordV2),
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
 pub struct NameRecord {
     record: VersionedNameRecord,
 }
@@ -384,7 +391,9 @@ impl Decodable for NameRecord {
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
 pub enum Capability {}
 
-#[derive(Debug, Clone, PartialEq, RlpEncodable, RlpDecodable, Eq)]
+#[derive(Debug, Clone, PartialEq, RlpEncodable, RlpDecodable, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+#[serde(bound = "ST: CertificateSignatureRecoverable")]
 pub struct MonadNameRecord<ST: CertificateSignatureRecoverable> {
     pub name_record: NameRecord,
     pub signature: ST,