approval-distribution: split peer knowledge into sent and received (#2809)

* approval-distribution: split peer knowledge into sent and received

* guide updates

* fixes

* revert doc changes

polkadot/node/network/approval-distribution/src/lib.rs

@@ -95,11 +95,35 @@ struct Knowledge {
 	known_messages: HashSet<MessageFingerprint>,
 }
 
+impl Knowledge {
+	fn contains(&self, fingerprint: &MessageFingerprint) -> bool {
+		self.known_messages.contains(fingerprint)
+	}
+
+	fn insert(&mut self, fingerprint: MessageFingerprint) -> bool {
+		self.known_messages.insert(fingerprint)
+	}
+}
+
+#[derive(Debug, Clone, Default)]
+struct PeerKnowledge {
+	/// The knowledge we've sent to the peer.
+	sent: Knowledge,
+	/// The knowledge we've received from the peer.
+	received: Knowledge,
+}
+
+impl PeerKnowledge {
+	fn contains(&self, fingerprint: &MessageFingerprint) -> bool {
+		self.sent.contains(fingerprint) || self.received.contains(fingerprint)
+	}
+}
+
 /// Information about blocks in our current view as well as whether peers know of them.
 struct BlockEntry {
 	/// Peers who we know are aware of this block and thus, the candidates within it.
 	/// This maps to their knowledge of messages.
-	known_by: HashMap<PeerId, Knowledge>,
+	known_by: HashMap<PeerId, PeerKnowledge>,
 	/// The number of the block.
 	number: BlockNumber,
 	/// The parent hash of the block.
@@ -212,7 +236,6 @@ impl State {
 							"Cleaning up stale pending messages",
 						);
 					}
-
 					live
 				});
 			}
@@ -513,15 +536,19 @@ impl State {
 		if let Some(peer_id) = source.peer_id() {
 			// check if our knowledge of the peer already contains this assignment
 			match entry.known_by.entry(peer_id.clone()) {
-				hash_map::Entry::Occupied(knowledge) => {
-					if knowledge.get().known_messages.contains(&fingerprint) {
-						tracing::debug!(
-							target: LOG_TARGET,
-							?peer_id,
-							?fingerprint,
-							"Duplicate assignment",
-						);
-						modify_reputation(ctx, peer_id, COST_DUPLICATE_MESSAGE).await;
+				hash_map::Entry::Occupied(mut peer_knowledge) => {
+					let peer_knowledge = peer_knowledge.get_mut();
+					if peer_knowledge.contains(&fingerprint) {
+						if peer_knowledge.received.contains(&fingerprint) {
+							tracing::debug!(
+								target: LOG_TARGET,
+								?peer_id,
+								?fingerprint,
+								"Duplicate assignment",
+							);
+							modify_reputation(ctx, peer_id, COST_DUPLICATE_MESSAGE).await;
+						}
+						peer_knowledge.received.insert(fingerprint);
 						return;
 					}
 				}
@@ -546,7 +573,7 @@ impl State {
 						?fingerprint,
 						"Known assignment",
 					);
-					peer_knowledge.known_messages.insert(fingerprint.clone());
+					peer_knowledge.received.insert(fingerprint.clone());
 				}
 				return;
 			}
@@ -584,7 +611,7 @@ impl State {
 					modify_reputation(ctx, peer_id.clone(), BENEFIT_VALID_MESSAGE_FIRST).await;
 					entry.knowledge.known_messages.insert(fingerprint.clone());
 					if let Some(peer_knowledge) = entry.known_by.get_mut(&peer_id) {
-						peer_knowledge.known_messages.insert(fingerprint.clone());
+						peer_knowledge.received.insert(fingerprint.clone());
 					}
 				}
 				AssignmentCheckResult::AcceptedDuplicate => {
@@ -592,7 +619,7 @@ impl State {
 					// There is more than one way each validator can be assigned to each core.
 					// cf. https://github.com/paritytech/polkadot/pull/2160#discussion_r557628699
 					if let Some(peer_knowledge) = entry.known_by.get_mut(&peer_id) {
-						peer_knowledge.known_messages.insert(fingerprint);
+						peer_knowledge.received.insert(fingerprint);
 					}
 					return;
 				}
@@ -663,8 +690,8 @@ impl State {
 		// Add the fingerprint of the assignment to the knowledge of each peer.
 		for peer in peers.iter() {
 			// we already filtered peers above, so this should always be Some
-			if let Some(entry) = entry.known_by.get_mut(peer) {
-				entry.known_messages.insert(fingerprint.clone());
+			if let Some(peer_knowledge) = entry.known_by.get_mut(peer) {
+				peer_knowledge.sent.insert(fingerprint.clone());
 			}
 		}
 
@@ -735,16 +762,20 @@ impl State {
 
 			// check if our knowledge of the peer already contains this approval
 			match entry.known_by.entry(peer_id.clone()) {
-				hash_map::Entry::Occupied(knowledge) => {
-					if knowledge.get().known_messages.contains(&fingerprint) {
-						tracing::debug!(
-							target: LOG_TARGET,
-							?peer_id,
-							?fingerprint,
-							"Duplicate approval",
-						);
+				hash_map::Entry::Occupied(mut knowledge) => {
+					let peer_knowledge = knowledge.get_mut();
+					if peer_knowledge.contains(&fingerprint) {
+						if peer_knowledge.received.contains(&fingerprint) {
+							tracing::debug!(
+								target: LOG_TARGET,
+								?peer_id,
+								?fingerprint,
+								"Duplicate approval",
+							);
 
-						modify_reputation(ctx, peer_id, COST_DUPLICATE_MESSAGE).await;
+							modify_reputation(ctx, peer_id, COST_DUPLICATE_MESSAGE).await;
+						}
+						peer_knowledge.received.insert(fingerprint);
 						return;
 					}
 				}
@@ -760,7 +791,7 @@ impl State {
 			}
 
 			// if the approval is known to be valid, reward the peer
-			if entry.knowledge.known_messages.contains(&fingerprint) {
+			if entry.knowledge.contains(&fingerprint) {
 				tracing::trace!(
 					target: LOG_TARGET,
 					?peer_id,
@@ -769,7 +800,7 @@ impl State {
 				);
 				modify_reputation(ctx, peer_id.clone(), BENEFIT_VALID_MESSAGE).await;
 				if let Some(peer_knowledge) = entry.known_by.get_mut(&peer_id) {
-					peer_knowledge.known_messages.insert(fingerprint.clone());
+					peer_knowledge.received.insert(fingerprint.clone());
 				}
 				return;
 			}
@@ -805,9 +836,9 @@ impl State {
 				ApprovalCheckResult::Accepted => {
 					modify_reputation(ctx, peer_id.clone(), BENEFIT_VALID_MESSAGE_FIRST).await;
 
-					entry.knowledge.known_messages.insert(fingerprint.clone());
+					entry.knowledge.insert(fingerprint.clone());
 					if let Some(peer_knowledge) = entry.known_by.get_mut(&peer_id) {
-						peer_knowledge.known_messages.insert(fingerprint.clone());
+						peer_knowledge.received.insert(fingerprint.clone());
 					}
 				}
 				ApprovalCheckResult::Bad => {
@@ -821,7 +852,7 @@ impl State {
 				}
 			}
 		} else {
-			if !entry.knowledge.known_messages.insert(fingerprint.clone()) {
+			if !entry.knowledge.insert(fingerprint.clone()) {
 				// if we already imported an approval, there is no need to distribute it again
 				tracing::warn!(
 					target: LOG_TARGET,
@@ -898,7 +929,7 @@ impl State {
 		for peer in peers.iter() {
 			// we already filtered peers above, so this should always be Some
 			if let Some(entry) = entry.known_by.get_mut(peer) {
-				entry.known_messages.insert(fingerprint.clone());
+				entry.sent.insert(fingerprint.clone());
 			}
 		}
 
@@ -947,7 +978,11 @@ impl State {
 					hash_map::Entry::Occupied(_) => return None,
 					// step 4.
 					hash_map::Entry::Vacant(vacant) => {
-						vacant.insert(entry.knowledge.clone());
+						let knowledge = PeerKnowledge {
+							sent: entry.knowledge.clone(),
+							received: Default::default(),
+						};
+						vacant.insert(knowledge);
 						block
 					}
 				};