consensus: refactor aura and babe proposer (#3377)

substrate/Cargo.lock

@@ -4544,6 +4544,7 @@ dependencies = [
  "substrate-consensus-common 2.0.0",
  "substrate-inherents 2.0.0",
  "substrate-primitives 2.0.0",
+ "substrate-telemetry 2.0.0",
  "substrate-test-runtime-client 2.0.0",
 ]
 

substrate/core/consensus/aura/src/lib.rs

@@ -145,13 +145,13 @@ pub fn start_aura<B, C, SC, E, I, P, SO, Error, H>(
 	C: ProvideRuntimeApi + BlockOf + ProvideCache<B> + AuxStore + Send + Sync,
 	C::Api: AuraApi<B, AuthorityId<P>>,
 	SC: SelectChain<B>,
+	E: Environment<B, Error=Error> + Send + Sync + 'static,
 	E::Proposer: Proposer<B, Error=Error>,
-	<E::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
-	P: Pair + Send + Sync + 'static,
+	<E::Proposer as Proposer<B>>::Create: Unpin + Send,
+	P: Pair + Send + Sync,
 	P::Public: Hash + Member + Encode + Decode,
 	P::Signature: Hash + Member + Encode + Decode,
 	H: Header<Hash=B::Hash>,
-	E: Environment<B, Error=Error>,
 	I: BlockImport<B> + Send + Sync + 'static,
 	Error: ::std::error::Error + Send + From<::consensus_common::Error> + From<I::Error> + 'static,
 	SO: SyncOracle + Send + Sync + Clone,
@@ -189,143 +189,72 @@ struct AuraWorker<C, E, I, P, SO> {
 	_key_type: PhantomData<P>,
 }
 
-impl<H, B, C, E, I, P, Error, SO> SlotWorker<B> for AuraWorker<C, E, I, P, SO> where
+impl<H, B, C, E, I, P, Error, SO> slots::SimpleSlotWorker<B> for AuraWorker<C, E, I, P, SO> where
 	B: BlockT<Header=H>,
 	C: ProvideRuntimeApi + BlockOf + ProvideCache<B> + Sync,
 	C::Api: AuraApi<B, AuthorityId<P>>,
 	E: Environment<B, Error=Error>,
 	E::Proposer: Proposer<B, Error=Error>,
-	<E::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
+	<E::Proposer as Proposer<B>>::Create: Unpin + Send,
 	H: Header<Hash=B::Hash>,
 	I: BlockImport<B> + Send + Sync + 'static,
-	P: Pair + Send + Sync + 'static,
+	P: Pair + Send + Sync,
 	P::Public: Member + Encode + Decode + Hash,
 	P::Signature: Member + Encode + Decode + Hash + Debug,
 	SO: SyncOracle + Send + Clone,
 	Error: ::std::error::Error + Send + From<::consensus_common::Error> + From<I::Error> + 'static,
 {
-	type OnSlot = Pin<Box<dyn Future<Output = Result<(), consensus_common::Error>> + Send>>;
+	type EpochData = Vec<AuthorityId<P>>;
+	type Claim = P;
+	type SyncOracle = SO;
+	type Proposer = E::Proposer;
+	type BlockImport = I;
 
-	fn on_slot(
-		&mut self,
-		chain_head: B::Header,
-		slot_info: SlotInfo,
-	) -> Self::OnSlot {
-		let client = self.client.clone();
-		let block_import = self.block_import.clone();
+	fn logging_target(&self) -> &'static str {
+		"aura"
+	}
 
-		let (timestamp, slot_num, slot_duration) =
-			(slot_info.timestamp, slot_info.number, slot_info.duration);
+	fn block_import(&self) -> Arc<Mutex<Self::BlockImport>> {
+		self.block_import.clone()
+	}
 
-		let authorities = match authorities(client.as_ref(), &BlockId::Hash(chain_head.hash())) {
-			Ok(authorities) => authorities,
-			Err(e) => {
-				warn!("Unable to fetch authorities at block {:?}: {:?}", chain_head.hash(), e);
+	fn epoch_data(&self, block: &B::Hash) -> Result<Self::EpochData, consensus_common::Error> {
+		authorities(self.client.as_ref(), &BlockId::Hash(*block))
+	}
 
-				telemetry!(
-					CONSENSUS_WARN; "aura.unable_fetching_authorities";
-					"slot" => ?chain_head.hash(),
-					"err" => ?e,
-				);
-				return Box::pin(future::ready(Ok(())));
-			}
-		};
+	fn authorities_len(&self, epoch_data: &Self::EpochData) -> usize {
+		epoch_data.len()
+	}
 
-		if !self.force_authoring && self.sync_oracle.is_offline() && authorities.len() > 1 {
-			debug!(target: "aura", "Skipping proposal slot. Waiting for the network.");
-			telemetry!(
-				CONSENSUS_DEBUG;
-				"aura.skipping_proposal_slot";
-				"authorities_len" => authorities.len(),
-			);
-			return Box::pin(future::ready(Ok(())));
-		}
-		let maybe_author = slot_author::<P>(slot_num, &authorities);
-		let maybe_pair = maybe_author.and_then(|p|
-			self.keystore.as_ref().and_then(|k|
+	fn claim_slot(&self, slot_number: u64, epoch_data: &Self::EpochData) -> Option<Self::Claim> {
+		let expected_author = slot_author::<P>(slot_number, epoch_data);
+
+		expected_author.and_then(|p| {
+			self.keystore.as_ref().and_then(|k| {
 				k.read().key_pair_by_type::<P>(&p, app_crypto::key_types::AURA).ok()
-			)
-		);
-		let proposal_work = match maybe_pair {
-			None => return Box::pin(future::ready(Ok(()))),
-			Some(pair) => {
-				debug!(
-					target: "aura", "Starting authorship at slot {}; timestamp = {}",
-					slot_num,
-					timestamp,
-				);
-				telemetry!(CONSENSUS_DEBUG; "aura.starting_authorship";
-					"slot_num" => slot_num,
-					"timestamp" => timestamp,
-				);
+			})
+		})
+	}
 
-				// we are the slot author. make a block and sign it.
-				let mut proposer = match self.env.init(&chain_head) {
-					Ok(p) => p,
-					Err(e) => {
-						warn!("Unable to author block in slot {:?}: {:?}", slot_num, e);
-						telemetry!(CONSENSUS_WARN; "aura.unable_authoring_block";
-							"slot" => slot_num, "err" => ?e
-						);
-						return Box::pin(future::ready(Ok(())))
-					}
-				};
-
-				let remaining_duration = slot_info.remaining_duration();
-				// deadline our production to approx. the end of the
-				// slot
-				futures::future::select(
-					proposer.propose(
-						slot_info.inherent_data,
-						generic::Digest {
-							logs: vec![
-								<DigestItemFor<B> as CompatibleDigestItem<P>>::aura_pre_digest(slot_num),
-							],
-						},
-						remaining_duration,
-					).map_err(|e| consensus_common::Error::ClientImport(format!("{:?}", e)).into()),
-					Delay::new(remaining_duration)
-						.map_err(|err| consensus_common::Error::FaultyTimer(err).into())
-				).map(|v| match v {
-					futures::future::Either::Left((v, _)) => v.map(|v| (v, pair)),
-					futures::future::Either::Right((Ok(_), _)) =>
-						Err(consensus_common::Error::ClientImport("Timeout in the AuRa proposer".into())),
-					futures::future::Either::Right((Err(err), _)) => Err(err),
-				})
-			}
-		};
-
-		Box::pin(proposal_work.map_ok(move |(b, pair)| {
-			// minor hack since we don't have access to the timestamp
-			// that is actually set by the proposer.
-			let slot_after_building = SignedDuration::default().slot_now(slot_duration);
-			if slot_after_building != slot_num {
-				info!("Discarding proposal for slot {}; block production took too long", slot_num);
-				telemetry!(CONSENSUS_INFO; "aura.discarding_proposal_took_too_long";
-					"slot" => slot_num,
-				);
-				return
-			}
-
-			let (header, body) = b.deconstruct();
-			let pre_digest: Result<u64, String> = find_pre_digest::<B, P>(&header);
-			if let Err(e) = pre_digest {
-				error!(target: "aura", "FATAL ERROR: Invalid pre-digest: {}!", e);
-				return
-			} else {
-				trace!(target: "aura", "Got correct number of seals. Good!")
-			};
-
-			let header_num = header.number().clone();
-			let parent_hash = header.parent_hash().clone();
+	fn pre_digest_data(&self, slot_number: u64, _claim: &Self::Claim) -> Vec<sr_primitives::DigestItem<B::Hash>> {
+		vec![
+			<DigestItemFor<B> as CompatibleDigestItem<P>>::aura_pre_digest(slot_number),
+		]
+	}
 
+	fn import_block(&self) -> Box<dyn Fn(
+		B::Header,
+		&B::Hash,
+		Vec<B::Extrinsic>,
+		Self::Claim,
+	) -> consensus_common::BlockImportParams<B> + Send> {
+		Box::new(|header, header_hash, body, pair| {
 			// sign the pre-sealed hash of the block and then
 			// add it to a digest item.
-			let header_hash = header.hash();
 			let signature = pair.sign(header_hash.as_ref());
 			let signature_digest_item = <DigestItemFor<B> as CompatibleDigestItem<P>>::aura_seal(signature);
 
-			let import_block: BlockImportParams<B> = BlockImportParams {
+			BlockImportParams {
 				origin: BlockOrigin::Own,
 				header,
 				justification: None,
@@ -334,27 +263,44 @@ impl<H, B, C, E, I, P, Error, SO> SlotWorker<B> for AuraWorker<C, E, I, P, SO> w
 				finalized: false,
 				auxiliary: Vec::new(),
 				fork_choice: ForkChoiceStrategy::LongestChain,
-			};
-
-			info!("Pre-sealed block for proposal at {}. Hash now {:?}, previously {:?}.",
-					header_num,
-					import_block.post_header().hash(),
-					header_hash
-			);
-			telemetry!(CONSENSUS_INFO; "aura.pre_sealed_block";
-				"header_num" => ?header_num,
-				"hash_now" => ?import_block.post_header().hash(),
-				"hash_previously" => ?header_hash,
-			);
-
-			if let Err(e) = block_import.lock().import_block(import_block, Default::default()) {
-				warn!(target: "aura", "Error with block built on {:?}: {:?}", parent_hash, e);
-
-				telemetry!(CONSENSUS_WARN; "aura.err_with_block_built_on";
-					"hash" => ?parent_hash, "err" => ?e,
-				);
 			}
-		}))
+		})
+	}
+
+	fn force_authoring(&self) -> bool {
+		self.force_authoring
+	}
+
+	fn sync_oracle(&mut self) -> &mut Self::SyncOracle {
+		&mut self.sync_oracle
+	}
+
+	fn proposer(&mut self, block: &B::Header) -> Result<Self::Proposer, consensus_common::Error> {
+		self.env.init(block).map_err(|e| {
+			consensus_common::Error::ClientImport(format!("{:?}", e)).into()
+		})
+	}
+}
+
+impl<H, B: BlockT, C, E, I, P, Error, SO> SlotWorker<B> for AuraWorker<C, E, I, P, SO> where
+	B: BlockT<Header=H>,
+	C: ProvideRuntimeApi + BlockOf + ProvideCache<B> + Sync + Send,
+	C::Api: AuraApi<B, AuthorityId<P>>,
+	E: Environment<B, Error=Error> + Send + Sync,
+	E::Proposer: Proposer<B, Error=Error>,
+	<E::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
+	H: Header<Hash=B::Hash>,
+	I: BlockImport<B> + Send + Sync + 'static,
+	P: Pair + Send + Sync,
+	P::Public: Member + Encode + Decode + Hash,
+	P::Signature: Member + Encode + Decode + Hash + Debug,
+	SO: SyncOracle + Send + Sync + Clone,
+	Error: ::std::error::Error + Send + From<::consensus_common::Error> + From<I::Error> + 'static,
+{
+	type OnSlot = Pin<Box<dyn Future<Output = Result<(), consensus_common::Error>> + Send>>;
+
+	fn on_slot(&mut self, chain_head: B::Header, slot_info: SlotInfo) -> Self::OnSlot {
+		<Self as slots::SimpleSlotWorker<B>>::on_slot(self, chain_head, slot_info)
 	}
 }
 
@@ -384,7 +330,6 @@ fn find_pre_digest<B: BlockT, P: Pair>(header: &B::Header) -> Result<u64, String
 	pre_digest.ok_or_else(|| aura_err!("No AuRa pre-runtime digest found"))
 }
 
-
 /// check a header has been signed by the right key. If the slot is too far in the future, an error will be returned.
 /// if it's successful, returns the pre-header and the digest item containing the seal.
 ///

substrate/core/consensus/babe/src/lib.rs

@@ -186,10 +186,10 @@ pub fn start_babe<B, C, SC, E, I, SO, Error, H>(BabeParams {
 	C: ProvideRuntimeApi + ProvideCache<B> + ProvideUncles<B> + Send + Sync + 'static,
 	C::Api: BabeApi<B>,
 	SC: SelectChain<B> + 'static,
+	E: Environment<B, Error=Error> + Send + Sync,
 	E::Proposer: Proposer<B, Error=Error>,
 	<E::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
 	H: Header<Hash=B::Hash>,
-	E: Environment<B, Error=Error>,
 	I: BlockImport<B> + Send + Sync + 'static,
 	Error: std::error::Error + Send + From<::consensus_common::Error> + From<I::Error> + 'static,
 	SO: SyncOracle + Send + Sync + Clone,
@@ -229,155 +229,77 @@ struct BabeWorker<C, E, I, SO> {
 	keystore: KeyStorePtr,
 }
 
-impl<Hash, H, B, C, E, I, Error, SO> SlotWorker<B> for BabeWorker<C, E, I, SO> where
-	B: BlockT<Header=H, Hash=Hash>,
+impl<H, B, C, E, I, Error, SO> slots::SimpleSlotWorker<B> for BabeWorker<C, E, I, SO> where
+	B: BlockT<Header=H>,
 	C: ProvideRuntimeApi + ProvideCache<B>,
 	C::Api: BabeApi<B>,
 	E: Environment<B, Error=Error>,
 	E::Proposer: Proposer<B, Error=Error>,
 	<E::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
-	Hash: Debug + Eq + Copy + SimpleBitOps + Encode + Decode + Serialize +
-		for<'de> Deserialize<'de> + Debug + Default + AsRef<[u8]> + AsMut<[u8]> +
-		std::hash::Hash + Display + Send + Sync + 'static,
 	H: Header<Hash=B::Hash>,
 	I: BlockImport<B> + Send + Sync + 'static,
 	SO: SyncOracle + Send + Clone,
 	Error: std::error::Error + Send + From<::consensus_common::Error> + From<I::Error> + 'static,
 {
-	type OnSlot = Pin<Box<dyn Future<Output = Result<(), consensus_common::Error>> + Send>>;
+	type EpochData = Epoch;
+	type Claim = (VRFInOut, VRFProof, u32, AuthorityPair);
+	type SyncOracle = SO;
+	type Proposer = E::Proposer;
+	type BlockImport = I;
 
-	fn on_slot(
-		&mut self,
-		chain_head: B::Header,
-		slot_info: SlotInfo,
-	) -> Self::OnSlot {
-		let ref client = self.client;
-		let block_import = self.block_import.clone();
+	fn logging_target(&self) -> &'static str {
+		"babe"
+	}
 
-		let (timestamp, slot_number, slot_duration) =
-			(slot_info.timestamp, slot_info.number, slot_info.duration);
+	fn block_import(&self) -> Arc<Mutex<Self::BlockImport>> {
+		self.block_import.clone()
+	}
 
-		let epoch = match epoch(client.as_ref(), &BlockId::Hash(chain_head.hash())) {
-			Ok(authorities) => authorities,
-			Err(e) => {
-				error!(
-					target: "babe",
-					"Unable to fetch authorities at block {:?}: {:?}",
-					chain_head.hash(),
-					e
-				);
-				telemetry!(CONSENSUS_WARN; "babe.unable_fetching_authorities";
-					"slot" => ?chain_head.hash(), "err" => ?e
-				);
-				return Box::pin(future::ready(Ok(())));
-			}
-		};
+	fn epoch_data(&self, block: &B::Hash) -> Result<Self::EpochData, consensus_common::Error> {
+		epoch(self.client.as_ref(), &BlockId::Hash(*block))
+	}
 
-		let Epoch { ref authorities, .. } = epoch;
+	fn authorities_len(&self, epoch_data: &Self::EpochData) -> usize {
+		epoch_data.authorities.len()
+	}
 
-		if authorities.is_empty() {
-			error!(target: "babe", "No authorities at block {:?}", chain_head.hash());
-		}
-
-		if !self.force_authoring && self.sync_oracle.is_offline() && authorities.len() > 1 {
-			debug!(target: "babe", "Skipping proposal slot. Waiting for the network.");
-			telemetry!(CONSENSUS_DEBUG; "babe.skipping_proposal_slot";
-				"authorities_len" => authorities.len()
-			);
-			return Box::pin(future::ready(Ok(())));
-		}
-
-		let proposal_work = if let Some(claim) = claim_slot(
-			slot_info.number,
-			epoch,
+	fn claim_slot(&self, slot_number: u64, epoch_data: &Self::EpochData) -> Option<Self::Claim> {
+		claim_slot(
+			slot_number,
+			epoch_data,
 			self.c,
 			&self.keystore,
-		) {
-			let ((inout, vrf_proof, _batchable_proof), authority_index, key) = claim;
+		).map(|((inout, vrf_proof, _), authority_index, key)| {
+			(inout, vrf_proof, authority_index as u32, key)
+		})
+	}
 
-			debug!(
-				target: "babe", "Starting authorship at slot {}; timestamp = {}",
-				slot_number,
-				timestamp,
-			);
-			telemetry!(CONSENSUS_DEBUG; "babe.starting_authorship";
-				"slot_number" => slot_number, "timestamp" => timestamp
-			);
-
-			// we are the slot author. make a block and sign it.
-			let mut proposer = match self.env.init(&chain_head) {
-				Ok(p) => p,
-				Err(e) => {
-					warn!(target: "babe",
-						"Unable to author block in slot {:?}: {:?}",
-						slot_number,
-						e,
-					);
-					telemetry!(CONSENSUS_WARN; "babe.unable_authoring_block";
-						"slot" => slot_number, "err" => ?e
-					);
-					return Box::pin(future::ready(Ok(())))
-				}
-			};
-
-			let inherent_digest = BabePreDigest {
-				vrf_proof,
-				vrf_output: inout.to_output(),
-				authority_index: authority_index as u32,
-				slot_number,
-			};
-
-			// deadline our production to approx. the end of the slot
-			let remaining_duration = slot_info.remaining_duration();
-			futures::future::select(
-				proposer.propose(
-					slot_info.inherent_data,
-					generic::Digest {
-						logs: vec![
-							generic::DigestItem::babe_pre_digest(inherent_digest.clone()),
-						],
-					},
-					remaining_duration,
-				).map_err(|e| consensus_common::Error::ClientImport(format!("{:?}", e)).into()),
-				Delay::new(remaining_duration)
-					.map_err(|err| consensus_common::Error::FaultyTimer(err).into())
-			).map(|v| match v {
-				futures::future::Either::Left((v, _)) => v.map(|v| (v, key)),
-				futures::future::Either::Right((Ok(_), _)) =>
-					Err(consensus_common::Error::ClientImport("Timeout in the BaBe proposer".into())),
-				futures::future::Either::Right((Err(err), _)) => Err(err),
-			})
-		} else {
-			return Box::pin(future::ready(Ok(())));
+	fn pre_digest_data(&self, slot_number: u64, claim: &Self::Claim) -> Vec<sr_primitives::DigestItem<B::Hash>> {
+		let inherent_digest = BabePreDigest {
+			vrf_proof: claim.1.clone(),
+			vrf_output: claim.0.to_output(),
+			authority_index: claim.2,
+			slot_number,
 		};
 
-		Box::pin(proposal_work.map_ok(move |(b, key)| {
-			// minor hack since we don't have access to the timestamp
-			// that is actually set by the proposer.
-			let slot_after_building = SignedDuration::default().slot_now(slot_duration);
-			if slot_after_building != slot_number {
-				info!(
-					target: "babe",
-					"Discarding proposal for slot {}; block production took too long",
-					slot_number
-				);
-				telemetry!(CONSENSUS_INFO; "babe.discarding_proposal_took_too_long";
-					"slot" => slot_number
-				);
-				return;
-			}
-
-			let (header, body) = b.deconstruct();
-			let header_num = header.number().clone();
-			let parent_hash = header.parent_hash().clone();
+		vec![
+			<DigestItemFor<B> as CompatibleDigestItem>::babe_pre_digest(inherent_digest),
+		]
+	}
 
+	fn import_block(&self) -> Box<dyn Fn(
+		B::Header,
+		&B::Hash,
+		Vec<B::Extrinsic>,
+		Self::Claim,
+	) -> consensus_common::BlockImportParams<B> + Send> {
+		Box::new(|header, header_hash, body, (_, _, _, pair)| {
 			// sign the pre-sealed hash of the block and then
 			// add it to a digest item.
-			let header_hash = header.hash();
-			let signature = key.sign(header_hash.as_ref());
-			let signature_digest_item = DigestItemFor::<B>::babe_seal(signature);
+			let signature = pair.sign(header_hash.as_ref());
+			let signature_digest_item = <DigestItemFor<B> as CompatibleDigestItem>::babe_seal(signature);
 
-			let import_block = BlockImportParams::<B> {
+			BlockImportParams {
 				origin: BlockOrigin::Own,
 				header,
 				justification: None,
@@ -386,29 +308,41 @@ impl<Hash, H, B, C, E, I, Error, SO> SlotWorker<B> for BabeWorker<C, E, I, SO> w
 				finalized: false,
 				auxiliary: Vec::new(),
 				fork_choice: ForkChoiceStrategy::LongestChain,
-			};
-
-			info!(target: "babe",
-					"Pre-sealed block for proposal at {}. Hash now {:?}, previously {:?}.",
-					header_num,
-					import_block.post_header().hash(),
-					header_hash,
-			);
-
-			telemetry!(CONSENSUS_INFO; "babe.pre_sealed_block";
-				"header_num" => ?header_num,
-				"hash_now" => ?import_block.post_header().hash(),
-				"hash_previously" => ?header_hash,
-			);
-
-			if let Err(e) = block_import.lock().import_block(import_block, Default::default()) {
-				warn!(target: "babe", "Error with block built on {:?}: {:?}",
-						parent_hash, e);
-				telemetry!(CONSENSUS_WARN; "babe.err_with_block_built_on";
-					"hash" => ?parent_hash, "err" => ?e
-				);
 			}
-		}))
+		})
+	}
+
+	fn force_authoring(&self) -> bool {
+		self.force_authoring
+	}
+
+	fn sync_oracle(&mut self) -> &mut Self::SyncOracle {
+		&mut self.sync_oracle
+	}
+
+	fn proposer(&mut self, block: &B::Header) -> Result<Self::Proposer, consensus_common::Error> {
+		self.env.init(block).map_err(|e| {
+			consensus_common::Error::ClientImport(format!("{:?}", e)).into()
+		})
+	}
+}
+
+impl<H, B, C, E, I, Error, SO> SlotWorker<B> for BabeWorker<C, E, I, SO> where
+	B: BlockT<Header=H>,
+	C: ProvideRuntimeApi + ProvideCache<B> + Send + Sync,
+	C::Api: BabeApi<B>,
+	E: Environment<B, Error=Error> + Send + Sync,
+	E::Proposer: Proposer<B, Error=Error>,
+	<E::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
+	H: Header<Hash=B::Hash>,
+	I: BlockImport<B> + Send + Sync + 'static,
+	SO: SyncOracle + Send + Sync + Clone,
+	Error: std::error::Error + Send + From<::consensus_common::Error> + From<I::Error> + 'static,
+{
+	type OnSlot = Pin<Box<dyn Future<Output = Result<(), consensus_common::Error>> + Send>>;
+
+	fn on_slot(&mut self, chain_head: B::Header, slot_info: SlotInfo) -> Self::OnSlot {
+		<Self as slots::SimpleSlotWorker<B>>::on_slot(self, chain_head, slot_info)
 	}
 }
 
@@ -831,7 +765,7 @@ fn calculate_threshold(
 /// so it returns `Some(_)`.  Otherwise, it returns `None`.
 fn claim_slot(
 	slot_number: u64,
-	Epoch { ref authorities, ref randomness, epoch_index, .. }: Epoch,
+	Epoch { authorities, randomness, epoch_index, .. }: &Epoch,
 	c: (u64, u64),
 	keystore: &KeyStorePtr,
 ) -> Option<((VRFInOut, VRFProof, VRFProofBatchable), usize, AuthorityPair)> {
@@ -841,7 +775,7 @@ fn claim_slot(
 		.find_map(|(i, a)| {
 			keystore.key_pair::<AuthorityPair>(&a.0).ok().map(|kp| (kp, i))
 		})?;
-	let transcript = make_transcript(randomness, slot_number, epoch_index);
+	let transcript = make_transcript(randomness, slot_number, *epoch_index);
 
 	// Compute the threshold we will use.
 	//
@@ -1228,7 +1162,7 @@ pub mod test_helpers {
 
 		super::claim_slot(
 			slot_number,
-			epoch,
+			&epoch,
 			c,
 			keystore,
 		).map(|((inout, vrf_proof, _), authority_index, _)| {

substrate/core/consensus/babe/src/tests.rs

@@ -325,7 +325,7 @@ fn can_author_block() {
 		duration: 100,
 	};
 	loop {
-		match claim_slot(i, epoch.clone(), (3, 10), &keystore) {
+		match claim_slot(i, &epoch.clone(), (3, 10), &keystore) {
 			None => i += 1,
 			Some(s) => {
 				debug!(target: "babe", "Authored block {:?}", s.0);

substrate/core/consensus/slots/Cargo.toml

@@ -10,6 +10,7 @@ codec = { package = "parity-scale-codec", version = "1.0.0" }
 client = { package = "substrate-client", path = "../../client" }
 primitives = { package = "substrate-primitives", path = "../../primitives" }
 sr-primitives = {  path = "../../sr-primitives" }
+substrate-telemetry = { path = "../../telemetry" }
 consensus_common = { package = "substrate-consensus-common", path = "../common" }
 inherents = { package = "substrate-inherents", path = "../../inherents" }
 futures-preview = "=0.3.0-alpha.17"

substrate/core/consensus/slots/src/lib.rs

@@ -31,13 +31,16 @@ use slots::Slots;
 pub use aux_schema::{check_equivocation, MAX_SLOT_CAPACITY, PRUNING_BOUND};
 
 use codec::{Decode, Encode};
-use consensus_common::{SyncOracle, SelectChain};
+use consensus_common::{BlockImport, Proposer, SyncOracle, SelectChain};
 use futures::{prelude::*, future::{self, Either}};
+use futures_timer::Delay;
 use inherents::{InherentData, InherentDataProviders};
 use log::{debug, error, info, warn};
 use sr_primitives::generic::BlockId;
-use sr_primitives::traits::{ApiRef, Block as BlockT, ProvideRuntimeApi};
-use std::{fmt::Debug, ops::Deref};
+use sr_primitives::traits::{ApiRef, Block as BlockT, Header, ProvideRuntimeApi};
+use std::{fmt::Debug, ops::Deref, pin::Pin, sync::Arc};
+use substrate_telemetry::{telemetry, CONSENSUS_DEBUG, CONSENSUS_WARN, CONSENSUS_INFO};
+use parking_lot::Mutex;
 
 /// A worker that should be invoked at every new slot.
 pub trait SlotWorker<B: BlockT> {
@@ -49,6 +52,203 @@ pub trait SlotWorker<B: BlockT> {
 	fn on_slot(&mut self, chain_head: B::Header, slot_info: SlotInfo) -> Self::OnSlot;
 }
 
+/// A skeleton implementation for `SlotWorker` which tries to claim a slot at at
+/// its beginning and tries to produce a block if successfully claimed, timing
+/// out if block production takes too long.
+pub trait SimpleSlotWorker<B: BlockT> {
+	/// A handle to a `BlockImport`.
+	type BlockImport: BlockImport<B> + Send + 'static;
+
+	/// A handle to a `SyncOracle`.
+	type SyncOracle: SyncOracle;
+
+	/// The type of proposer to use to build blocks.
+	type Proposer: Proposer<B>;
+
+	/// Data associated with a slot claim.
+	type Claim: Send + 'static;
+
+	/// Epoch data necessary for authoring.
+	type EpochData;
+
+	/// The logging target to use when logging messages.
+	fn logging_target(&self) -> &'static str;
+
+	/// A handle to a `BlockImport`.
+	fn block_import(&self) -> Arc<Mutex<Self::BlockImport>>;
+
+	/// Returns the epoch data necessary for authoring.
+	fn epoch_data(&self, block: &B::Hash) -> Result<Self::EpochData, consensus_common::Error>;
+
+	/// Returns the number of authorities given the epoch data.
+	fn authorities_len(&self, epoch_data: &Self::EpochData) -> usize;
+
+	/// Tries to claim the given slot, returning an object with claim data if successful.
+	fn claim_slot(&self, slot_number: u64, epoch_data: &Self::EpochData) -> Option<Self::Claim>;
+
+	/// Return the pre digest data to include in a block authored with the given claim.
+	fn pre_digest_data(&self, slot_number: u64, claim: &Self::Claim) -> Vec<sr_primitives::DigestItem<B::Hash>>;
+
+	/// Returns a function which produces a `BlockImportParams`.
+	fn import_block(&self) -> Box<dyn Fn(
+		B::Header,
+		&B::Hash,
+		Vec<B::Extrinsic>,
+		Self::Claim,
+	) -> consensus_common::BlockImportParams<B> + Send>;
+
+	/// Whether to force authoring if offline.
+	fn force_authoring(&self) -> bool;
+
+	/// Returns a handle to a `SyncOracle`.
+	fn sync_oracle(&mut self) -> &mut Self::SyncOracle;
+
+	/// Returns a `Proposer` to author on top of the given block.
+	fn proposer(&mut self, block: &B::Header) -> Result<Self::Proposer, consensus_common::Error>;
+
+	/// Implements the `on_slot` functionality from `SlotWorker`.
+	fn on_slot(&mut self, chain_head: B::Header, slot_info: SlotInfo)
+		-> Pin<Box<dyn Future<Output = Result<(), consensus_common::Error>> + Send>> where
+		Self: Send + Sync,
+		<Self::Proposer as Proposer<B>>::Create: Unpin + Send + 'static,
+	{
+		let (timestamp, slot_number, slot_duration) =
+			(slot_info.timestamp, slot_info.number, slot_info.duration);
+
+		let epoch_data = match self.epoch_data(&chain_head.hash()) {
+			Ok(epoch_data) => epoch_data,
+			Err(err) => {
+				warn!("Unable to fetch epoch data at block {:?}: {:?}", chain_head.hash(), err);
+
+				telemetry!(
+					CONSENSUS_WARN; "slots.unable_fetching_authorities";
+					"slot" => ?chain_head.hash(),
+					"err" => ?err,
+				);
+
+				return Box::pin(future::ready(Ok(())));
+			}
+		};
+
+		let authorities_len = self.authorities_len(&epoch_data);
+
+		if !self.force_authoring() && self.sync_oracle().is_offline() && authorities_len > 1 {
+			debug!(target: self.logging_target(), "Skipping proposal slot. Waiting for the network.");
+			telemetry!(
+				CONSENSUS_DEBUG;
+				"slots.skipping_proposal_slot";
+				"authorities_len" => authorities_len,
+			);
+
+			return Box::pin(future::ready(Ok(())));
+		}
+
+		let claim = match self.claim_slot(slot_number, &epoch_data) {
+			None => return Box::pin(future::ready(Ok(()))),
+			Some(claim) => claim,
+		};
+
+		debug!(
+			target: self.logging_target(), "Starting authorship at slot {}; timestamp = {}",
+			slot_number,
+			timestamp,
+		);
+
+		telemetry!(CONSENSUS_DEBUG; "slots.starting_authorship";
+			"slot_num" => slot_number,
+			"timestamp" => timestamp,
+		);
+
+		let mut proposer = match self.proposer(&chain_head) {
+			Ok(proposer) => proposer,
+			Err(err) => {
+				warn!("Unable to author block in slot {:?}: {:?}", slot_number, err);
+
+				telemetry!(CONSENSUS_WARN; "slots.unable_authoring_block";
+					"slot" => slot_number, "err" => ?err
+				);
+
+				return Box::pin(future::ready(Ok(())));
+			},
+		};
+
+		let remaining_duration = slot_info.remaining_duration();
+		let logs = self.pre_digest_data(slot_number, &claim);
+
+		// deadline our production to approx. the end of the slot
+		let proposal_work = futures::future::select(
+			proposer.propose(
+				slot_info.inherent_data,
+				sr_primitives::generic::Digest {
+					logs,
+				},
+				remaining_duration,
+			).map_err(|e| consensus_common::Error::ClientImport(format!("{:?}", e)).into()),
+			Delay::new(remaining_duration)
+				.map_err(|err| consensus_common::Error::FaultyTimer(err).into())
+		).map(|v| match v {
+			futures::future::Either::Left((b, _)) => b.map(|b| (b, claim)),
+			futures::future::Either::Right((Ok(_), _)) =>
+				Err(consensus_common::Error::ClientImport("Timeout in the Slots proposer".into())),
+			futures::future::Either::Right((Err(err), _)) => Err(err),
+		});
+
+		let import_block = self.import_block();
+		let block_import = self.block_import();
+		let logging_target = self.logging_target();
+
+		Box::pin(proposal_work.map_ok(move |(block, claim)| {
+			// minor hack since we don't have access to the timestamp
+			// that is actually set by the proposer.
+			let slot_after_building = SignedDuration::default().slot_now(slot_duration);
+			if slot_after_building != slot_number {
+				info!("Discarding proposal for slot {}; block production took too long", slot_number);
+				telemetry!(CONSENSUS_INFO; "slots.discarding_proposal_took_too_long";
+					"slot" => slot_number,
+				);
+
+				return;
+			}
+
+			let (header, body) = block.deconstruct();
+			let header_num = header.number().clone();
+			let header_hash = header.hash();
+			let parent_hash = header.parent_hash().clone();
+
+			let import_block = import_block(
+				header,
+				&header_hash,
+				body,
+				claim,
+			);
+
+			info!("Pre-sealed block for proposal at {}. Hash now {:?}, previously {:?}.",
+					header_num,
+					import_block.post_header().hash(),
+					header_hash,
+			);
+
+			telemetry!(CONSENSUS_INFO; "slots.pre_sealed_block";
+				"header_num" => ?header_num,
+				"hash_now" => ?import_block.post_header().hash(),
+				"hash_previously" => ?header_hash,
+			);
+
+			if let Err(err) = block_import.lock().import_block(import_block, Default::default()) {
+				warn!(target: logging_target,
+					"Error with block built on {:?}: {:?}",
+					parent_hash,
+					err,
+				);
+
+				telemetry!(CONSENSUS_WARN; "slots.err_with_block_built_on";
+					"hash" => ?parent_hash, "err" => ?err,
+				);
+			}
+		}))
+	}
+}
+
 /// Slot compatible inherent data.
 pub trait SlotCompatible {
 	/// Extract timestamp and slot from inherent data.