Merge remote-tracking branch 'origin/master' into oty-try-runtime-checks

Author: ggwpez

Cargo.lock

@@ -27,7 +27,7 @@ tikv-jemallocator = "0.5.0"
 assert_cmd = "2.0.4"
 nix = "0.24.1"
 tempfile = "3.2.0"
-tokio = "1.22.0"
+tokio = "1.24.1"
 substrate-rpc-client = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-core-primitives = { path = "core-primitives" }
 

Cargo.toml

@@ -570,6 +570,14 @@ async fn run_iteration<Context>(
 				FromOrchestra::Signal(OverseerSignal::BlockFinalized(hash, number)) => {
 					let _timer = subsystem.metrics.time_process_block_finalized();
 
+					if !subsystem.known_blocks.is_known(&hash) {
+						// If we haven't processed this block yet,
+						// make sure we write the metadata about the
+						// candidates backed in this finalized block.
+						// Otherwise, we won't be able to store our chunk
+						// for these candidates.
+						process_block_activated(ctx, subsystem, hash).await?;
+					}
 					subsystem.finalized_number = Some(number);
 					subsystem.known_blocks.prune_finalized(number);
 					process_block_finalized(

node/core/av-store/src/lib.rs

@@ -732,8 +732,49 @@ fn we_dont_miss_anything_if_import_notifications_are_missed() {
 	let test_state = TestState::default();
 
 	test_harness(test_state.clone(), store.clone(), |mut virtual_overseer| async move {
-		overseer_signal(&mut virtual_overseer, OverseerSignal::BlockFinalized(Hash::zero(), 1))
-			.await;
+		let block_hash = Hash::repeat_byte(1);
+		overseer_signal(&mut virtual_overseer, OverseerSignal::BlockFinalized(block_hash, 1)).await;
+
+		let header = Header {
+			parent_hash: Hash::repeat_byte(0),
+			number: 1,
+			state_root: Hash::zero(),
+			extrinsics_root: Hash::zero(),
+			digest: Default::default(),
+		};
+
+		assert_matches!(
+			overseer_recv(&mut virtual_overseer).await,
+			AllMessages::ChainApi(ChainApiMessage::BlockHeader(
+				relay_parent,
+				tx,
+			)) => {
+				assert_eq!(relay_parent, block_hash);
+				tx.send(Ok(Some(header))).unwrap();
+			}
+		);
+
+		assert_matches!(
+			overseer_recv(&mut virtual_overseer).await,
+			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+				relay_parent,
+				RuntimeApiRequest::CandidateEvents(tx),
+			)) => {
+				assert_eq!(relay_parent, block_hash);
+				tx.send(Ok(Vec::new())).unwrap();
+			}
+		);
+
+		assert_matches!(
+			overseer_recv(&mut virtual_overseer).await,
+			AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+				relay_parent,
+				RuntimeApiRequest::Validators(tx),
+			)) => {
+				assert_eq!(relay_parent, Hash::zero());
+				tx.send(Ok(Vec::new())).unwrap();
+			}
+		);
 
 		let header = Header {
 			parent_hash: Hash::repeat_byte(3),

node/core/av-store/src/tests.rs

@@ -16,8 +16,8 @@ parity-scale-codec = { version = "3.1.5", default-features = false, features = [
 polkadot-primitives = { path = "../../../primitives" }
 polkadot-parachain = { path = "../../../parachain" }
 polkadot-node-primitives = { path = "../../primitives" }
-polkadot-node-subsystem = {path = "../../subsystem" }
-polkadot-node-subsystem-util = { path = "../../subsystem-util" }
+polkadot-node-subsystem = { path = "../../subsystem" }
+polkadot-node-metrics = { path = "../../metrics" }
 
 [target.'cfg(not(any(target_os = "android", target_os = "unknown")))'.dependencies]
 polkadot-node-core-pvf = { path = "../pvf" }
@@ -27,5 +27,6 @@ sp-keyring = { git = "https://github.com/paritytech/substrate", branch = "master
 futures = { version = "0.3.21", features = ["thread-pool"] }
 assert_matches = "1.4.0"
 polkadot-node-subsystem-test-helpers = { path = "../../subsystem-test-helpers" }
+polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 sp-core = { git = "https://github.com/paritytech/substrate", branch = "master" }
 test-helpers = { package = "polkadot-primitives-test-helpers", path = "../../../primitives/test-helpers" }

node/core/candidate-validation/Cargo.toml

@@ -604,13 +604,16 @@ async fn validate_candidate_exhaustive(
 
 #[async_trait]
 trait ValidationBackend {
+	/// Tries executing a PVF a single time (no retries).
 	async fn validate_candidate(
 		&mut self,
 		pvf: Pvf,
 		timeout: Duration,
 		encoded_params: Vec<u8>,
 	) -> Result<WasmValidationResult, ValidationError>;
 
+	/// Tries executing a PVF. Will retry once if an error is encountered that may have been
+	/// transient.
 	async fn validate_candidate_with_retry(
 		&mut self,
 		raw_validation_code: Vec<u8>,
@@ -620,7 +623,7 @@ trait ValidationBackend {
 		// Construct the PVF a single time, since it is an expensive operation. Cloning it is cheap.
 		let pvf = Pvf::from_code(raw_validation_code);
 
-		let validation_result =
+		let mut validation_result =
 			self.validate_candidate(pvf.clone(), timeout, params.encode()).await;
 
 		// If we get an AmbiguousWorkerDeath error, retry once after a brief delay, on the
@@ -630,12 +633,19 @@ trait ValidationBackend {
 		{
 			// Wait a brief delay before retrying.
 			futures_timer::Delay::new(PVF_EXECUTION_RETRY_DELAY).await;
+
+			gum::debug!(
+				target: LOG_TARGET,
+				?pvf,
+				"Re-trying failed candidate validation due to AmbiguousWorkerDeath."
+			);
+
 			// Encode the params again when re-trying. We expect the retry case to be relatively
 			// rare, and we want to avoid unconditionally cloning data.
-			self.validate_candidate(pvf, timeout, params.encode()).await
-		} else {
-			validation_result
+			validation_result = self.validate_candidate(pvf, timeout, params.encode()).await;
 		}
+
+		validation_result
 	}
 
 	async fn precheck_pvf(&mut self, pvf: Pvf) -> Result<Duration, PrepareError>;

node/core/candidate-validation/src/lib.rs

@@ -15,7 +15,7 @@
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
 use super::{ValidationFailed, ValidationResult};
-use polkadot_node_subsystem_util::metrics::{self, prometheus};
+use polkadot_node_metrics::metrics::{self, prometheus};
 
 #[derive(Clone)]
 pub(crate) struct MetricsInner {

node/core/candidate-validation/src/metrics.rs

@@ -9,8 +9,8 @@ futures = "0.3.21"
 gum = { package = "tracing-gum", path = "../../gum" }
 sp-blockchain = { git = "https://github.com/paritytech/substrate", branch = "master" }
 polkadot-primitives = { path = "../../../primitives" }
+polkadot-node-metrics = { path = "../../metrics" }
 polkadot-node-subsystem = {path = "../../subsystem" }
-polkadot-node-subsystem-util = { path = "../../subsystem-util" }
 sc-client-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 sc-consensus-babe = { git = "https://github.com/paritytech/substrate", branch = "master" }
 

node/core/chain-api/Cargo.toml

@@ -14,7 +14,7 @@
 // You should have received a copy of the GNU General Public License
 // along with Polkadot.  If not, see <http://www.gnu.org/licenses/>.
 
-use polkadot_node_subsystem_util::metrics::{self, prometheus};
+use polkadot_node_metrics::metrics::{self, prometheus};
 
 #[derive(Clone)]
 pub(crate) struct MetricsInner {

node/core/chain-api/src/metrics.rs

@@ -395,57 +395,45 @@ impl TestState {
 					);
 					finished_steps.got_scraping_information = true;
 					tx.send(Ok(0)).unwrap();
-
-					// If the activated block number is > 1 the scraper will ask for block ancestors. Handle this case.
-					if block_number > 1 {
-						assert_matches!(
-							overseer_recv(virtual_overseer).await,
-							AllMessages::ChainApi(ChainApiMessage::Ancestors{
-								hash,
-								k,
-								response_channel,
-							}) => {
-								assert_eq!(hash, block_hash);	// A bit restrictive, remove if it causes problems.
-								let target_header = self.headers.get(&hash).expect("The function is called for this block so it should exist");
-								let mut response = Vec::new();
-								for i in target_header.number.saturating_sub(k as u32)..target_header.number {
-									response.push(self.block_num_to_header.get(&i).expect("headers and block_num_to_header should always be in sync").clone());
-								}
-								let _ = response_channel.send(Ok(response));
-							}
-						);
-					}
-
-					assert_matches!(
-					overseer_recv(virtual_overseer).await,
-					AllMessages::RuntimeApi(RuntimeApiMessage::Request(
-							_new_leaf,
-							RuntimeApiRequest::CandidateEvents(tx),
-							)) => {
-						tx.send(Ok(candidate_events.clone())).unwrap();
-					}
-					);
-					gum::trace!("After answering runtime api request");
-					assert_matches!(
-					overseer_recv(virtual_overseer).await,
-					AllMessages::RuntimeApi(RuntimeApiMessage::Request(
-							_new_leaf,
-							RuntimeApiRequest::FetchOnChainVotes(tx),
-							)) => {
-						//add some `BackedCandidates` or resolved disputes here as needed
-						tx.send(Ok(Some(ScrapedOnChainVotes {
-							session,
-							backing_validators_per_candidate: Vec::default(),
-							disputes: MultiDisputeStatementSet::default(),
-						}))).unwrap();
-					}
-					);
-					gum::trace!("After answering runtime API request (votes)");
 				},
 				AllMessages::ChainApi(ChainApiMessage::BlockNumber(hash, tx)) => {
 					let block_num = self.headers.get(&hash).map(|header| header.number);
 					tx.send(Ok(block_num)).unwrap();
 				},
+				AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+					_new_leaf,
+					RuntimeApiRequest::CandidateEvents(tx),
+				)) => {
+					tx.send(Ok(candidate_events.clone())).unwrap();
+				},
+				AllMessages::RuntimeApi(RuntimeApiMessage::Request(
+					_new_leaf,
+					RuntimeApiRequest::FetchOnChainVotes(tx),
+				)) => {
+					//add some `BackedCandidates` or resolved disputes here as needed
+					tx.send(Ok(Some(ScrapedOnChainVotes {
+						session,
+						backing_validators_per_candidate: Vec::default(),
+						disputes: MultiDisputeStatementSet::default(),
+					})))
+					.unwrap();
+				},
+				AllMessages::ChainApi(ChainApiMessage::Ancestors { hash, k, response_channel }) => {
+					let target_header = self
+						.headers
+						.get(&hash)
+						.expect("The function is called for this block so it should exist");
+					let mut response = Vec::new();
+					for i in target_header.number.saturating_sub(k as u32)..target_header.number {
+						response.push(
+							self.block_num_to_header
+								.get(&i)
+								.expect("headers and block_num_to_header should always be in sync")
+								.clone(),
+						);
+					}
+					let _ = response_channel.send(Ok(response));
+				},
 				msg => {
 					panic!("Received unexpected message in `handle_sync_queries`: {:?}", msg);
 				},