approval-voting: Make sure we always mark approved candidates approved in a different relay chain context (#4153)

... see for more detail why this is needed

https://github.com/paritytech/polkadot-sdk/issues/4149#issuecomment-2058472444

## TODO:
- [x] Unittests
- [x] Replicate scenario from
https://github.com/paritytech/polkadot-sdk/issues/4149 and confirm this
fixes it: https://github.com/paritytech/polkadot-sdk/issues/4149 [
Replicated on a zombienet with some hacked nodes, that we can end up in
this state where no-wake is schedule and the nodes are pending new
assignments]

---------

Signed-off-by: Alexandru Gheorghe <alexandru.gheorghe@parity.io>
Co-authored-by: Andrei Sandu <54316454+sandreim@users.noreply.github.com>

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

@@ -978,6 +978,7 @@ where
 					woken_block,
 					woken_candidate,
 					&subsystem.metrics,
+					&wakeups,
 				).await?
 			}
 			next_msg = ctx.recv().fuse() => {
@@ -1152,6 +1153,7 @@ async fn handle_actions<Context>(
 					candidate_hash,
 					delayed_approvals_timers,
 					approval_request,
+					&wakeups,
 				)
 				.await?
 				.into_iter()
@@ -1663,6 +1665,7 @@ async fn handle_from_overseer<Context>(
 					|r| {
 						let _ = res.send(r);
 					},
+					&wakeups,
 				)
 				.await?
 				.0,
@@ -2477,6 +2480,7 @@ async fn check_and_import_approval<T, Sender>(
 	metrics: &Metrics,
 	approval: IndirectSignedApprovalVoteV2,
 	with_response: impl FnOnce(ApprovalCheckResult) -> T,
+	wakeups: &Wakeups,
 ) -> SubsystemResult<(Vec<Action>, T)>
 where
 	Sender: SubsystemSender<RuntimeApiMessage>,
@@ -2655,6 +2659,7 @@ where
 			approved_candidate_hash,
 			candidate_entry,
 			ApprovalStateTransition::RemoteApproval(approval.validator),
+			wakeups,
 		)
 		.await;
 		actions.extend(new_actions);
@@ -2689,6 +2694,10 @@ impl ApprovalStateTransition {
 			ApprovalStateTransition::WakeupProcessed => false,
 		}
 	}
+
+	fn is_remote_approval(&self) -> bool {
+		matches!(*self, ApprovalStateTransition::RemoteApproval(_))
+	}
 }
 
 // Advance the approval state, either by importing an approval vote which is already checked to be
@@ -2705,6 +2714,7 @@ async fn advance_approval_state<Sender>(
 	candidate_hash: CandidateHash,
 	mut candidate_entry: CandidateEntry,
 	transition: ApprovalStateTransition,
+	wakeups: &Wakeups,
 ) -> Vec<Action>
 where
 	Sender: SubsystemSender<RuntimeApiMessage>,
@@ -2835,6 +2845,43 @@ where
 			status.required_tranches,
 		));
 
+		if is_approved && transition.is_remote_approval() {
+			// Make sure we wake other blocks in case they have
+			// a no-show that might be covered by this approval.
+			for (fork_block_hash, fork_approval_entry) in candidate_entry
+				.block_assignments
+				.iter()
+				.filter(|(hash, _)| **hash != block_hash)
+			{
+				let assigned_on_fork_block = validator_index
+					.as_ref()
+					.map(|validator_index| fork_approval_entry.is_assigned(*validator_index))
+					.unwrap_or_default();
+				if wakeups.wakeup_for(*fork_block_hash, candidate_hash).is_none() &&
+					!fork_approval_entry.is_approved() &&
+					assigned_on_fork_block
+				{
+					let fork_block_entry = db.load_block_entry(fork_block_hash);
+					if let Ok(Some(fork_block_entry)) = fork_block_entry {
+						actions.push(Action::ScheduleWakeup {
+							block_hash: *fork_block_hash,
+							block_number: fork_block_entry.block_number(),
+							candidate_hash,
+							// Schedule the wakeup next tick, since the assignment must be a
+							// no-show, because there is no-wakeup scheduled.
+							tick: tick_now + 1,
+						})
+					} else {
+						gum::debug!(
+							target: LOG_TARGET,
+							?fork_block_entry,
+							?fork_block_hash,
+							"Failed to load block entry"
+						)
+					}
+				}
+			}
+		}
 		// We have no need to write the candidate entry if all of the following
 		// is true:
 		//
@@ -2896,6 +2943,7 @@ async fn process_wakeup<Context>(
 	relay_block: Hash,
 	candidate_hash: CandidateHash,
 	metrics: &Metrics,
+	wakeups: &Wakeups,
 ) -> SubsystemResult<Vec<Action>> {
 	let mut span = state
 		.spans
@@ -3064,6 +3112,7 @@ async fn process_wakeup<Context>(
 			candidate_hash,
 			candidate_entry,
 			ApprovalStateTransition::WakeupProcessed,
+			wakeups,
 		)
 		.await,
 	);
@@ -3294,6 +3343,7 @@ async fn issue_approval<Context>(
 	candidate_hash: CandidateHash,
 	delayed_approvals_timers: &mut DelayedApprovalTimer,
 	ApprovalVoteRequest { validator_index, block_hash }: ApprovalVoteRequest,
+	wakeups: &Wakeups,
 ) -> SubsystemResult<Vec<Action>> {
 	let mut issue_approval_span = state
 		.spans
@@ -3415,6 +3465,7 @@ async fn issue_approval<Context>(
 		candidate_hash,
 		candidate_entry,
 		ApprovalStateTransition::LocalApproval(validator_index as _),
+		wakeups,
 	)
 	.await;