Lean BEEFY to Full BEEFY - don't skip (older) mandatory blocks and import justifications (#11821)

* client/beefy: don't accept vote for older rounds

* client/beefy: clean up and reorg the worker struct

* client/beefy: first step towards Full BEEFY

The first step from Lean->Full BEEFY is to have the worker
enforce uninterrupted line of BEEFY finalized mandatory blocks.

There is one mandatory block per session (the first block in the
session). As such, votes processing and votes generation now
enforces that all mandatory blocks are finalized in strict
monotonically increasing sequence and no block 'N' will be worked
on if there is any GRANDPA finalized but BEEFY non-final mandatory
block 'M', where 'M < N'.

Implementation details:

- Introduced 'VoterOracle' to separate the voting decisions logic,
  and track new/pending sessions.

- New sessions get queued up with the worker operating either:
  1. up-to-date - all mandatory blocks leading up to current GRANDPA
     finalized: queue has ONE element, the 'current session' where
     `mandatory_done == true`,
  2. lagging behind GRANDPA: queue has [1, N] elements, where all
     `mandatory_done == false`.
     In this state, everytime a session gets its mandatory block
     BEEFY finalized, the session is popped off the queue,
     eventually getting to operating mode `1. up-to-date`.

- Votes get triaged and those that fall withing the `VoterOracle`
  allowed window get processed, the others get dropped if stale,
  or buffered for later processing (when they reach the window).

- Worker general code was also updated to fall in one of two roles:
  1. react to external events and change internal 'state',
  2. generate events/votes based on internal 'state'.

Signed-off-by: acatangiu <adrian@parity.io>

* client/beefy: sketch idea for block import and sync

Signed-off-by: acatangiu <adrian@parity.io>

* client/beefy: add BEEFY block import

* client/beefy: process justifications from block import

* client/beefy: add TODOs for sync protocol

* client/beefy: add more docs and comments

* client/beefy-rpc: fix RPC error

* client/beefy: verify justification validity on block import

* client/beefy: more tests

* client/beefy: small fixes

- first handle and note the self vote before gossiping it,
- don't shortcircuit on err when processing pending votes.

* client/beefy: remove invalid justifications at block import

* todo: beefy block import tests

* RFC: ideas for multiple justifications per block

* Revert "RFC: ideas for multiple justifications per block"

This reverts commit 8256fb07d3124db69daf252720b3c0208202624d.

* client/beefy: append justif to backend on block import

* client/beefy: groundwork for block import test

* client/beefy: groundwork2 for block import test

* client/beefy: groundwork3 for block import test

* client/beefy: add block import test

* client/beefy: add required trait bounds to block import builder

* remove client from beefy block import, backend gets the job done

Signed-off-by: acatangiu <adrian@parity.io>

substrate/client/beefy/src/lib.rs

@@ -16,23 +16,18 @@
 // You should have received a copy of the GNU General Public License
 // along with this program. If not, see <https://www.gnu.org/licenses/>.
 
-use std::sync::Arc;
-
+use beefy_primitives::{BeefyApi, MmrRootHash};
 use prometheus::Registry;
-
 use sc_client_api::{Backend, BlockchainEvents, Finalizer};
+use sc_consensus::BlockImport;
 use sc_network_gossip::Network as GossipNetwork;
-
 use sp_api::ProvideRuntimeApi;
 use sp_blockchain::HeaderBackend;
-use sp_consensus::SyncOracle;
+use sp_consensus::{Error as ConsensusError, SyncOracle};
 use sp_keystore::SyncCryptoStorePtr;
 use sp_mmr_primitives::MmrApi;
 use sp_runtime::traits::Block;
-
-use beefy_primitives::{BeefyApi, MmrRootHash};
-
-use crate::notification::{BeefyBestBlockSender, BeefySignedCommitmentSender};
+use std::sync::Arc;
 
 mod error;
 mod gossip;
@@ -41,11 +36,21 @@ mod metrics;
 mod round;
 mod worker;
 
+pub mod import;
+pub mod justification;
 pub mod notification;
 
 #[cfg(test)]
 mod tests;
 
+use crate::{
+	import::BeefyBlockImport,
+	notification::{
+		BeefyBestBlockSender, BeefyBestBlockStream, BeefySignedCommitmentSender,
+		BeefySignedCommitmentStream,
+	},
+};
+
 pub use beefy_protocol_name::standard_name as protocol_standard_name;
 
 pub(crate) mod beefy_protocol_name {
@@ -110,6 +115,68 @@ where
 	// empty
 }
 
+/// Links between the block importer, the background voter and the RPC layer,
+/// to be used by the voter.
+#[derive(Clone)]
+pub struct BeefyVoterLinks<B: Block> {
+	// BlockImport -> Voter links
+	/// Stream of BEEFY signed commitments from block import to voter.
+	pub from_block_import_justif_stream: BeefySignedCommitmentStream<B>,
+
+	// Voter -> RPC links
+	/// Sends BEEFY signed commitments from voter to RPC.
+	pub to_rpc_justif_sender: BeefySignedCommitmentSender<B>,
+	/// Sends BEEFY best block hashes from voter to RPC.
+	pub to_rpc_best_block_sender: BeefyBestBlockSender<B>,
+}
+
+/// Links used by the BEEFY RPC layer, from the BEEFY background voter.
+#[derive(Clone)]
+pub struct BeefyRPCLinks<B: Block> {
+	/// Stream of signed commitments coming from the voter.
+	pub from_voter_justif_stream: BeefySignedCommitmentStream<B>,
+	/// Stream of BEEFY best block hashes coming from the voter.
+	pub from_voter_best_beefy_stream: BeefyBestBlockStream<B>,
+}
+
+/// Make block importer and link half necessary to tie the background voter to it.
+pub fn beefy_block_import_and_links<B, BE, RuntimeApi, I>(
+	wrapped_block_import: I,
+	backend: Arc<BE>,
+	runtime: Arc<RuntimeApi>,
+) -> (BeefyBlockImport<B, BE, RuntimeApi, I>, BeefyVoterLinks<B>, BeefyRPCLinks<B>)
+where
+	B: Block,
+	BE: Backend<B>,
+	I: BlockImport<B, Error = ConsensusError, Transaction = sp_api::TransactionFor<RuntimeApi, B>>
+		+ Send
+		+ Sync,
+	RuntimeApi: ProvideRuntimeApi<B> + Send + Sync,
+	RuntimeApi::Api: BeefyApi<B>,
+{
+	// Voter -> RPC links
+	let (to_rpc_justif_sender, from_voter_justif_stream) =
+		notification::BeefySignedCommitmentStream::<B>::channel();
+	let (to_rpc_best_block_sender, from_voter_best_beefy_stream) =
+		notification::BeefyBestBlockStream::<B>::channel();
+
+	// BlockImport -> Voter links
+	let (to_voter_justif_sender, from_block_import_justif_stream) =
+		notification::BeefySignedCommitmentStream::<B>::channel();
+
+	// BlockImport
+	let import =
+		BeefyBlockImport::new(backend, runtime, wrapped_block_import, to_voter_justif_sender);
+	let voter_links = BeefyVoterLinks {
+		from_block_import_justif_stream,
+		to_rpc_justif_sender,
+		to_rpc_best_block_sender,
+	};
+	let rpc_links = BeefyRPCLinks { from_voter_best_beefy_stream, from_voter_justif_stream };
+
+	(import, voter_links, rpc_links)
+}
+
 /// BEEFY gadget initialization parameters.
 pub struct BeefyParams<B, BE, C, N, R>
 where
@@ -130,16 +197,14 @@ where
 	pub key_store: Option<SyncCryptoStorePtr>,
 	/// Gossip network
 	pub network: N,
-	/// BEEFY signed commitment sender
-	pub signed_commitment_sender: BeefySignedCommitmentSender<B>,
-	/// BEEFY best block sender
-	pub beefy_best_block_sender: BeefyBestBlockSender<B>,
 	/// Minimal delta between blocks, BEEFY should vote for
 	pub min_block_delta: u32,
 	/// Prometheus metric registry
 	pub prometheus_registry: Option<Registry>,
 	/// Chain specific GRANDPA protocol name. See [`beefy_protocol_name::standard_name`].
 	pub protocol_name: std::borrow::Cow<'static, str>,
+	/// Links between the block importer, the background voter and the RPC layer.
+	pub links: BeefyVoterLinks<B>,
 }
 
 /// Start the BEEFY gadget.
@@ -160,11 +225,10 @@ where
 		runtime,
 		key_store,
 		network,
-		signed_commitment_sender,
-		beefy_best_block_sender,
 		min_block_delta,
 		prometheus_registry,
 		protocol_name,
+		links,
 	} = beefy_params;
 
 	let sync_oracle = network.clone();
@@ -194,14 +258,13 @@ where
 		client,
 		backend,
 		runtime,
+		sync_oracle,
 		key_store: key_store.into(),
-		signed_commitment_sender,
-		beefy_best_block_sender,
 		gossip_engine,
 		gossip_validator,
-		min_block_delta,
+		links,
 		metrics,
-		sync_oracle,
+		min_block_delta,
 	};
 
 	let worker = worker::BeefyWorker::<_, _, _, _, _>::new(worker_params);