fix: bound gossipsub history and seen message ids

src/graphcast_agent/mod.rs

@@ -11,14 +11,15 @@
 //!
 use self::message_typing::{GraphcastMessage, IdentityValidation, MessageError, RadioPayload};
 use self::waku_handling::{
-    build_content_topics, handle_signal, pubsub_topic, setup_node_handle, WakuHandlingError,
+    build_content_topics, handle_signal, pubsub_topic, setup_node_handle, SeenMsgIds,
+    WakuHandlingError,
 };
 use ethers::signers::WalletError;
 
 use serde::{Deserialize, Serialize};
 
 use async_graphql::{self, Result, SimpleObject};
-use std::collections::{HashMap, HashSet};
+use std::collections::HashMap;
 use std::str::FromStr;
 use std::sync::mpsc::Sender;
 use std::sync::{Arc, Mutex as SyncMutex};
@@ -215,8 +216,8 @@ pub struct GraphcastAgent {
     /// Callbook that make query requests
     pub callbook: CallBook,
     /// msg_seen_ttl is for only relay messages they have not seen before, not effective for client nodes
-    /// A set of message ids sent from the agent
-    pub seen_msg_ids: Arc<SyncMutex<HashSet<String>>>,
+    /// Message ids received or sent by the agent, bounded by SEEN_MSG_IDS_CAP
+    pub seen_msg_ids: Arc<SyncMutex<SeenMsgIds>>,
     /// Sender identity validation mechanism used by the Graphcast agent
     pub id_validation: IdentityValidation,
     //TODO: Consider deprecating this field as it isn't utilized in network_check anymore
@@ -340,7 +341,7 @@ impl GraphcastAgent {
 
         let callbook = CallBook::new(registry_subgraph, network_subgraph, graph_node_endpoint);
 
-        let seen_msg_ids = Arc::new(SyncMutex::new(HashSet::new()));
+        let seen_msg_ids = Arc::new(SyncMutex::new(SeenMsgIds::default()));
         let content_topics = Arc::new(SyncMutex::new(content_topics));
         register_handler(sender, seen_msg_ids.clone(), content_topics.clone())
             .expect("Could not register handler");
@@ -593,7 +594,7 @@ pub enum GraphcastAgentError {
 /// Establish handler for incoming Waku messages
 pub fn register_handler(
     sender: Sender<WakuMessage>,
-    seen_msg_ids: Arc<SyncMutex<HashSet<String>>>,
+    seen_msg_ids: Arc<SyncMutex<SeenMsgIds>>,
     content_topics: Arc<SyncMutex<Vec<WakuContentTopic>>>,
 ) -> Result<(), GraphcastAgentError> {
     let handle_async = move |signal: Signal| {

src/graphcast_agent/waku_handling.rs

@@ -1,6 +1,9 @@
 use std::sync::Mutex as SyncMutex;
 use std::{borrow::Cow, env, num::ParseIntError, sync::Arc};
-use std::{collections::HashSet, time::Duration};
+use std::{
+    collections::{HashSet, VecDeque},
+    time::Duration,
+};
 use std::{net::IpAddr, str::FromStr};
 
 use tracing::{debug, error, info, trace};
@@ -131,7 +134,6 @@ fn node_config(
 
     let gossipsub_params = GossipSubParams {
         seen_messages_ttl_seconds: Some(1800),
-        history_length: Some(100_000),
         ..Default::default()
     };
 
@@ -392,7 +394,7 @@ pub fn boot_node_handle(
 /// Parse and validate incoming message
 pub fn handle_signal(
     signal: Signal,
-    seen_msg_ids: &Arc<SyncMutex<HashSet<String>>>,
+    seen_msg_ids: &Arc<SyncMutex<SeenMsgIds>>,
     content_topics: &Arc<SyncMutex<Vec<WakuContentTopic>>>,
 ) -> Result<WakuMessage, WakuHandlingError> {
     // Do not accept messages that were already received or sent by self
@@ -490,10 +492,63 @@ impl WakuHandlingError {
     }
 }
 
+/// Upper bound on remembered message ids. Gossipsub already drops repeats
+/// within its 1800s seen-TTL, so this only needs to outlast that window.
+pub const SEEN_MSG_IDS_CAP: usize = 10_000;
+
+/// Message ids received or sent by this node, oldest evicted once full.
+#[derive(Debug, Default)]
+pub struct SeenMsgIds {
+    ids: HashSet<String>,
+    order: VecDeque<String>,
+}
+
+impl SeenMsgIds {
+    pub fn contains(&self, id: &str) -> bool {
+        self.ids.contains(id)
+    }
+
+    /// Returns false if the id was already present.
+    pub fn insert(&mut self, id: String) -> bool {
+        if !self.ids.insert(id.clone()) {
+            return false;
+        }
+        self.order.push_back(id);
+        if self.order.len() > SEEN_MSG_IDS_CAP {
+            if let Some(oldest) = self.order.pop_front() {
+                self.ids.remove(&oldest);
+            }
+        }
+        true
+    }
+
+    pub fn len(&self) -> usize {
+        self.ids.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.ids.is_empty()
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
 
+    #[test]
+    fn test_seen_msg_ids_bounded() {
+        let mut seen = SeenMsgIds::default();
+        for i in 0..(SEEN_MSG_IDS_CAP + 5) {
+            assert!(seen.insert(i.to_string()));
+        }
+        assert_eq!(seen.len(), SEEN_MSG_IDS_CAP);
+        assert!(!seen.contains("0"));
+        assert!(!seen.contains("4"));
+        assert!(seen.contains("5"));
+        assert!(seen.contains(&(SEEN_MSG_IDS_CAP + 4).to_string()));
+        assert!(!seen.insert("5".to_string()));
+    }
+
     #[test]
     fn test_empty_topics() {
         let empty_vec = [].to_vec();