Persist name records for Peer Discovery

Author: boicotinho

Cargo.lock

@@ -45,6 +45,7 @@ unresponsive_prune_threshold = 5
 last_participation_prune_threshold = 5000
 min_num_peers = 0
 max_num_peers = 200
+persisted_peers_path = "/tmp/peers.toml"
 
 [fullnode_dedicated]
 identities = []

docker/devnet/monad/config/node.toml

@@ -13,6 +13,8 @@
 // You should have received a copy of the GNU General Public License
 // along with this program.  If not, see <http://www.gnu.org/licenses/>.
 
+use std::path::PathBuf;
+
 use monad_crypto::certificate_signature::CertificateSignatureRecoverable;
 use monad_types::Round;
 use serde::{Deserialize, Serialize};
@@ -32,4 +34,5 @@ pub struct PeerDiscoveryConfig<ST: CertificateSignatureRecoverable> {
     pub last_participation_prune_threshold: Round,
     pub min_num_peers: usize,
     pub max_num_peers: usize,
+    pub persisted_peers_path: PathBuf,
 }

monad-node-config/src/peers.rs

@@ -643,6 +643,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: peer_discovery_config.persisted_peers_path,
     };
 
     MultiRouter::new(

monad-node/src/main.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,
 };
 
@@ -27,6 +28,7 @@ use monad_executor_glue::PeerEntry;
 use monad_types::{Epoch, NodeId, Round};
 use rand::{RngCore, seq::IteratorRandom};
 use rand_chacha::ChaCha8Rng;
+use serde::{Deserialize, Serialize};
 use tracing::{debug, info, trace, warn};
 
 use crate::{
@@ -87,13 +89,24 @@ pub enum PeerDiscoveryRole {
     FullNodeClient,     // full node set as Client in secondary raptorcast
 }
 
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+#[serde(bound = "ST: CertificateSignatureRecoverable")]
 pub struct ConnectionInfo<ST: CertificateSignatureRecoverable> {
     pub last_ping: Ping<ST>,
     pub unresponsive_pings: u32,
     pub name_record: MonadNameRecord<ST>,
 }
 
+// Helper struct to deserialize both routing_info and pending_queue from TOML
+#[derive(Debug, Clone, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+#[serde(bound = "ST: CertificateSignatureRecoverable")]
+struct PersistedPeers<ST: CertificateSignatureRecoverable> {
+    routing_info: BTreeMap<NodeId<CertificateSignaturePubKey<ST>>, MonadNameRecord<ST>>,
+    pending_queue: BTreeMap<NodeId<CertificateSignaturePubKey<ST>>, ConnectionInfo<ST>>,
+}
+
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum SecondaryRaptorcastConnectionStatus {
     Connected,
@@ -165,6 +178,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 +200,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 +273,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 +285,60 @@ impl<ST: CertificateSignatureRecoverable> PeerDiscoveryAlgoBuilder for PeerDisco
                 }
             });
 
+        // Load persisted peers from JSON file (admin-friendly).
+        if let Ok(toml_data) = std::fs::read_to_string(&state.persisted_peers_path) {
+            match toml::from_str::<PersistedPeers<ST>>(&toml_data) {
+                Ok(persisted) => {
+                    debug!(
+                        path = ?state.persisted_peers_path,
+                        routing_count = ?persisted.routing_info.len(),
+                        pending_count = ?persisted.pending_queue.len(),
+                        "loaded persisted peers from disk"
+                    );
+
+                    for (peer_id, name_record) in persisted.routing_info {
+                        match state.insert_peer_to_pending(peer_id, name_record) {
+                            Ok(cmds_from_insert) => {
+                                debug!(?peer_id, "Persisted routing peer accepted");
+                                cmds.extend(cmds_from_insert);
+                            }
+                            Err(err) => {
+                                debug!(?peer_id, ?err, "Persisted routing peer rejected");
+                            }
+                        }
+                    }
+
+                    for (peer_id, conn_info) in persisted.pending_queue {
+                        match validate_socket_ipv4_address(
+                            &conn_info.name_record.udp_address(),
+                            &state.self_record.udp_address(),
+                        ) {
+                            Ok(_) => {
+                                state.pending_queue.insert(peer_id, conn_info);
+                                debug!(?peer_id, "Loaded persisted pending peer");
+                            }
+                            Err(err) => {
+                                debug!(?peer_id, ?err, "Persisted pending peer rejected");
+                            }
+                        }
+                    }
+                    state.metrics[GAUGE_PEER_DISC_NUM_PENDING_PEERS] =
+                        state.pending_queue.len() as u64;
+                }
+                Err(err) => {
+                    warn!(
+                        path = ?state.persisted_peers_path,
+                        ?err,
+                        "failed to deserialize persisted peers as TOML"
+                    );
+                }
+            }
+        } else {
+            debug!(path = ?state.persisted_peers_path, "no persisted peers file found, continuing");
+        }
+
+        state.persist_peers(); // Early test before timer cycle
+
         cmds.extend(state.refresh());
 
         (state, cmds)
@@ -1291,6 +1361,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());
 
@@ -1302,6 +1374,27 @@ where
         cmds
     }
 
+    fn persist_peers(&self) {
+        // Persist both routing_info and pending_queue as PersistedPeers (TOML)
+        let persisted = PersistedPeers {
+            routing_info: self.routing_info.clone(),
+            pending_queue: self.pending_queue.clone(),
+        };
+        match toml::to_string(&persisted) {
+            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(), pending_len =? self.pending_queue.len(),
+                    "persisted peers to disk");
+                }
+            }
+            Err(err) => {
+                warn!(path = ?self.persisted_peers_path, ?err, "failed to serialize peers");
+            }
+        }
+    }
+
     fn update_current_round(
         &mut self,
         round: Round,
@@ -1625,6 +1718,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,
@@ -656,6 +665,8 @@ pub trait PeerDiscoveryAlgo {
         NodeId<CertificateSignaturePubKey<Self::SignatureType>>,
         MonadNameRecord<Self::SignatureType>,
     >;
+
+    fn persist_peers(&self);
 }
 
 pub trait PeerDiscoveryAlgoBuilder {

monad-peer-discovery/src/lib.rs

@@ -19,6 +19,7 @@ use monad_crypto::certificate_signature::{
 };
 use monad_executor_glue::Message;
 use monad_types::NodeId;
+use serde::{Deserialize, Serialize};
 
 use crate::{MonadNameRecord, PeerDiscoveryEvent};
 
@@ -114,7 +115,9 @@ impl<ST: CertificateSignatureRecoverable> Decodable for PeerDiscoveryMessage<ST>
     }
 }
 
-#[derive(Debug, Clone, PartialEq, RlpDecodable, RlpEncodable)]
+#[derive(Debug, Clone, PartialEq, RlpDecodable, RlpEncodable, Deserialize, Serialize)]
+#[serde(deny_unknown_fields)]
+#[serde(bound = "ST: CertificateSignatureRecoverable")]
 #[rlp(trailing)]
 pub struct Ping<ST: CertificateSignatureRecoverable> {
     pub id: u32,

monad-peer-discovery/src/message.rs

@@ -279,4 +279,6 @@ where
     ) -> HashMap<NodeId<CertificateSignaturePubKey<ST>>, MonadNameRecord<ST>> {
         HashMap::new()
     }
+
+    fn persist_peers(&self) {}
 }