fix mem-leak on topic peers

Signed-off-by: onur-ozkan <[email protected]>

Author: onur-ozkan

protocols/gossipsub/src/behaviour.rs

@@ -20,9 +20,7 @@
 
 use std::{
     cmp::{max, Ordering},
-    collections::HashSet,
-    collections::{hash_map::Entry, VecDeque},
-    collections::{BTreeSet, HashMap},
+    collections::{hash_map::Entry, BTreeSet, HashMap, HashSet, VecDeque},
     fmt,
     net::IpAddr,
     task::{Context, Poll},
@@ -2070,6 +2068,10 @@ where
 
                     // remove topic from the peer_topics mapping
                     subscribed_topics.remove(topic_hash);
+                    if let Some(peers) = self.topic_peers.get_mut(topic_hash) {
+                        peers.remove(propagation_source);
+                    }
+
                     unsubscribed_peers.push((*propagation_source, topic_hash.clone()));
                     // generate an unsubscribe event to be polled
                     application_event.push(ToSwarm::GenerateEvent(Event::Unsubscribed {
@@ -2079,8 +2081,11 @@ where
                 }
             }
 
-            if let Some(m) = self.metrics.as_mut() {
-                m.set_topic_peers(topic_hash, peer_list.len());
+            if let (Some(m), Some(len)) = (
+                self.metrics.as_mut(),
+                self.topic_peers.get(topic_hash).map(|set| set.len()),
+            ) {
+                m.set_topic_peers(topic_hash, len);
             }
         }
 
@@ -3337,6 +3342,11 @@ where
             // support the protocol.
             self.peer_topics.remove(&peer_id);
 
+            self.topic_peers.retain(|_, peers| {
+                peers.remove(&peer_id);
+                !peers.is_empty()
+            });
+
             // If metrics are enabled, register the disconnection of a peer based on its protocol.
             if let Some(metrics) = self.metrics.as_mut() {
                 let peer_kind = &self