Fix polkadot parachains not producing blocks until next session (pari…

…tytech#4269)

... a few sessions too late :(, this already happened on polkadot, so as
of now there are no known relay-chains without async backing enabled in
runtime, but let's fix it in case someone else wants to repeat our
steps.

Fixes: paritytech#4226

---------

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

polkadot/node/network/statement-distribution/src/v2/mod.rs

@@ -826,7 +826,16 @@ pub(crate) fn handle_deactivate_leaves(state: &mut State, leaves: &[Hash]) {
 	// clean up sessions based on everything remaining.
 	let sessions: HashSet<_> = state.per_relay_parent.values().map(|r| r.session).collect();
 	state.per_session.retain(|s, _| sessions.contains(s));
-	state.unused_topologies.retain(|s, _| sessions.contains(s));
+
+	let last_session_index = state.unused_topologies.keys().max().copied();
+	// Do not clean-up the last saved toplogy unless we moved to the next session
+	// This is needed because handle_deactive_leaves, gets also called when
+	// prospective_parachains APIs are not present, so we would actually remove
+	// the topology without using it because `per_relay_parent` is empty until
+	// prospective_parachains gets enabled
+	state
+		.unused_topologies
+		.retain(|s, _| sessions.contains(s) || last_session_index == Some(*s));
 }
 
 #[overseer::contextbounds(StatementDistribution, prefix=self::overseer)]

polkadot/node/network/statement-distribution/src/v2/tests/mod.rs

@@ -509,6 +509,12 @@ async fn setup_test_and_connect_peers(
 	// Send gossip topology and activate leaf.
 	if send_topology_before_leaf {
 		send_new_topology(overseer, state.make_dummy_topology()).await;
+		// Send cleaning up of a leaf to make sure it does not clear the save topology as well.
+		overseer
+			.send(FromOrchestra::Signal(OverseerSignal::ActiveLeaves(
+				ActiveLeavesUpdate::stop_work(Hash::random()),
+			)))
+			.await;
 		activate_leaf(overseer, &test_leaf, &state, true, vec![]).await;
 	} else {
 		activate_leaf(overseer, &test_leaf, &state, true, vec![]).await;