bitfield dist logging cleanup and enhancements (#5147)

* split metrics from bitfield signing

* cleanup all logging

* add a unit test for subset generation

* chore: add one more test to assert need is properly represented

* u8 as usize

* chore: overseer fixin

* fix test

* Update node/network/bitfield-distribution/src/metrics.rs

Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com>

* Update node/network/bitfield-distribution/src/metrics.rs

Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com>

* fallout from suggested rename

* consistency

Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com>

polkadot/node/network/bitfield-distribution/src/tests.rs

@@ -23,9 +23,15 @@ use polkadot_node_network_protocol::{our_view, view, ObservedRole};
 use polkadot_node_subsystem_test_helpers::make_subsystem_context;
 use polkadot_node_subsystem_util::TimeoutExt;
 use polkadot_primitives::v2::{AvailabilityBitfield, Signed, ValidatorIndex};
-use polkadot_subsystem::jaeger;
+use polkadot_subsystem::{
+	jaeger,
+	jaeger::{PerLeafSpan, Span},
+};
 use sp_application_crypto::AppKey;
+use sp_core::Pair as PairT;
+use sp_keyring::Sr25519Keyring;
 use sp_keystore::{testing::KeyStore, SyncCryptoStore, SyncCryptoStorePtr};
+
 use std::{iter::FromIterator as _, sync::Arc, time::Duration};
 
 macro_rules! launch {
@@ -720,3 +726,63 @@ fn do_not_send_message_back_to_origin() {
 		);
 	});
 }
+
+#[test]
+fn need_message_works() {
+	let validators = vec![Sr25519Keyring::Alice.pair(), Sr25519Keyring::Bob.pair()];
+
+	let validator_set = Vec::from_iter(validators.iter().map(|k| ValidatorId::from(k.public())));
+
+	let signing_context = SigningContext { session_index: 1, parent_hash: Hash::repeat_byte(0x00) };
+	let mut state = PerRelayParentData::new(
+		signing_context,
+		validator_set.clone(),
+		PerLeafSpan::new(Arc::new(Span::Disabled), "foo"),
+	);
+
+	let peer_a = PeerId::random();
+	let peer_b = PeerId::random();
+	assert_ne!(peer_a, peer_b);
+
+	let pretend_send =
+		|state: &mut PerRelayParentData, dest_peer: PeerId, signed_by: &ValidatorId| -> bool {
+			if state.message_from_validator_needed_by_peer(&dest_peer, signed_by) {
+				state
+					.message_sent_to_peer
+					.entry(dest_peer)
+					.or_default()
+					.insert(signed_by.clone());
+				true
+			} else {
+				false
+			}
+		};
+
+	let pretend_receive =
+		|state: &mut PerRelayParentData, source_peer: PeerId, signed_by: &ValidatorId| {
+			state
+				.message_received_from_peer
+				.entry(source_peer)
+				.or_default()
+				.insert(signed_by.clone());
+		};
+
+	assert!(true == pretend_send(&mut state, peer_a, &validator_set[0]));
+	assert!(true == pretend_send(&mut state, peer_b, &validator_set[1]));
+	// sending the same thing must not be allowed
+	assert!(false == pretend_send(&mut state, peer_a, &validator_set[0]));
+
+	// receive by Alice
+	pretend_receive(&mut state, peer_a, &validator_set[0]);
+	// must be marked as not needed by Alice, so attempt to send to Alice must be false
+	assert!(false == pretend_send(&mut state, peer_a, &validator_set[0]));
+	// but ok for Bob
+	assert!(false == pretend_send(&mut state, peer_b, &validator_set[1]));
+
+	// receive by Bob
+	pretend_receive(&mut state, peer_a, &validator_set[0]);
+	// not ok for Alice
+	assert!(false == pretend_send(&mut state, peer_a, &validator_set[0]));
+	// also not ok for Bob
+	assert!(false == pretend_send(&mut state, peer_b, &validator_set[1]));
+}