Beefy: small fixes (#2378)

Related to #2285

- save the state of the BEEFY gadget after processing a finality proof.
We need this in order to avoid skipping blocks.
- avoid reprocessing the old state when not necessary

substrate/client/consensus/beefy/src/worker.rs

@@ -456,6 +456,7 @@ where
 			.filter(|genesis| *genesis == self.persisted_state.pallet_genesis)
 			.ok_or(Error::ConsensusReset)?;
 
+		let mut new_session_added = false;
 		if *header.number() > self.best_grandpa_block() {
 			// update best GRANDPA finalized block we have seen
 			self.persisted_state.set_best_grandpa(header.clone());
@@ -475,9 +476,15 @@ where
 			{
 				if let Some(new_validator_set) = find_authorities_change::<B>(&header) {
 					self.init_session_at(new_validator_set, *header.number());
+					new_session_added = true;
 				}
 			}
 
+			if new_session_added {
+				crate::aux_schema::write_voter_state(&*self.backend, &self.persisted_state)
+					.map_err(|e| Error::Backend(e.to_string()))?;
+			}
+
 			// Update gossip validator votes filter.
 			if let Err(e) = self
 				.persisted_state
@@ -848,15 +855,10 @@ where
 				.fuse(),
 		);
 
+		self.process_new_state();
 		let error = loop {
-			// Act on changed 'state'.
-			self.process_new_state();
-
 			// Mutable reference used to drive the gossip engine.
 			let mut gossip_engine = &mut self.comms.gossip_engine;
-			// Use temp val and report after async section,
-			// to avoid having to Mutex-wrap `gossip_engine`.
-			let mut gossip_report: Option<PeerReport> = None;
 
 			// Wait for, and handle external events.
 			// The branches below only change 'state', actual voting happens afterwards,
@@ -884,10 +886,15 @@ where
 							if let Err(err) = self.triage_incoming_justif(justif) {
 								debug!(target: LOG_TARGET, "🥩 {}", err);
 							}
-							gossip_report = Some(peer_report);
+							self.comms.gossip_engine.report(peer_report.who, peer_report.cost_benefit);
+						},
+						ResponseInfo::PeerReport(peer_report) => {
+							self.comms.gossip_engine.report(peer_report.who, peer_report.cost_benefit);
+							continue;
+						},
+						ResponseInfo::Pending => {
+							continue;
 						},
-						ResponseInfo::PeerReport(peer_report) => gossip_report = Some(peer_report),
-						ResponseInfo::Pending => (),
 					}
 				},
 				justif = block_import_justif.next() => {
@@ -924,12 +931,15 @@ where
 				},
 				// Process peer reports.
 				report = self.comms.gossip_report_stream.next() => {
-					gossip_report = report;
+					if let Some(PeerReport { who, cost_benefit }) = report {
+						self.comms.gossip_engine.report(who, cost_benefit);
+					}
+					continue;
 				},
 			}
-			if let Some(PeerReport { who, cost_benefit }) = gossip_report {
-				self.comms.gossip_engine.report(who, cost_benefit);
-			}
+
+			// Act on changed 'state'.
+			self.process_new_state();
 		};
 
 		// return error _and_ `comms` that can be reused