Fix accidental no-shows on node restart (#3277)

If approval was in progress we didn't actually restart it, so we end up
in a situation where we distribute our assignment, but we don't
distribute any approval.

---------

Signed-off-by: Alexandru Gheorghe <alexandru.gheorghe@parity.io>

polkadot/node/core/approval-voting/src/lib.rs

@@ -1253,13 +1253,20 @@ async fn handle_actions<Context>(
 			Action::BecomeActive => {
 				*mode = Mode::Active;
 
-				let messages = distribution_messages_for_activation(
+				let (messages, next_actions) = distribution_messages_for_activation(
+					ctx,
 					overlayed_db,
 					state,
 					delayed_approvals_timers,
-				)?;
+					session_info_provider,
+				)
+				.await?;
 
 				ctx.send_messages(messages.into_iter()).await;
+				let next_actions: Vec<Action> =
+					next_actions.into_iter().map(|v| v.clone()).chain(actions_iter).collect();
+
+				actions_iter = next_actions.into_iter();
 			},
 			Action::Conclude => {
 				conclude = true;
@@ -1313,15 +1320,19 @@ fn get_assignment_core_indices(
 	}
 }
 
-fn distribution_messages_for_activation(
+#[overseer::contextbounds(ApprovalVoting, prefix = self::overseer)]
+async fn distribution_messages_for_activation<Context>(
+	ctx: &mut Context,
 	db: &OverlayedBackend<'_, impl Backend>,
 	state: &State,
 	delayed_approvals_timers: &mut DelayedApprovalTimer,
-) -> SubsystemResult<Vec<ApprovalDistributionMessage>> {
+	session_info_provider: &mut RuntimeInfo,
+) -> SubsystemResult<(Vec<ApprovalDistributionMessage>, Vec<Action>)> {
 	let all_blocks: Vec<Hash> = db.load_all_blocks()?;
 
 	let mut approval_meta = Vec::with_capacity(all_blocks.len());
 	let mut messages = Vec::new();
+	let mut actions = Vec::new();
 
 	messages.push(ApprovalDistributionMessage::NewBlocks(Vec::new())); // dummy value.
 
@@ -1396,16 +1407,60 @@ fn distribution_messages_for_activation(
 									&claimed_core_indices,
 									&block_entry,
 								) {
-									Ok(bitfield) => messages.push(
-										ApprovalDistributionMessage::DistributeAssignment(
-											IndirectAssignmentCertV2 {
-												block_hash,
-												validator: assignment.validator_index(),
-												cert: assignment.cert().clone(),
-											},
-											bitfield,
-										),
-									),
+									Ok(bitfield) => {
+										gum::debug!(
+											target: LOG_TARGET,
+											candidate_hash = ?candidate_entry.candidate_receipt().hash(),
+											?block_hash,
+											"Discovered, triggered assignment, not approved yet",
+										);
+
+										let indirect_cert = IndirectAssignmentCertV2 {
+											block_hash,
+											validator: assignment.validator_index(),
+											cert: assignment.cert().clone(),
+										};
+										messages.push(
+											ApprovalDistributionMessage::DistributeAssignment(
+												indirect_cert.clone(),
+												bitfield.clone(),
+											),
+										);
+
+										if !block_entry
+											.candidate_is_pending_signature(*candidate_hash)
+										{
+											let ExtendedSessionInfo { ref executor_params, .. } =
+												match get_extended_session_info(
+													session_info_provider,
+													ctx.sender(),
+													block_entry.block_hash(),
+													block_entry.session(),
+												)
+												.await
+												{
+													Some(i) => i,
+													None => continue,
+												};
+
+											actions.push(Action::LaunchApproval {
+												claimed_candidate_indices: bitfield,
+												candidate_hash: candidate_entry
+													.candidate_receipt()
+													.hash(),
+												indirect_cert,
+												assignment_tranche: assignment.tranche(),
+												relay_block_hash: block_hash,
+												session: block_entry.session(),
+												executor_params: executor_params.clone(),
+												candidate: candidate_entry
+													.candidate_receipt()
+													.clone(),
+												backing_group: approval_entry.backing_group(),
+												distribute_assignment: false,
+											});
+										}
+									},
 									Err(err) => {
 										// Should never happen. If we fail here it means the
 										// assignment is null (no cores claimed).
@@ -1496,7 +1551,7 @@ fn distribution_messages_for_activation(
 	}
 
 	messages[0] = ApprovalDistributionMessage::NewBlocks(approval_meta);
-	Ok(messages)
+	Ok((messages, actions))
 }
 
 // Handle an incoming signal from the overseer. Returns true if execution should conclude.

polkadot/node/core/approval-voting/src/persisted_entries.rs

@@ -588,6 +588,13 @@ impl BlockEntry {
 		!self.candidates_pending_signature.is_empty()
 	}
 
+	/// Returns true if candidate hash is in the queue for a signature.
+	pub fn candidate_is_pending_signature(&self, candidate_hash: CandidateHash) -> bool {
+		self.candidates_pending_signature
+			.values()
+			.any(|context| context.candidate_hash == candidate_hash)
+	}
+
 	/// Candidate hashes  for candidates pending signatures
 	fn candidate_hashes_pending_signature(&self) -> Vec<CandidateHash> {
 		self.candidates_pending_signature