Update futures and tokio for browser light client (#673)

* Make availability-store compile for WASM * Use --manifest-path instead * Make validation work on wasm! * Switch to Spawn trait * Migrate validation to std futures * Migrate network to std futures * Final changes to validation * Tidy up network * Tidy up validation * Switch branch * Migrate service * Get polkadot to compile via wasm! * Add browser-demo * Add initial browser file * Add browser-demo * Tidy * Temp switch back to substrate/master * tidy * Fix wasm build * Re-add release flag * Switch to polkadot-master * Revert cli tokio version to avoid libp2p panic * Update tokio version * Fix availability store tests * Fix validation tests * Remove futures01 from availability-store * Fix network tests * Small changes * Fix collator * Fix typo * Revert removal of tokio_executor that causes tokio version mismatch panic * Fix adder test parachain * Revert "Revert removal of tokio_executor that causes tokio version mismatch panic" This reverts commit cfeb50c01d8df5e209483406a711e64761b44ae9. * Update availability-store/src/worker.rs Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Update network/src/lib.rs Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Update network/src/lib.rs Co-Authored-By: Pierre Krieger <pierre.krieger1708@gmail.com> * Box pin changes * Asyncify network functions * Clean up browser validation worker error * Fix av store test * Nits * Fix validation test * Switch favicon * Fix validation test again * Revert "Asyncify network functions" This reverts commit f20ae6548dc482cb1e75bc80641cfe55c6131a53. * Add async blocks back in
2026-05-31 15:41:02 +00:00 · 2019-12-10 11:58:22 +01:00
parent df3ea965e7
commit 25aa988f9b
34 changed files with 826 additions and 444 deletions
@@ -5,16 +5,13 @@ authors = ["Parity Technologies <admin@parity.io>"]
 edition = "2018"

 [dependencies]
-futures = "0.1.17"
-futures03 = { package = "futures", version = "0.3.1", features = ["compat"] }
+futures = "0.3.1"
 futures-timer = "2.0"
-async-std = { version = "1.0.1", features = ["unstable"] }
 parking_lot = "0.9.0"
-tokio = "0.1.22"
+tokio = { version = "0.2.4", features = ["rt-core", "blocking"] }
 derive_more = "0.14.1"
 log = "0.4.8"
 exit-future = "0.2.0"
-tokio-executor = { version = "0.2.0-alpha.6", features = ["blocking"] }
 codec = { package = "parity-scale-codec", version = "1.1.0", default-features = false, features = ["derive"] }
 availability_store = { package = "polkadot-availability-store", path = "../availability-store" }
 parachain = { package = "polkadot-parachain", path = "../parachain" }
@@ -23,15 +23,14 @@
 /// such as candidate verification while performing event-driven work
 /// on a local event loop.

-use std::{thread, time::{Duration, Instant}, sync::Arc};
+use std::{thread, time::Duration, sync::Arc};

 use client::{BlockchainEvents, BlockBody};
 use sp_blockchain::HeaderBackend;
 use block_builder::BlockBuilderApi;
 use consensus::SelectChain;
 use futures::prelude::*;
-use futures03::{TryStreamExt as _, StreamExt as _, FutureExt as _, TryFutureExt as _};
-use log::error;
+use futures::{future::{ready, select}, task::{Spawn, SpawnExt}};
 use polkadot_primitives::Block;
 use polkadot_primitives::parachain::ParachainHost;
 use runtime_primitives::traits::{ProvideRuntimeApi};
@@ -39,12 +38,12 @@ use babe_primitives::BabeApi;
 use keystore::KeyStorePtr;
 use sp_api::ApiExt;

-use tokio::{timer::Interval, runtime::current_thread::Runtime as LocalRuntime};
-use log::{warn, debug};
+use tokio::{runtime::Runtime as LocalRuntime};
+use log::{warn, error};

 use super::{Network, Collators};

-type TaskExecutor = Arc<dyn futures::future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>> + Send + Sync>;
+type TaskExecutor = Arc<dyn Spawn + Send + Sync>;

 /// Parachain candidate attestation service handle.
 pub(crate) struct ServiceHandle {
@@ -62,8 +61,8 @@ pub(crate) fn start<C, N, P, SC>(
 	max_block_data_size: Option<u64>,
 ) -> ServiceHandle
 	where
-		C: Collators + Send + Sync + 'static,
-		<C::Collation as IntoFuture>::Future: Send + 'static,
+		C: Collators + Send + Sync + Unpin + 'static,
+		C::Collation: Send + Unpin + 'static,
 		P: BlockchainEvents<Block> + BlockBody<Block>,
 		P: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync + 'static,
 		P::Api: ParachainHost<Block> +
@@ -72,10 +71,9 @@ pub(crate) fn start<C, N, P, SC>(
 			ApiExt<Block, Error = sp_blockchain::Error>,
 		N: Network + Send + Sync + 'static,
 		N::TableRouter: Send + 'static,
-		<N::BuildTableRouter as IntoFuture>::Future: Send + 'static,
+		N::BuildTableRouter: Send + Unpin + 'static,
 		SC: SelectChain<Block> + 'static,
 {
-	const TIMER_DELAY: Duration = Duration::from_secs(5);
 	const TIMER_INTERVAL: Duration = Duration::from_secs(30);

 	let (signal, exit) = ::exit_future::signal();
@@ -87,8 +85,7 @@ pub(crate) fn start<C, N, P, SC>(

 			let keystore = keystore.clone();

-			client.import_notification_stream()
-				.map(|v| Ok::<_, ()>(v)).compat()
+			let notifications = client.import_notification_stream()
 				.for_each(move |notification| {
 					let parent_hash = notification.hash;
 					if notification.is_new_best {
@@ -105,43 +102,35 @@ pub(crate) fn start<C, N, P, SC>(
 							);
 						}
 					}
-					Ok(())
-				})
-				.select(exit.clone().unit_error().compat())
-				.then(|_| Ok(()))
+					ready(())
+				});
+
+			select(notifications, exit.clone())
 		};

 		let prune_old_sessions = {
 			let select_chain = select_chain.clone();
-			let interval = Interval::new(
-				Instant::now() + TIMER_DELAY,
-				TIMER_INTERVAL,
-			);
-
-			interval
+			let interval = crate::interval(TIMER_INTERVAL)
 				.for_each(move |_| match select_chain.leaves() {
 					Ok(leaves) => {
 						parachain_validation.retain(|h| leaves.contains(h));
-						Ok(())
+						ready(())
 					}
 					Err(e) => {
 						warn!("Error fetching leaves from client: {:?}", e);
-						Ok(())
+						ready(())
 					}
-				})
-				.map_err(|e| warn!("Timer error {:?}", e))
-				.select(exit.clone().unit_error().compat())
-				.then(|_| Ok(()))
+				});
+
+			select(interval, exit.clone()).map(|_| ())
 		};

 		runtime.spawn(notifications);
-		if let Err(_) = thread_pool.execute(Box::new(prune_old_sessions)) {
+		if let Err(_) = thread_pool.spawn(prune_old_sessions) {
 			error!("Failed to spawn old sessions pruning task");
 		}

-		if let Err(e) = runtime.block_on(exit.unit_error().compat()) {
-			debug!("BFT event loop error {:?}", e);
-		}
+		runtime.block_on(exit);
 	});

 	ServiceHandle {
@@ -32,6 +32,8 @@ use parachain::{wasm_executor::{self, ExternalitiesError, ExecutionMode}, Messag
 use trie::TrieConfiguration;
 use futures::prelude::*;
 use log::debug;
+use std::task::{Poll, Context};
+use std::pin::Pin;

 /// Encapsulates connections to collators and allows collation on any parachain.
 ///
@@ -40,7 +42,7 @@ pub trait Collators: Clone {
 	/// Errors when producing collations.
 	type Error: std::fmt::Debug;
 	/// A full collation.
-	type Collation: IntoFuture<Item=Collation,Error=Self::Error>;
+	type Collation: Future<Output=Result<Collation, Self::Error>>;

 	/// Collate on a specific parachain, building on a given relay chain parent hash.
 	///
@@ -63,7 +65,7 @@ pub struct CollationFetch<C: Collators, P> {
 	relay_parent_hash: Hash,
 	relay_parent: BlockId,
 	collators: C,
-	live_fetch: Option<<C::Collation as IntoFuture>::Future>,
+	live_fetch: Option<C::Collation>,
 	client: Arc<P>,
 	max_block_data_size: Option<u64>,
 }
@@ -99,41 +101,50 @@ impl<C: Collators, P> CollationFetch<C, P> {
 	}
 }

-impl<C: Collators, P: ProvideRuntimeApi> Future for CollationFetch<C, P>
-	where P::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
+impl<C, P> Future for CollationFetch<C, P>
+	where
+		P::Api: ParachainHost<Block, Error = sp_blockchain::Error>,
+		C: Collators + Unpin,
+		P: ProvideRuntimeApi,
+		<C as Collators>::Collation: Unpin,
 {
-	type Item = (Collation, OutgoingMessages, Balance);
-	type Error = C::Error;
+	type Output = Result<(Collation, OutgoingMessages, Balance),C::Error>;
+
+	fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+		let this = Pin::into_inner(self);

-	fn poll(&mut self) -> Poll<(Collation, OutgoingMessages, Balance), C::Error> {
 		loop {
 			let collation = {
-				let parachain = self.parachain.clone();
-				let (r, c)  = (self.relay_parent_hash, &self.collators);
-				let poll = self.live_fetch
-					.get_or_insert_with(move || c.collate(parachain, r).into_future())
-					.poll();
+				let parachain = this.parachain.clone();
+				let (r, c)  = (this.relay_parent_hash, &this.collators);

-				futures::try_ready!(poll)
+				let future = this.live_fetch
+					.get_or_insert_with(move || c.collate(parachain, r));
+
+				match Pin::new(future).poll(cx) {
+					Poll::Ready(Ok(c)) => c,
+					Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
+					Poll::Pending => return Poll::Pending
+				}
 			};

 			let res = validate_collation(
-				&*self.client,
-				&self.relay_parent,
+				&*this.client,
+				&this.relay_parent,
 				&collation,
-				self.max_block_data_size,
+				this.max_block_data_size,
 			);

 			match res {
 				Ok((messages, fees)) => {
-					return Ok(Async::Ready((collation, messages, fees)))
+					return Poll::Ready(Ok((collation, messages, fees)))
 				}
 				Err(e) => {
 					debug!("Failed to validate parachain due to API error: {}", e);

 					// just continue if we got a bad collation or failed to validate
-					self.live_fetch = None;
-					self.collators.note_bad_collator(collation.info.collator)
+					this.live_fetch = None;
+					this.collators.note_bad_collator(collation.info.collator)
 				}
 			}
 		}
@@ -46,9 +46,7 @@ pub enum Error {
 	Timer(std::io::Error),
 	#[display(fmt = "Failed to compute deadline of now + {:?}", _0)]
 	DeadlineComputeFailure(std::time::Duration),
-	/// Unable to dispatch agreement future
-	#[display(fmt = "Unable to dispatch agreement future: {:?}", _0)]
-	Executor(futures::future::ExecuteErrorKind),
+	Join(tokio::task::JoinError)
 }

 impl std::error::Error for Error {
@@ -56,12 +56,11 @@ use polkadot_primitives::parachain::{
 use primitives::Pair;
 use runtime_primitives::traits::{ProvideRuntimeApi, DigestFor};
 use futures_timer::Delay;
-use async_std::stream::{interval, Interval};
 use txpool_api::{TransactionPool, InPoolTransaction};

 use attestation_service::ServiceHandle;
 use futures::prelude::*;
-use futures03::{future::{self, Either}, FutureExt, StreamExt, TryFutureExt};
+use futures::{future::{self, Either, select, ready}, stream::unfold, task::{Spawn, SpawnExt}};
 use collation::CollationFetch;
 use dynamic_inclusion::DynamicInclusion;
 use inherents::InherentData;
@@ -70,10 +69,13 @@ use log::{info, debug, warn, trace, error};
 use keystore::KeyStorePtr;
 use sp_api::ApiExt;

-type TaskExecutor =
-	Arc<
-		dyn futures::future::Executor<Box<dyn Future<Item = (), Error = ()> + Send>>
-	+ Send + Sync>;
+type TaskExecutor = Arc<dyn Spawn + Send + Sync>;
+
+fn interval(duration: Duration) -> impl Stream<Item=()> + Send + Unpin {
+	unfold((), move |_| {
+		futures_timer::Delay::new(duration).map(|_| Some(((), ())))
+	}).map(drop)
+}

 pub use self::collation::{
 	validate_collation, validate_incoming, message_queue_root, egress_roots, Collators,
@@ -84,6 +86,8 @@ pub use self::shared_table::{
 	SharedTable, ParachainWork, PrimedParachainWork, Validated, Statement, SignedStatement,
 	GenericStatement,
 };
+
+#[cfg(not(target_os = "unknown"))]
 pub use parachain::wasm_executor::{run_worker as run_validation_worker};

 mod attestation_service;
@@ -107,7 +111,7 @@ pub trait TableRouter: Clone {
 	/// Errors when fetching data from the network.
 	type Error: std::fmt::Debug;
 	/// Future that resolves when candidate data is fetched.
-	type FetchValidationProof: IntoFuture<Item=PoVBlock,Error=Self::Error>;
+	type FetchValidationProof: Future<Output=Result<PoVBlock, Self::Error>>;

 	/// Call with local candidate data. This will make the data available on the network,
 	/// and sign, import, and broadcast a statement about the candidate.
@@ -134,7 +138,7 @@ pub trait Network {

 	/// The future used for asynchronously building the table router.
 	/// This should not fail.
-	type BuildTableRouter: IntoFuture<Item=Self::TableRouter,Error=Self::Error>;
+	type BuildTableRouter: Future<Output=Result<Self::TableRouter,Self::Error>>;

 	/// Instantiate a table router using the given shared table.
 	/// Also pass through any outgoing messages to be broadcast to peers.
@@ -273,13 +277,13 @@ struct ParachainValidation<C, N, P> {
 }

 impl<C, N, P> ParachainValidation<C, N, P> where
-	C: Collators + Send + 'static,
+	C: Collators + Send + Unpin + 'static,
 	N: Network,
 	P: ProvideRuntimeApi + HeaderBackend<Block> + BlockBody<Block> + Send + Sync + 'static,
 	P::Api: ParachainHost<Block> + BlockBuilderApi<Block> + ApiExt<Block, Error = sp_blockchain::Error>,
-	<C::Collation as IntoFuture>::Future: Send + 'static,
+	C::Collation: Send + Unpin + 'static,
 	N::TableRouter: Send + 'static,
-	<N::BuildTableRouter as IntoFuture>::Future: Send + 'static,
+	N::BuildTableRouter: Unpin + Send + 'static,
 {
 	/// Get an attestation table for given parent hash.
 	///
@@ -400,7 +404,7 @@ impl<C, N, P> ParachainValidation<C, N, P> where
 				max_block_data_size,
 			);

-			collation_work.then(move |result| match result {
+			collation_work.map(move |result| match result {
 				Ok((collation, outgoing_targeted, fees_charged)) => {
 					match produce_receipt_and_chunks(
 						authorities_num,
@@ -427,10 +431,9 @@ impl<C, N, P> ParachainValidation<C, N, P> where
 							}
 							.unit_error()
 							.boxed()
-							.compat()
 							.then(move |_| {
 								router.local_collation(collation, receipt, outgoing_targeted, (local_id, &chunks));
-								Ok(())
+								ready(())
 							});


@@ -449,18 +452,16 @@ impl<C, N, P> ParachainValidation<C, N, P> where
 			})
 		};

-		let cancellable_work = build_router
-			.into_future()
+		let router = build_router
+			.map_ok(with_router)
 			.map_err(|e| {
 				warn!(target: "validation" , "Failed to build table router: {:?}", e);
-			})
-			.and_then(with_router)
-			.then(|_| Ok(()))
-			.select(exit.unit_error().compat())
-			.then(|_| Ok(()));
+			});
+
+		let cancellable_work = select(exit, router).map(drop);

 		// spawn onto thread pool.
-		if self.handle.execute(Box::new(cancellable_work)).is_err() {
+		if self.handle.spawn(cancellable_work).is_err() {
 			error!("Failed to spawn cancellable work task");
 		}
 	}
@@ -485,8 +486,8 @@ pub struct ProposerFactory<C, N, P, SC, TxPool: TransactionPool> {
 }

 impl<C, N, P, SC, TxPool> ProposerFactory<C, N, P, SC, TxPool> where
-	C: Collators + Send + Sync + 'static,
-	<C::Collation as IntoFuture>::Future: Send + 'static,
+	C: Collators + Send + Sync + Unpin + 'static,
+	C::Collation: Send + Unpin + 'static,
 	P: BlockchainEvents<Block> + BlockBody<Block>,
 	P: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync + 'static,
 	P::Api: ParachainHost<Block> +
@@ -495,7 +496,7 @@ impl<C, N, P, SC, TxPool> ProposerFactory<C, N, P, SC, TxPool> where
 		ApiExt<Block, Error = sp_blockchain::Error>,
 	N: Network + Send + Sync + 'static,
 	N::TableRouter: Send + 'static,
-	<N::BuildTableRouter as IntoFuture>::Future: Send + 'static,
+	N::BuildTableRouter: Send + Unpin + 'static,
 	TxPool: TransactionPool,
 	SC: SelectChain<Block> + 'static,
 {
@@ -543,7 +544,7 @@ impl<C, N, P, SC, TxPool> ProposerFactory<C, N, P, SC, TxPool> where
 }

 impl<C, N, P, SC, TxPool> consensus::Environment<Block> for ProposerFactory<C, N, P, SC, TxPool> where
-	C: Collators + Send + 'static,
+	C: Collators + Send + Unpin + 'static,
 	N: Network,
 	TxPool: TransactionPool<Block=Block> + 'static,
 	P: ProvideRuntimeApi + HeaderBackend<Block> + BlockBody<Block> + Send + Sync + 'static,
@@ -551,9 +552,9 @@ impl<C, N, P, SC, TxPool> consensus::Environment<Block> for ProposerFactory<C, N
 		BlockBuilderApi<Block> +
 		BabeApi<Block> +
 		ApiExt<Block, Error = sp_blockchain::Error>,
-	<C::Collation as IntoFuture>::Future: Send + 'static,
+	C::Collation: Send + Unpin + 'static,
 	N::TableRouter: Send + 'static,
-	<N::BuildTableRouter as IntoFuture>::Future: Send + 'static,
+	N::BuildTableRouter: Send + Unpin + 'static,
 	SC: SelectChain<Block>,
 {
 	type Proposer = Proposer<P, TxPool>;
@@ -649,7 +650,7 @@ impl<C, TxPool> consensus::Proposer<Block> for Proposer<C, TxPool> where

 		let timing = ProposalTiming {
 			minimum: delay_future,
-			attempt_propose: interval(ATTEMPT_PROPOSE_EVERY),
+			attempt_propose: Box::new(interval(ATTEMPT_PROPOSE_EVERY)),
 			enough_candidates: Delay::new(enough_candidates),
 			dynamic_inclusion,
 			last_included: initial_included,
@@ -690,7 +691,7 @@ fn current_timestamp() -> u64 {

 struct ProposalTiming {
 	minimum: Option<Delay>,
-	attempt_propose: Interval,
+	attempt_propose: Box<dyn Stream<Item=()> + Send + Unpin>,
 	dynamic_inclusion: DynamicInclusion,
 	enough_candidates: Delay,
 	last_included: usize,
@@ -746,7 +747,7 @@ enum CreateProposalState<C: Send + Sync, TxPool> {
 	/// Represents the state when we switch from pending to fired.
 	Switching,
 	/// Block proposing has fired.
-	Fired(tokio_executor::blocking::Blocking<Result<Block, Error>>),
+	Fired(tokio::task::JoinHandle<Result<Block, Error>>),
 }

 /// Inner data of the create proposal.
@@ -858,7 +859,7 @@ impl<C, TxPool> CreateProposalData<C, TxPool> where
 	}
 }

-impl<C, TxPool> futures03::Future for CreateProposal<C, TxPool> where
+impl<C, TxPool> Future for CreateProposal<C, TxPool> where
 	TxPool: TransactionPool<Block=Block> + 'static,
 	C: ProvideRuntimeApi + HeaderBackend<Block> + Send + Sync + 'static,
 	C::Api: ParachainHost<Block> + BlockBuilderApi<Block> + ApiExt<Block, Error = sp_blockchain::Error>,
@@ -892,18 +893,22 @@ impl<C, TxPool> futures03::Future for CreateProposal<C, TxPool> where
 					thus Switching will never be reachable here; qed"
 				),
 			CreateProposalState::Fired(mut future) => {
-				let ret = Pin::new(&mut future).poll(cx);
+				let ret = Pin::new(&mut future)
+					.poll(cx)
+					.map(|res| res.map_err(Error::Join).and_then(|res| res));
 				self.state = CreateProposalState::Fired(future);
 				return ret
 			},
 		};

 		// 2. propose
-		let mut future = tokio_executor::blocking::run(move || {
+		let mut future = tokio::task::spawn_blocking(move || {
 			let proposed_candidates = data.table.proposed_set();
 			data.propose_with(proposed_candidates)
 		});
-		let polled = Pin::new(&mut future).poll(cx);
+		let polled = Pin::new(&mut future)
+			.poll(cx)
+			.map(|res| res.map_err(Error::Join).and_then(|res| res));
 		self.state = CreateProposalState::Fired(future);

 		polled
@@ -19,7 +19,9 @@
 use std::collections::HashMap;

 use futures::prelude::*;
-use futures::sync::oneshot;
+use futures::channel::oneshot;
+use std::pin::Pin;
+use std::task::{Poll, Context};

 use polkadot_primitives::Hash;

@@ -95,17 +97,17 @@ impl IncludabilitySender {
 pub struct Includable(oneshot::Receiver<()>);

 impl Future for Includable {
-	type Item = ();
-	type Error = oneshot::Canceled;
+	type Output = Result<(), oneshot::Canceled>;

-	fn poll(&mut self) -> Poll<(), oneshot::Canceled> {
-		self.0.poll()
+	fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<Self::Output> {
+		Pin::new(&mut Pin::into_inner(self).0).poll(cx)
 	}
 }

 #[cfg(test)]
 mod tests {
 	use super::*;
+	use futures::executor::block_on;

 	#[test]
 	fn it_works() {
@@ -132,6 +134,6 @@ mod tests {
 		sender.update_candidate(hash1, true);
 		assert!(sender.is_complete());

-		recv.wait().unwrap();
+		block_on(recv).unwrap();
 	}
 }
@@ -139,7 +139,7 @@ impl SharedTableInner {
 		statement: table::SignedStatement,
 		max_block_data_size: Option<u64>,
 	) -> Option<ParachainWork<
-		<R::FetchValidationProof as IntoFuture>::Future,
+		R::FetchValidationProof
 	>> {
 		let summary = self.table.import_statement(context, statement)?;
 		self.update_trackers(&summary.candidate, context);
@@ -172,7 +172,7 @@ impl SharedTableInner {
 					None
 				}
 				Some(candidate) => {
-					let fetch = router.fetch_pov_block(candidate).into_future();
+					let fetch = router.fetch_pov_block(candidate);

 					Some(Work {
 						candidate_receipt: candidate.clone(),
@@ -267,13 +267,13 @@ pub struct ParachainWork<Fetch> {
 	max_block_data_size: Option<u64>,
 }

-impl<Fetch: Future> ParachainWork<Fetch> {
+impl<Fetch: Future + Unpin> ParachainWork<Fetch> {
 	/// Prime the parachain work with an API reference for extracting
 	/// chain information.
 	pub fn prime<P: ProvideRuntimeApi>(self, api: Arc<P>)
 		-> PrimedParachainWork<
 			Fetch,
-			impl Send + FnMut(&BlockId, &PoVBlock, &CandidateReceipt) -> Result<(OutgoingMessages, ErasureChunk), ()>,
+			impl Send + FnMut(&BlockId, &PoVBlock, &CandidateReceipt) -> Result<(OutgoingMessages, ErasureChunk), ()> + Unpin,
 		>
 		where
 			P: Send + Sync + 'static,
@@ -326,14 +326,13 @@ pub struct PrimedParachainWork<Fetch, F> {

 impl<Fetch, F, Err> PrimedParachainWork<Fetch, F>
 	where
-		Fetch: Future<Item=PoVBlock,Error=Err>,
-		F: FnMut(&BlockId, &PoVBlock, &CandidateReceipt) -> Result<(OutgoingMessages, ErasureChunk), ()>,
+		Fetch: Future<Output=Result<PoVBlock,Err>> + Unpin,
+		F: FnMut(&BlockId, &PoVBlock, &CandidateReceipt) -> Result<(OutgoingMessages, ErasureChunk), ()> + Unpin,
 		Err: From<::std::io::Error>,
 {
 	pub async fn validate(mut self) -> Result<(Validated, Option<ErasureChunk>), Err> {
-		use futures03::compat::Future01CompatExt;
 		let candidate = &self.inner.work.candidate_receipt;
-		let pov_block = self.inner.work.fetch.compat().await?;
+		let pov_block = self.inner.work.fetch.await?;

 		let validation_res = (self.validate)(
 			&BlockId::hash(self.inner.relay_parent),
@@ -439,7 +438,7 @@ impl SharedTable {
 		router: &R,
 		statement: table::SignedStatement,
 	) -> Option<ParachainWork<
-		<R::FetchValidationProof as IntoFuture>::Future,
+		R::FetchValidationProof,
 	>> {
 		self.inner.lock().import_remote_statement(&*self.context, router, statement, self.max_block_data_size)
 	}
@@ -455,7 +454,7 @@ impl SharedTable {
 			R: TableRouter,
 			I: IntoIterator<Item=table::SignedStatement>,
 			U: ::std::iter::FromIterator<Option<ParachainWork<
-				<R::FetchValidationProof as IntoFuture>::Future,
+				R::FetchValidationProof,
 			>>>,
 	{
 		let mut inner = self.inner.lock();
@@ -575,8 +574,9 @@ mod tests {
 	use polkadot_primitives::parachain::{AvailableMessages, BlockData, ConsolidatedIngress, Collation};
 	use polkadot_erasure_coding::{self as erasure};
 	use availability_store::ProvideGossipMessages;
-
-	use futures::{future};
+	use futures::future;
+	use futures::executor::block_on;
+	use std::pin::Pin;

 	fn pov_block_with_data(data: Vec<u8>) -> PoVBlock {
 		PoVBlock {
@@ -592,8 +592,8 @@ mod tests {
 		fn gossip_messages_for(
 			&self,
 			_topic: Hash
-		) -> Box<dyn futures03::Stream<Item = (Hash, Hash, ErasureChunk)> + Unpin + Send> {
-			Box::new(futures03::stream::empty())
+		) -> Pin<Box<dyn futures::Stream<Item = (Hash, Hash, ErasureChunk)> + Send>> {
+			futures::stream::empty().boxed()
 		}

 		fn gossip_erasure_chunk(
@@ -609,7 +609,7 @@ mod tests {
 	struct DummyRouter;
 	impl TableRouter for DummyRouter {
 		type Error = ::std::io::Error;
-		type FetchValidationProof = future::FutureResult<PoVBlock,Self::Error>;
+		type FetchValidationProof = future::Ready<Result<PoVBlock,Self::Error>>;

 		fn local_collation(
 			&self,
@@ -766,7 +766,7 @@ mod tests {
 			n_validators as u32,
 		).unwrap();

-		let producer: ParachainWork<future::FutureResult<_, ::std::io::Error>> = ParachainWork {
+		let producer: ParachainWork<future::Ready<Result<_, ::std::io::Error>>> = ParachainWork {
 			work: Work {
 				candidate_receipt: candidate,
 				fetch: future::ok(pov_block.clone()),
@@ -777,7 +777,7 @@ mod tests {
 			max_block_data_size: None,
 		};

-		let validated = futures03::executor::block_on(producer.prime_with(|_, _, _| Ok((
+		let validated = block_on(producer.prime_with(|_, _, _| Ok((
 				OutgoingMessages { outgoing_messages: Vec::new() },
 				ErasureChunk {
 					chunk: vec![1, 2, 3],
@@ -841,7 +841,7 @@ mod tests {
 			max_block_data_size: None,
 		};

-		let validated = futures03::executor::block_on(producer.prime_with(|_, _, _| Ok((
+		let validated = block_on(producer.prime_with(|_, _, _| Ok((
 				OutgoingMessages { outgoing_messages: Vec::new() },
 				ErasureChunk {
 					chunk: chunks[local_index].clone(),