Connect the Tendermint machine to a GossipEngine

substrate/node/src/service.rs

@@ -184,7 +184,11 @@ pub async fn new_full(config: Configuration) -> Result<TaskManager, ServiceError
   })?;
 
   if is_authority {
-    authority.validate().await;
+    task_manager.spawn_essential_handle().spawn(
+      "tendermint",
+      None,
+      authority.validate(network, None),
+    );
   }
 
   network_starter.start_network();

substrate/tendermint/client/src/gossip.rs

@@ -9,11 +9,23 @@ use sc_network_gossip::{Validator, ValidatorContext, ValidationResult};
 use tendermint_machine::{SignedMessage, ext::SignatureScheme};
 
 #[derive(Clone)]
-struct TendermintGossip<S: SignatureScheme<ValidatorId = u16, Signature = Signature>> {
+pub struct TendermintGossip<S: SignatureScheme<ValidatorId = u16, Signature = Signature>> {
   number: Arc<RwLock<u64>>,
   signature_scheme: Arc<S>,
 }
 
+impl<S: SignatureScheme<ValidatorId = u16, Signature = Signature>> TendermintGossip<S> {
+  pub(crate) fn new(number: Arc<RwLock<u64>>, signature_scheme: Arc<S>) -> TendermintGossip<S> {
+    TendermintGossip { number, signature_scheme }
+  }
+
+  pub(crate) fn topic<B: Block>(number: u64) -> B::Hash {
+    <<B::Header as Header>::Hashing as Hash>::hash(
+      &[b"Tendermint Block Topic".as_ref(), &number.to_le_bytes()].concat(),
+    )
+  }
+}
+
 impl<B: Block, S: SignatureScheme<ValidatorId = u16, Signature = Signature>> Validator<B>
   for TendermintGossip<S>
 {
@@ -36,8 +48,6 @@ impl<B: Block, S: SignatureScheme<ValidatorId = u16, Signature = Signature>> Val
       return ValidationResult::Discard;
     }
 
-    ValidationResult::ProcessAndKeep(<<B::Header as Header>::Hashing as Hash>::hash(
-      &[b"Tendermint Topic".as_ref(), &msg.number().0.to_le_bytes()].concat(),
-    ))
+    ValidationResult::ProcessAndKeep(Self::topic::<B>(msg.number().0))
   }
 }

substrate/tendermint/client/src/lib.rs

@@ -7,6 +7,7 @@ use sp_api::{TransactionFor, ProvideRuntimeApi};
 
 use sc_executor::{NativeVersion, NativeExecutionDispatch, NativeElseWasmExecutor};
 use sc_transaction_pool::FullPool;
+use sc_network::NetworkService;
 use sc_service::{TaskManager, TFullClient};
 
 use substrate_prometheus_endpoint::Registry;
@@ -26,7 +27,7 @@ mod verifier;
 mod import_queue;
 use import_queue::TendermintImportQueue;
 
-mod gossip;
+pub(crate) mod gossip;
 
 mod select_chain;
 pub use select_chain::TendermintSelectChain;
@@ -85,6 +86,7 @@ impl<A: Announce<Block>> TendermintValidator for TendermintValidatorFirm<A> {
     DisableProofRecording,
   >;
 
+  type Network = Arc<NetworkService<Block, <Block as BlockTrait>::Hash>>;
   type Announce = A;
 }
 

substrate/tendermint/client/src/tendermint.rs

@@ -8,7 +8,7 @@ use async_trait::async_trait;
 
 use log::warn;
 
-use tokio::sync::RwLock as AsyncRwLock;
+use tokio::{sync::RwLock as AsyncRwLock, task::yield_now};
 
 use sp_core::{Encode, Decode, sr25519::Signature};
 use sp_inherents::{InherentData, InherentDataProvider, CreateInherentDataProviders};
@@ -24,6 +24,9 @@ use sc_consensus::{ForkChoiceStrategy, BlockImportParams, import_queue::Incoming
 
 use sc_service::ImportQueue;
 use sc_client_api::{BlockBackend, Finalizer};
+use sc_network_gossip::GossipEngine;
+
+use substrate_prometheus_endpoint::Registry;
 
 use tendermint_machine::{
   ext::{BlockError, BlockNumber, Commit, Network},
@@ -35,6 +38,7 @@ use crate::{
   types::TendermintValidator,
   validators::TendermintValidators,
   import_queue::{ImportFuture, TendermintImportQueue},
+  gossip::TendermintGossip,
   Announce,
 };
 
@@ -43,6 +47,7 @@ pub(crate) struct TendermintImport<T: TendermintValidator> {
 
   validators: Arc<TendermintValidators<T>>,
 
+  number: Arc<RwLock<u64>>,
   importing_block: Arc<RwLock<Option<<T::Block as Block>::Hash>>>,
   pub(crate) machine: Arc<RwLock<Option<TendermintHandle<Self>>>>,
 
@@ -57,7 +62,7 @@ pub(crate) struct TendermintImport<T: TendermintValidator> {
 
 pub struct TendermintAuthority<T: TendermintValidator>(pub(crate) TendermintImport<T>);
 impl<T: TendermintValidator> TendermintAuthority<T> {
-  pub async fn validate(mut self) {
+  pub async fn validate(mut self, network: T::Network, registry: Option<&Registry>) {
     let info = self.0.client.info();
 
     // Header::Number: TryInto<u64> doesn't implement Debug and can't be unwrapped
@@ -84,13 +89,55 @@ impl<T: TendermintValidator> TendermintAuthority<T> {
       .get_proposal(&self.0.client.header(BlockId::Hash(info.best_hash)).unwrap().unwrap())
       .await;
 
-    *self.0.machine.write().unwrap() = Some(TendermintMachine::new(
+    *self.0.number.write().unwrap() = last_number.0 + 1;
+    let mut gossip = GossipEngine::new(
+      network,
+      "tendermint",
+      Arc::new(TendermintGossip::new(self.0.number.clone(), self.0.validators.clone())),
+      registry,
+    );
+
+    let handle = TendermintMachine::new(
       self.0.clone(),
       // TODO
       0, // ValidatorId
       (last_number, last_time),
       proposal,
-    ));
+    );
+
+    let mut last_number = last_number.0 + 1;
+    let mut recv = gossip
+      .messages_for(TendermintGossip::<TendermintValidators<T>>::topic::<T::Block>(last_number));
+    loop {
+      match recv.try_next() {
+        Ok(Some(msg)) => handle
+          .messages
+          .send(match SignedMessage::decode(&mut msg.message.as_ref()) {
+            Ok(msg) => msg,
+            Err(e) => {
+              warn!("couldn't decode valid message: {}", e);
+              continue;
+            }
+          })
+          .await
+          .unwrap(),
+        Ok(None) => break,
+        // No messages available
+        Err(_) => {
+          // Check if we the block updated and should be listening on a different topic
+          let curr = *self.0.number.read().unwrap();
+          if last_number != curr {
+            last_number = curr;
+            // TODO: Will this return existing messages on the new height? Or will those have been
+            // ignored and are now gone?
+            recv = gossip.messages_for(TendermintGossip::<TendermintValidators<T>>::topic::<
+              T::Block,
+            >(last_number));
+          }
+          yield_now().await;
+        }
+      }
+    }
   }
 }
 
@@ -101,6 +148,7 @@ impl<T: TendermintValidator> Clone for TendermintImport<T> {
 
       validators: self.validators.clone(),
 
+      number: self.number.clone(),
       importing_block: self.importing_block.clone(),
       machine: self.machine.clone(),
 
@@ -126,6 +174,7 @@ impl<T: TendermintValidator> TendermintImport<T> {
 
       validators: Arc::new(TendermintValidators::new(client.clone())),
 
+      number: Arc::new(RwLock::new(0)),
       importing_block: Arc::new(RwLock::new(None)),
       machine: Arc::new(RwLock::new(None)),
 
@@ -388,6 +437,7 @@ impl<T: TendermintValidator> Network for TendermintImport<T> {
       .finalize_block(BlockId::Hash(hash), Some(justification), true)
       .map_err(|_| Error::InvalidJustification)
       .unwrap();
+    *self.number.write().unwrap() += 1;
     self.announce.announce(hash);
 
     self.get_proposal(block.header()).await

substrate/tendermint/client/src/types.rs

@@ -8,6 +8,7 @@ use sp_consensus::Environment;
 use sc_consensus::BlockImport;
 
 use sc_client_api::{BlockBackend, Backend, Finalizer};
+use sc_network_gossip::Network;
 
 use sp_tendermint::TendermintApi;
 
@@ -71,5 +72,6 @@ pub trait TendermintValidator: TendermintClient {
   type CIDP: CreateInherentDataProviders<Self::Block, ()> + 'static;
   type Environment: Send + Sync + Environment<Self::Block> + 'static;
 
+  type Network: Clone + Send + Sync + Network<Self::Block> + 'static;
   type Announce: Announce<Self::Block>;
 }