referrerpolicy=no-referrer-when-downgrade
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
// 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/>.

use std::collections::HashSet;

use rand::{seq::SliceRandom, thread_rng};
use schnellru::{ByLength, LruMap};

use polkadot_node_subsystem::overseer;
use polkadot_node_subsystem_util::runtime::{request_node_features, RuntimeInfo};
use polkadot_primitives::{
	AuthorityDiscoveryId, GroupIndex, Hash, NodeFeatures, SessionIndex, ValidatorIndex,
};

use crate::{
	error::{Error, Result},
	LOG_TARGET,
};

/// Caching of session info as needed by availability chunk distribution.
///
/// It should be ensured that a cached session stays live in the cache as long as we might need it.
pub struct SessionCache {
	/// Look up cached sessions by `SessionIndex`.
	///
	/// Note: Performance of fetching is really secondary here, but we need to ensure we are going
	/// to get any existing cache entry, before fetching new information, as we should not mess up
	/// the order of validators in `SessionInfo::validator_groups`.
	session_info_cache: LruMap<SessionIndex, SessionInfo>,
}

/// Localized session information, tailored for the needs of availability distribution.
#[derive(Clone)]
pub struct SessionInfo {
	/// The index of this session.
	pub session_index: SessionIndex,

	/// Validator groups of the current session.
	///
	/// Each group's order is randomized. This way we achieve load balancing when requesting
	/// chunks, as the validators in a group will be tried in that randomized order. Each node
	/// should arrive at a different order, therefore we distribute the load on individual
	/// validators.
	pub validator_groups: Vec<Vec<AuthorityDiscoveryId>>,

	/// Information about ourselves:
	pub our_index: ValidatorIndex,

	/// Remember to which group we belong, so we won't start fetching chunks for candidates with
	/// our group being responsible. (We should have that chunk already.)
	///
	/// `None`, if we are not in fact part of any group.
	pub our_group: Option<GroupIndex>,

	/// Node features.
	pub node_features: Option<NodeFeatures>,
}

/// Report of bad validators.
///
/// Fetching tasks will report back validators that did not respond as expected, so we can re-order
/// them.
pub struct BadValidators {
	/// The session index that was used.
	pub session_index: SessionIndex,
	/// The group, the not properly responding validators belong to.
	pub group_index: GroupIndex,
	/// The list of bad validators.
	pub bad_validators: Vec<AuthorityDiscoveryId>,
}

#[overseer::contextbounds(AvailabilityDistribution, prefix = self::overseer)]
impl SessionCache {
	/// Create a new `SessionCache`.
	pub fn new() -> Self {
		SessionCache {
			// We need to cache the current and the last session the most:
			session_info_cache: LruMap::new(ByLength::new(2)),
		}
	}

	/// Tries to retrieve `SessionInfo`.
	/// If this node is not a validator, the function will return `None`.
	pub async fn get_session_info<'a, Context>(
		&'a mut self,
		ctx: &mut Context,
		runtime: &mut RuntimeInfo,
		parent: Hash,
		session_index: SessionIndex,
	) -> Result<Option<&'a SessionInfo>> {
		gum::trace!(target: LOG_TARGET, session_index, "Calling `get_session_info`");

		if self.session_info_cache.get(&session_index).is_none() {
			if let Some(info) =
				Self::query_info_from_runtime(ctx, runtime, parent, session_index).await?
			{
				gum::trace!(target: LOG_TARGET, session_index, "Storing session info in lru!");
				self.session_info_cache.insert(session_index, info);
			} else {
				return Ok(None)
			}
		}

		Ok(self.session_info_cache.get(&session_index).map(|i| &*i))
	}

	/// Variant of `report_bad` that never fails, but just logs errors.
	///
	/// Not being able to report bad validators is not fatal, so we should not shutdown the
	/// subsystem on this.
	pub fn report_bad_log(&mut self, report: BadValidators) {
		if let Err(err) = self.report_bad(report) {
			gum::warn!(
				target: LOG_TARGET,
				err = ?err,
				"Reporting bad validators failed with error"
			);
		}
	}

	/// Make sure we try unresponsive or misbehaving validators last.
	///
	/// We assume validators in a group are tried in reverse order, so the reported bad validators
	/// will be put at the beginning of the group.
	pub fn report_bad(&mut self, report: BadValidators) -> Result<()> {
		let available_sessions = self.session_info_cache.iter().map(|(k, _)| *k).collect();
		let session = self.session_info_cache.get(&report.session_index).ok_or(
			Error::NoSuchCachedSession {
				available_sessions,
				missing_session: report.session_index,
			},
		)?;
		let group = session.validator_groups.get_mut(report.group_index.0 as usize).expect(
			"A bad validator report must contain a valid group for the reported session. qed.",
		);
		let bad_set = report.bad_validators.iter().collect::<HashSet<_>>();

		// Get rid of bad boys:
		group.retain(|v| !bad_set.contains(v));

		// We are trying validators in reverse order, so bad ones should be first:
		let mut new_group = report.bad_validators;
		new_group.append(group);
		*group = new_group;
		Ok(())
	}

	/// Query needed information from runtime.
	///
	/// We need to pass in the relay parent for our call to `request_session_info`. We should
	/// actually don't need that: I suppose it is used for internal caching based on relay parents,
	/// which we don't use here. It should not do any harm though.
	///
	/// Returns: `None` if not a validator.
	async fn query_info_from_runtime<Context>(
		ctx: &mut Context,
		runtime: &mut RuntimeInfo,
		relay_parent: Hash,
		session_index: SessionIndex,
	) -> Result<Option<SessionInfo>> {
		let info = runtime
			.get_session_info_by_index(ctx.sender(), relay_parent, session_index)
			.await?;

		let node_features =
			request_node_features(relay_parent, session_index, ctx.sender()).await?;

		let discovery_keys = info.session_info.discovery_keys.clone();
		let mut validator_groups = info.session_info.validator_groups.clone();

		if let Some(our_index) = info.validator_info.our_index {
			// Get our group index:
			let our_group = info.validator_info.our_group;

			// Shuffle validators in groups:
			let mut rng = thread_rng();
			for g in validator_groups.iter_mut() {
				g.shuffle(&mut rng)
			}
			// Look up `AuthorityDiscoveryId`s right away:
			let validator_groups: Vec<Vec<_>> = validator_groups
				.into_iter()
				.map(|group| {
					group
						.into_iter()
						.map(|index| {
							discovery_keys.get(index.0 as usize)
								.expect("There should be a discovery key for each validator of each validator group. qed.")
								.clone()
						})
						.collect()
				})
				.collect();

			let info = SessionInfo {
				validator_groups,
				our_index,
				session_index,
				our_group,
				node_features,
			};
			return Ok(Some(info))
		}
		return Ok(None)
	}
}