rewards-statistics-collector subsystem

Cargo.toml

@@ -180,13 +180,15 @@ members = [
 	"polkadot/node/core/dispute-coordinator",
 	"polkadot/node/core/parachains-inherent",
 	"polkadot/node/core/prospective-parachains",
+	"polkadot/node/core/consensus-statistics-collector",
 	"polkadot/node/core/provisioner",
 	"polkadot/node/core/pvf",
 	"polkadot/node/core/pvf-checker",
 	"polkadot/node/core/pvf/common",
 	"polkadot/node/core/pvf/execute-worker",
 	"polkadot/node/core/pvf/prepare-worker",
 	"polkadot/node/core/runtime-api",
+	"polkadot/node/core/consensus-statistics-collector",
 	"polkadot/node/gum",
 	"polkadot/node/gum/proc-macro",
 	"polkadot/node/malus",
@@ -1126,6 +1128,7 @@ polkadot-node-core-chain-selection = { path = "polkadot/node/core/chain-selectio
 polkadot-node-core-dispute-coordinator = { path = "polkadot/node/core/dispute-coordinator", default-features = false }
 polkadot-node-core-parachains-inherent = { path = "polkadot/node/core/parachains-inherent", default-features = false }
 polkadot-node-core-prospective-parachains = { path = "polkadot/node/core/prospective-parachains", default-features = false }
+polkadot-node-core-consensus-statistics-collector = { path = "polkadot/node/core/consensus-statistics-collector", default-features = false }
 polkadot-node-core-provisioner = { path = "polkadot/node/core/provisioner", default-features = false }
 polkadot-node-core-pvf = { path = "polkadot/node/core/pvf", default-features = false }
 polkadot-node-core-pvf-checker = { path = "polkadot/node/core/pvf-checker", default-features = false }

polkadot/node/core/approval-voting/src/import.rs

@@ -66,7 +66,7 @@ use crate::{
 };
 
 use polkadot_node_primitives::approval::time::{slot_number_to_tick, Tick};
-
+use polkadot_node_subsystem::messages::ConsensusStatisticsCollectorMessage;
 use super::{State, LOG_TARGET};
 
 #[derive(Debug)]
@@ -335,7 +335,8 @@ pub struct BlockImportedCandidates {
 pub(crate) async fn handle_new_head<
 	Sender: SubsystemSender<ChainApiMessage>
 		+ SubsystemSender<RuntimeApiMessage>
-		+ SubsystemSender<ChainSelectionMessage>,
+		+ SubsystemSender<ChainSelectionMessage>
+		+ SubsystemSender<ConsensusStatisticsCollectorMessage>,
 	AVSender: SubsystemSender<ApprovalDistributionMessage>,
 	B: Backend,
 >(

polkadot/node/core/approval-voting/src/lib.rs

@@ -95,6 +95,7 @@ use persisted_entries::{ApprovalEntry, BlockEntry, CandidateEntry};
 use polkadot_node_primitives::approval::time::{
 	slot_number_to_tick, Clock, ClockExt, DelayedApprovalTimer, SystemClock, Tick,
 };
+use polkadot_node_subsystem::messages::ConsensusStatisticsCollectorMessage;
 
 mod approval_checking;
 pub mod approval_db;
@@ -2883,7 +2884,8 @@ async fn import_approval<Sender>(
 	wakeups: &Wakeups,
 ) -> SubsystemResult<(Vec<Action>, ApprovalCheckResult)>
 where
-	Sender: SubsystemSender<RuntimeApiMessage>,
+	Sender: SubsystemSender<RuntimeApiMessage> +
+	SubsystemSender<ConsensusStatisticsCollectorMessage>,
 {
 	macro_rules! respond_early {
 		($e: expr) => {{
@@ -2922,7 +2924,7 @@ where
 	gum::trace!(
 		target: LOG_TARGET,
 		"Received approval for num_candidates {:}",
-		approval.candidate_indices.count_ones()
+		approval.candidate_indices.clone().count_ones()
 	);
 
 	let mut actions = Vec::new();
@@ -3036,7 +3038,8 @@ async fn advance_approval_state<Sender>(
 	wakeups: &Wakeups,
 ) -> Vec<Action>
 where
-	Sender: SubsystemSender<RuntimeApiMessage>,
+	Sender: SubsystemSender<RuntimeApiMessage> +
+	SubsystemSender<ConsensusStatisticsCollectorMessage>,
 {
 	let validator_index = transition.validator_index();
 
@@ -3091,6 +3094,14 @@ where
 		let is_approved = check.is_approved(tick_now.saturating_sub(APPROVAL_DELAY));
 		if status.last_no_shows != 0 {
 			metrics.on_observed_no_shows(status.last_no_shows);
+			sender
+				.send_message(
+					ConsensusStatisticsCollectorMessage::ObservedNoShows(
+						session_index,
+						status.no_show_validators.clone(),
+					))
+				.await;
+
 			gum::trace!(
 				target: LOG_TARGET,
 				?candidate_hash,
@@ -3163,7 +3174,50 @@ where
 		}
 		if newly_approved {
 			state.record_no_shows(session_index, para_id.into(), &status.no_show_validators);
+			let mut collected_useful_approvals = vec![];
+
+			match status.required_tranches {
+				RequiredTranches::All => {
+					collected_useful_approvals.extend(candidate_entry.approvals().iter_ones().into());
+				},
+				RequiredTranches::Exact {needed, ..} => {
+					let tranches = approval_entry.tranches();
+					for validator_approval_idx in candidate_entry.approvals().iter_ones() {
+						for tranche_entry in tranches {
+							if tranche_entry.tranche() <= needed {
+								let trance_assignments = tranche_entry.assignments();
+								let useful_vote = trance_assignments
+									.iter()
+									.find(|(validator_on_tranche, _)| *validator_on_tranche == ValidatorIndex(validator_approval_idx as _));
+
+								match useful_vote {
+									Some((vidx, _)) => collected_useful_approvals.push(*vidx),
+									// Found no useful votes on a needed tranche
+									None => {}
+								}
+							}
+						}
+					}
+				},
+				RequiredTranches::Pending {..} => panic!("Newly approved candidate should never be pending; qed"),
+			}
+
+			if collected_useful_approvals.len() > 0 {
+				_ = sender.try_send_message(ConsensusStatisticsCollectorMessage::CandidateApproved(
+					candidate_hash,
+					block_hash,
+					collected_useful_approvals,
+				)).map_err(|_| {
+					gum::warn!(
+					target: LOG_TARGET,
+					?candidate_hash,
+					?block_hash,
+					"Failed to send approvals to statistics subsystem",
+				);
+				});
+			}
 		}
+
 		actions.extend(schedule_wakeup_action(
 			&approval_entry,
 			block_hash,

polkadot/node/core/consensus-statistics-collector/Cargo.toml

@@ -0,0 +1,33 @@
+[package]
+name = "polkadot-node-core-consensus-statistics-collector"
+version = "6.0.0"
+authors.workspace = true
+edition.workspace = true
+license.workspace = true
+description = "The Statistics Collector subsystem. Collects Approval Voting and Approvals Distributions stats."
+homepage.workspace = true
+repository.workspace = true
+
+[lints]
+workspace = true
+
+[dependencies]
+fatality = { workspace = true }
+futures = { workspace = true }
+gum = { workspace = true, default-features = true }
+thiserror = { workspace = true }
+
+polkadot-node-subsystem = { workspace = true, default-features = true }
+polkadot-node-subsystem-util = { workspace = true, default-features = true }
+polkadot-primitives = { workspace = true, default-features = true }
+polkadot-node-primitives = { workspace = true, default-features = true }
+
+[dev-dependencies]
+assert_matches = { workspace = true }
+polkadot-node-subsystem-test-helpers = { workspace = true }
+polkadot-primitives = { workspace = true, features = ["test"] }
+polkadot-primitives-test-helpers = { workspace = true }
+rand = { workspace = true }
+rstest = { workspace = true }
+sp-core = { workspace = true, default-features = true }
+sp-tracing = { workspace = true }

polkadot/node/core/consensus-statistics-collector/src/approval_voting_metrics.rs

@@ -0,0 +1,48 @@
+use std::collections::{HashMap, HashSet};
+use polkadot_primitives::{CandidateHash, Hash, SessionIndex, ValidatorIndex};
+use crate::View;
+use crate::error::Result;
+
+#[derive(Debug, Clone, Default)]
+pub struct ApprovalsStats {
+    pub votes: HashSet<ValidatorIndex>,
+}
+
+impl ApprovalsStats {
+    pub fn new(votes: HashSet<ValidatorIndex>) -> Self {
+        Self { votes }
+    }
+}
+
+pub fn handle_candidate_approved(
+    view: &mut View,
+    block_hash: Hash,
+    candidate_hash: CandidateHash,
+    approvals: Vec<ValidatorIndex>
+) {
+    if let Some(relay_view) = view.per_relay.get_mut(&block_hash) {
+        relay_view.approvals_stats
+            .entry(candidate_hash)
+            .or_default()
+            .and_modify(|a: &mut ApprovalsStats| {
+                a.votes.extend(approvals.into());
+            });
+    }
+}
+
+pub fn handle_observed_no_shows(
+    view: &mut View,
+    session_index: SessionIndex,
+    no_show_validators: Vec<ValidatorIndex>,
+) {
+    view.no_shows_per_session
+        .entry(session_index)
+        .and_modify(|q: &mut HashMap<ValidatorIndex, usize>| {
+            for v_idx in no_show_validators {
+                q.entry(*v_idx)
+                    .and_modify(|v: &mut usize| *v += 1)
+                    .or_insert(1);
+            }
+        })
+        .or_insert(HashMap::new());
+}

polkadot/node/core/consensus-statistics-collector/src/availability_distribution_metrics.rs

@@ -0,0 +1,51 @@
+use std::collections::{HashMap, HashSet};
+use std::ops::Add;
+use gum::CandidateHash;
+use polkadot_primitives::{SessionIndex, ValidatorIndex};
+use crate::View;
+
+pub struct ChunksDownloaded {
+    pub per_candidate: HashMap<CandidateHash, HashMap<ValidatorIndex, u64>>,
+}
+
+impl ChunksDownloaded {
+    pub fn new() -> Self {
+        Self {
+            per_candidate: Default::default(),
+        }
+    }
+
+    pub fn note_candidate_chunk_downloaded(
+        &mut self,
+        hash: CandidateHash,
+        validator_index: ValidatorIndex,
+        count: u64,
+    ) {
+        self.per_candidate
+            .entry(hash)
+            .or_default()
+            .entry(validator_index)
+            .or_default()
+            .add(count);
+    }
+}
+
+// whenever chunks are acquired throughout availability
+// recovery we collect the metrics about what validator
+// provided and the amount of chunks
+pub fn handle_chunks_downloaded(
+    view: &mut View,
+    session_index: SessionIndex,
+    candidate_hash: CandidateHash,
+    downloads: HashMap<ValidatorIndex, u64>,
+) {
+    view.candidates_per_session
+        .entry(session_index)
+        .or_default()
+        .insert(candidate_hash);
+
+    for (validator_index, download_count) in downloads {
+        view.chunks_downloaded
+            .note_candidate_chunk_downloaded(candidate_hash, validator_index, download_count)
+    }
+}

polkadot/node/core/consensus-statistics-collector/src/error.rs

@@ -0,0 +1,56 @@
+// Copyright (C) Parity Technologies (UK) Ltd.
+// This file is part of Polkadot.
+
+// Polkadot is free software: you can redistribute it and/or modify
+// it under the terms of the GNU General Public License as published by
+// the Free Software Foundation, either version 3 of the License, or
+// (at your option) any later version.
+
+// Polkadot is distributed in the hope that it will be useful,
+// but WITHOUT ANY WARRANTY; without even the implied warranty of
+// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+// GNU General Public License for more details.
+
+// You should have received a copy of the GNU General Public License
+// along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
+
+//! Error types.
+
+use futures::channel::oneshot;
+
+use polkadot_node_subsystem::{
+	SubsystemError,
+};
+use polkadot_node_subsystem_util::runtime;
+
+use crate::LOG_TARGET;
+use fatality::Nested;
+
+#[allow(missing_docs)]
+#[fatality::fatality(splitable)]
+pub enum Error {
+	#[fatal]
+	#[error("Receiving message from overseer failed: {0}")]
+	SubsystemReceive(#[source] SubsystemError),
+}
+
+/// General `Result` type.
+pub type Result<R> = std::result::Result<R, Error>;
+/// Result for non-fatal only failures.
+pub type JfyiErrorResult<T> = std::result::Result<T, JfyiError>;
+/// Result for fatal only failures.
+pub type FatalResult<T> = std::result::Result<T, FatalError>;
+
+/// Utility for eating top level errors and log them.
+///
+/// We basically always want to try and continue on error. This utility function is meant to
+/// consume top-level errors by simply logging them
+pub fn log_error(result: Result<()>, ctx: &'static str) -> FatalResult<()> {
+	match result.into_nested()? {
+		Ok(()) => Ok(()),
+		Err(jfyi) => {
+			gum::debug!(target: LOG_TARGET, error = ?jfyi, ctx);
+			Ok(())
+		},
+	}
+}