Add from field for message tracking

gossip/src/crds.rs

@@ -96,11 +96,11 @@ pub enum CrdsError {
 }
 
 #[derive(Clone, Copy)]
-pub enum GossipRoute {
+pub enum GossipRoute<'a> {
     LocalMessage,
     PullRequest,
     PullResponse,
-    PushMessage,
+    PushMessage(/*from:*/ &'a Pubkey),
 }
 
 type CrdsCountsArray = [usize; 12];
@@ -301,7 +301,7 @@ impl Crds {
                 if entry.get().value_hash != value.value_hash {
                     self.purged.push_back((value.value_hash, now));
                     Err(CrdsError::InsertFailed)
-                } else if matches!(route, GossipRoute::PushMessage) {
+                } else if matches!(route, GossipRoute::PushMessage(_)) {
                     let entry = entry.get_mut();
                     entry.num_push_dups = entry.num_push_dups.saturating_add(1);
                     Err(CrdsError::DuplicatePush(entry.num_push_dups))
@@ -678,9 +678,11 @@ impl CrdsDataStats {
             }
         }
 
-        if matches!(route, GossipRoute::PushMessage)
-            && should_report_message_signature(&entry.value.signature)
-        {
+        let GossipRoute::PushMessage(from) = route else {
+            return;
+        };
+
+        if should_report_message_signature(&entry.value.signature) {
             datapoint_info!(
                 "gossip_crds_sample",
                 (
@@ -692,6 +694,11 @@ impl CrdsDataStats {
                     "signature",
                     entry.value.signature.to_string().get(..8),
                     Option<String>
+                ),
+                (
+                    "from",
+                    from.to_string().get(..8),
+                    Option<String>
                 )
             );
         }
@@ -725,7 +732,7 @@ impl CrdsStats {
         match route {
             GossipRoute::LocalMessage => (),
             GossipRoute::PullRequest => (),
-            GossipRoute::PushMessage => self.push.record_insert(entry, route),
+            GossipRoute::PushMessage(_) => self.push.record_insert(entry, route),
             GossipRoute::PullResponse => self.pull.record_insert(entry, route),
         }
     }
@@ -734,7 +741,7 @@ impl CrdsStats {
         match route {
             GossipRoute::LocalMessage => (),
             GossipRoute::PullRequest => (),
-            GossipRoute::PushMessage => self.push.record_fail(entry),
+            GossipRoute::PushMessage(_) => self.push.record_fail(entry),
             GossipRoute::PullResponse => self.pull.record_fail(entry),
         }
     }

gossip/src/crds_gossip_push.rs

@@ -144,7 +144,7 @@ impl CrdsGossipPush {
                     continue;
                 }
                 let origin = value.pubkey();
-                match crds.insert(value, now, GossipRoute::PushMessage) {
+                match crds.insert(value, now, GossipRoute::PushMessage(&from)) {
                     Ok(()) => {
                         received_cache.record(origin, from, /*num_dups:*/ 0);
                         origins.insert(origin);