Merge branch 'master' into bernhard-integrate-dispute-finality

* master:
  enable disputes (#3478)
  Do not leak active head data in statement distribution (#3519)
  Fix TransactAsset Implementation (#3345)

Author: ordian

Cargo.lock

@@ -28,6 +28,7 @@ use polkadot_node_subsystem::{
 use kvdb::KeyValueDB;
 use parity_scale_codec::Error as CodecError;
 use futures::channel::oneshot;
+use futures::future::Either;
 use futures::prelude::*;
 
 use std::time::{UNIX_EPOCH, Duration,SystemTime};
@@ -244,7 +245,7 @@ impl Clock for SystemClock {
 
 /// The interval, in seconds to check for stagnant blocks.
 #[derive(Debug, Clone)]
-pub struct StagnantCheckInterval(Duration);
+pub struct StagnantCheckInterval(Option<Duration>);
 
 impl Default for StagnantCheckInterval {
 	fn default() -> Self {
@@ -255,28 +256,37 @@ impl Default for StagnantCheckInterval {
 		// between 2 validators is D + 5s.
 		const DEFAULT_STAGNANT_CHECK_INTERVAL: Duration = Duration::from_secs(5);
 
-		StagnantCheckInterval(DEFAULT_STAGNANT_CHECK_INTERVAL)
+		StagnantCheckInterval(Some(DEFAULT_STAGNANT_CHECK_INTERVAL))
 	}
 }
 
 impl StagnantCheckInterval {
 	/// Create a new stagnant-check interval wrapping the given duration.
 	pub fn new(interval: Duration) -> Self {
-		StagnantCheckInterval(interval)
+		StagnantCheckInterval(Some(interval))
 	}
 
-	fn timeout_stream(&self) -> impl Stream<Item = ()> {
-		let interval = self.0;
-		let mut delay = futures_timer::Delay::new(interval);
+	/// Create a `StagnantCheckInterval` which never triggers.
+	pub fn never() -> Self {
+		StagnantCheckInterval(None)
+	}
 
-		futures::stream::poll_fn(move |cx| {
-			let poll = delay.poll_unpin(cx);
-			if poll.is_ready() {
-				delay.reset(interval)
-			}
+	fn timeout_stream(&self) -> impl Stream<Item = ()> {
+		match self.0 {
+			Some(interval) => Either::Left({
+				let mut delay = futures_timer::Delay::new(interval);
+
+				futures::stream::poll_fn(move |cx| {
+					let poll = delay.poll_unpin(cx);
+					if poll.is_ready() {
+						delay.reset(interval)
+					}
 
-			poll.map(Some)
-		})
+					poll.map(Some)
+				})
+			}),
+			None => Either::Right(futures::stream::pending()),
+		}
 	}
 }
 

node/core/chain-selection/src/lib.rs

@@ -27,7 +27,7 @@ use polkadot_cli::{
 	create_default_subsystems,
 	service::{
 		AuthorityDiscoveryApi, AuxStore, BabeApi, Block, Error, HeaderBackend, Overseer,
-		OverseerGen, OverseerGenArgs, Handle, ParachainHost, ProvideRuntimeApi,
+		OverseerGen, OverseerGenArgs, OverseerHandle, ParachainHost, ProvideRuntimeApi,
 		SpawnNamed,
 	},
 	Cli,
@@ -73,7 +73,7 @@ impl OverseerGen for BehaveMaleficient {
 	fn generate<'a, Spawner, RuntimeClient>(
 		&self,
 		args: OverseerGenArgs<'a, Spawner, RuntimeClient>,
-	) -> Result<(Overseer<Spawner, Arc<RuntimeClient>>, Handle), Error>
+	) -> Result<(Overseer<Spawner, Arc<RuntimeClient>>, OverseerHandle), Error>
 	where
 		RuntimeClient: 'static + ProvideRuntimeApi<Block> + HeaderBackend<Block> + AuxStore,
 		RuntimeClient::Api: ParachainHost<Block> + BabeApi<Block> + AuthorityDiscoveryApi<Block>,

node/malus/src/variant-a.rs

@@ -1847,6 +1847,16 @@ impl StatementDistribution {
 			FromOverseer::Signal(OverseerSignal::ActiveLeaves(ActiveLeavesUpdate { activated, deactivated })) => {
 				let _timer = metrics.time_active_leaves_update();
 
+				for deactivated in deactivated {
+					if active_heads.remove(&deactivated).is_some() {
+						tracing::trace!(
+							target: LOG_TARGET,
+							hash = ?deactivated,
+							"Deactivating leaf",
+						);
+					}
+				}
+
 				for activated in activated {
 					let relay_parent = activated.hash;
 					let span = PerLeafSpan::new(activated.span, "statement-distribution");
@@ -1862,18 +1872,6 @@ impl StatementDistribution {
 
 					active_heads.entry(relay_parent)
 						.or_insert(ActiveHeadData::new(session_info.validators.clone(), session_index, span));
-
-					active_heads.retain(|h, _| {
-						let live = !deactivated.contains(h);
-						if !live {
-							tracing::trace!(
-								target: LOG_TARGET,
-								hash = ?h,
-								"Deactivating leaf",
-							);
-						}
-						live
-					});
 				}
 			}
 			FromOverseer::Signal(OverseerSignal::BlockFinalized(..)) => {

node/network/statement-distribution/src/lib.rs

@@ -10,6 +10,7 @@ client = { package = "sc-client-api", git = "https://github.com/paritytech/subst
 sp-api = { git = "https://github.com/paritytech/substrate", branch = "master" }
 futures = "0.3.15"
 futures-timer = "3.0.2"
+parking_lot = "0.11.1"
 polkadot-node-network-protocol = { path = "../network/protocol" }
 polkadot-node-primitives = { path = "../primitives" }
 polkadot-node-subsystem-types = { path = "../subsystem-types" }

node/overseer/Cargo.toml

@@ -174,7 +174,7 @@ fn main() {
 			.replace_candidate_backing(Subsystem1)
 			;
 
-		let (overseer, _handler) = Overseer::new(
+		let (overseer, _handle) = Overseer::new(
 			vec![],
 			all_subsystems,
 			None,

node/overseer/examples/minimal-example.rs

@@ -123,7 +123,7 @@ impl SpawnNamed for DummySpawner{
 struct DummyCtx;
 
 fn main() {
-	let (overseer, _handler): (Xxx<_>, _) = Xxx::builder()
+	let (overseer, _handle): (Xxx<_>, _) = Xxx::builder()
 		.sub0(AwesomeSubSys::default())
 		.plinkos(GoblinTower::default())
 		.i_like_pi(::std::f64::consts::PI)