Fix #183

substrate/tendermint/client/src/authority/mod.rs

@@ -22,7 +22,7 @@ use sp_runtime::{
 };
 use sp_blockchain::HeaderBackend;
 
-use sp_consensus::{Error, BlockOrigin, Proposer, Environment};
+use sp_consensus::{Error, BlockOrigin, BlockStatus, Proposer, Environment};
 use sc_consensus::import_queue::IncomingBlock;
 
 use sc_service::ImportQueue;
@@ -86,7 +86,6 @@ async fn get_proposal<T: TendermintValidator>(
   env: &Arc<Mutex<T::Environment>>,
   import: &TendermintImport<T>,
   header: &<T::Block as Block>::Header,
-  stub: bool,
 ) -> T::Block {
   let proposer =
     env.lock().await.init(header).await.expect("Failed to create a proposer for the new block");
@@ -95,15 +94,11 @@ async fn get_proposal<T: TendermintValidator>(
     .propose(
       import.inherent_data(*header.parent_hash()).await,
       Digest::default(),
-      if stub {
-        Duration::ZERO
-      } else {
-        // The first processing time is to build the block
-        // The second is for it to be downloaded (assumes a block won't take longer to download
-        // than it'll take to process)
-        // The third is for it to actually be processed
-        Duration::from_secs((T::BLOCK_PROCESSING_TIME_IN_SECONDS / 3).into())
-      },
+      // The first processing time is to build the block
+      // The second is for it to be downloaded (assumes a block won't take longer to download
+      // than it'll take to process)
+      // The third is for it to actually be processed
+      Duration::from_secs((T::BLOCK_PROCESSING_TIME_IN_SECONDS / 3).into()),
       Some(T::PROPOSED_BLOCK_SIZE_LIMIT),
     )
     .await
@@ -118,7 +113,7 @@ impl<T: TendermintValidator> TendermintAuthority<T> {
   }
 
   async fn get_proposal(&self, header: &<T::Block as Block>::Header) -> T::Block {
-    get_proposal(&self.active.as_ref().unwrap().env, &self.import, header, false).await
+    get_proposal(&self.active.as_ref().unwrap().env, &self.import, header).await
   }
 
   /// Create and run a new Tendermint Authority, proposing and voting on blocks.
@@ -263,9 +258,10 @@ impl<T: TendermintValidator> TendermintAuthority<T> {
             step.send((
               BlockNumber(number),
               Commit::decode(&mut justifications.get(CONSENSUS_ID).unwrap().as_ref()).unwrap(),
-              // This will fail if syncing occurs radically faster than machine stepping takes
-              // TODO: Set true when initial syncing
-              get_proposal(&env, &import, &notif.header, false).await
+              // Creating a proposal will fail if syncing occurs radically faster than machine
+              // stepping takes
+              // Don't create proposals when stepping accordingly
+              None
             )).await.unwrap();
           } else {
             debug!(
@@ -437,10 +433,16 @@ impl<T: TendermintValidator> Network for TendermintAuthority<T> {
     &mut self,
     block: T::Block,
     commit: Commit<TendermintValidators<T>>,
-  ) -> T::Block {
+  ) -> Option<T::Block> {
     // Prevent import_block from being called while we run
     let _lock = self.import.sync_lock.lock().await;
 
+    // If we didn't import this block already, return
+    // If it's a legitimate block, we'll pick it up in the standard sync loop
+    if self.import.client.block_status(block.hash()).unwrap() != BlockStatus::InChainWithState {
+      return None;
+    }
+
     // Check if we already imported this externally
     if self.import.client.justifications(block.hash()).unwrap().is_some() {
       debug!(target: "tendermint", "Machine produced a commit after we already synced it");
@@ -487,6 +489,6 @@ impl<T: TendermintValidator> Network for TendermintAuthority<T> {
     // Clear any blocks for the previous slot which we were willing to recheck
     *self.import.recheck.write().unwrap() = HashSet::new();
 
-    self.get_proposal(block.header()).await
+    Some(self.get_proposal(block.header()).await)
   }
 }

substrate/tendermint/machine/src/block.rs

@@ -14,7 +14,7 @@ use crate::{
 pub(crate) struct BlockData<N: Network> {
   pub(crate) number: BlockNumber,
   pub(crate) validator_id: Option<N::ValidatorId>,
-  pub(crate) proposal: N::Block,
+  pub(crate) proposal: Option<N::Block>,
 
   pub(crate) log: MessageLog<N>,
   pub(crate) slashes: HashSet<N::ValidatorId>,
@@ -35,7 +35,7 @@ impl<N: Network> BlockData<N> {
     weights: Arc<N::Weights>,
     number: BlockNumber,
     validator_id: Option<N::ValidatorId>,
-    proposal: N::Block,
+    proposal: Option<N::Block>,
   ) -> BlockData<N> {
     BlockData {
       number,
@@ -106,12 +106,8 @@ impl<N: Network> BlockData<N> {
 
     // 14-21
     if Some(proposer) == self.validator_id {
-      let (round, block) = if let Some((round, block)) = &self.valid {
-        (Some(*round), block.clone())
-      } else {
-        (None, self.proposal.clone())
-      };
-      Some(Data::Proposal(round, block))
+      let (round, block) = self.valid.clone().unzip();
+      block.or_else(|| self.proposal.clone()).map(|block| Data::Proposal(round, block))
     } else {
       self.round_mut().set_timeout(Step::Propose);
       None

substrate/tendermint/machine/src/ext.rs

@@ -270,5 +270,5 @@ pub trait Network: Send + Sync {
     &mut self,
     block: Self::Block,
     commit: Commit<Self::SignatureScheme>,
-  ) -> Self::Block;
+  ) -> Option<Self::Block>;
 }

substrate/tendermint/machine/src/lib.rs

@@ -135,7 +135,8 @@ pub struct TendermintMachine<N: Network> {
 
   queue: VecDeque<MessageFor<N>>,
   msg_recv: mpsc::UnboundedReceiver<SignedMessageFor<N>>,
-  step_recv: mpsc::UnboundedReceiver<(BlockNumber, Commit<N::SignatureScheme>, N::Block)>,
+  #[allow(clippy::type_complexity)]
+  step_recv: mpsc::UnboundedReceiver<(BlockNumber, Commit<N::SignatureScheme>, Option<N::Block>)>,
 
   block: BlockData<N>,
 }
@@ -143,7 +144,7 @@ pub struct TendermintMachine<N: Network> {
 pub type StepSender<N> = mpsc::UnboundedSender<(
   BlockNumber,
   Commit<<N as Network>::SignatureScheme>,
-  <N as Network>::Block,
+  Option<<N as Network>::Block>,
 )>;
 
 pub type MessageSender<N> = mpsc::UnboundedSender<SignedMessageFor<N>>;
@@ -186,7 +187,7 @@ impl<N: Network + 'static> TendermintMachine<N> {
   }
 
   // 53-54
-  async fn reset(&mut self, end_round: RoundNumber, proposal: N::Block) {
+  async fn reset(&mut self, end_round: RoundNumber, proposal: Option<N::Block>) {
     // Ensure we have the end time data for the last round
     self.block.populate_end_time(end_round);
 
@@ -209,7 +210,11 @@ impl<N: Network + 'static> TendermintMachine<N> {
     self.round(RoundNumber(0), Some(round_end));
   }
 
-  async fn reset_by_commit(&mut self, commit: Commit<N::SignatureScheme>, proposal: N::Block) {
+  async fn reset_by_commit(
+    &mut self,
+    commit: Commit<N::SignatureScheme>,
+    proposal: Option<N::Block>,
+  ) {
     let mut round = self.block.round().number;
     // If this commit is for a round we don't have, jump up to it
     while self.block.end_time[&round].canonical() < commit.end_time {
@@ -271,7 +276,12 @@ impl<N: Network + 'static> TendermintMachine<N> {
           msg_recv,
           step_recv,
 
-          block: BlockData::new(weights, BlockNumber(last_block.0 + 1), validator_id, proposal),
+          block: BlockData::new(
+            weights,
+            BlockNumber(last_block.0 + 1),
+            validator_id,
+            Some(proposal),
+          ),
         };
 
         // The end time of the last block is the start time for this one

substrate/tendermint/machine/tests/ext.rs

@@ -140,11 +140,11 @@ impl Network for TestNetwork {
     &mut self,
     block: TestBlock,
     commit: Commit<TestSignatureScheme>,
-  ) -> TestBlock {
+  ) -> Option<TestBlock> {
     dbg!("Adding ", &block);
     assert!(block.valid.is_ok());
     assert!(self.verify_commit(block.id(), &commit));
-    TestBlock { id: (u32::from_le_bytes(block.id) + 1).to_le_bytes(), valid: Ok(()) }
+    Some(TestBlock { id: (u32::from_le_bytes(block.id) + 1).to_le_bytes(), valid: Ok(()) })
   }
 }