collator-protocol: Remove stale pending collations from the waiting queue (#10906)

This PR removes the stale pending collations from the waiting queue when
the peer that advertised the collation disconnects.

When the peer reconnects, the peer data is freshly created without any
prior information about advertised collations.
Then the state-pending collation is picked from the queue. The network
request will not be emitted since the `fn fetch_collation` sees no prior
advertisement via `peer_data.has_advertised` and returns
`Err(FetchError::NotAdvertised)`.

To avoid this, remove the stale entries immediately when the peer
disconnects.

Part of the stabilization of:
- #10425

---------

Signed-off-by: Alexandru Vasile <[email protected]>
Co-authored-by: Bastian Köcher <[email protected]>
Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: 3 people

polkadot/node/network/collator-protocol/src/validator_side/collation.rs

@@ -324,6 +324,13 @@ impl Collations {
 	pub(super) fn queued_for_para(&self, para_id: &ParaId) -> usize {
 		self.waiting_queue.get(para_id).map(|queue| queue.len()).unwrap_or_default()
 	}
+
+	/// Remove all pending collations for a specific peer from the waiting queue.
+	pub(super) fn remove_pending_for_peer(&mut self, peer_id: &PeerId) {
+		for queue in self.waiting_queue.values_mut() {
+			queue.retain(|(pending, _)| &pending.peer_id != peer_id);
+		}
+	}
 }
 
 // Any error that can occur when awaiting a collation fetch response.

polkadot/node/network/collator-protocol/src/validator_side/mod.rs

@@ -1695,6 +1695,11 @@ async fn handle_network_msg<Context>(
 		},
 		PeerDisconnected(peer_id) => {
 			state.peer_data.remove(&peer_id);
+			// Clean up any pending collations from this peer in all waiting queues.
+			// This prevents stale entries when the peer reconnects with empty advertisements.
+			for per_relay_parent in state.per_relay_parent.values_mut() {
+				per_relay_parent.collations.remove_pending_for_peer(&peer_id);
+			}
 			state.metrics.note_collator_peer_count(state.peer_data.len());
 		},
 		NewGossipTopology { .. } => {

polkadot/node/network/collator-protocol/src/validator_side/tests/mod.rs

@@ -1161,3 +1161,137 @@ fn view_change_clears_old_collators() {
 		virtual_overseer
 	})
 }
+
+/// Test that when a peer disconnects, their pending collations are removed from the waiting queue.
+/// This prevents "NotAdvertised" errors when the peer reconnects with empty advertisement state.
+#[test]
+fn peer_disconnect_clears_pending_collations_from_waiting_queue() {
+	let mut test_state = TestState::default();
+
+	test_harness(ReputationAggregator::new(|_| true), HashSet::new(), |test_harness| async move {
+		let TestHarness { mut virtual_overseer, .. } = test_harness;
+
+		let relay_parent = test_state.relay_parent;
+		update_view(&mut virtual_overseer, &mut test_state, vec![(relay_parent, 0)]).await;
+
+		// Connect first collator and have them advertise - this will trigger a fetch.
+		let peer_a = PeerId::random();
+		let collator_a = test_state.collators[0].clone();
+
+		connect_and_declare_collator(
+			&mut virtual_overseer,
+			peer_a,
+			collator_a.clone(),
+			test_state.chain_ids[0],
+			CollationVersion::V1,
+		)
+		.await;
+
+		advertise_collation(&mut virtual_overseer, peer_a, relay_parent, None).await;
+
+		// First collation fetch is initiated.
+		let response_channel_a = assert_fetch_collation_request(
+			&mut virtual_overseer,
+			relay_parent,
+			test_state.chain_ids[0],
+			None,
+		)
+		.await;
+
+		// Connect second collator and have them advertise.
+		// Since we're already fetching, this goes into the waiting queue.
+		let peer_b = PeerId::random();
+		let collator_b = test_state.collators[1].clone();
+
+		connect_and_declare_collator(
+			&mut virtual_overseer,
+			peer_b,
+			collator_b.clone(),
+			test_state.chain_ids[0],
+			CollationVersion::V1,
+		)
+		.await;
+
+		advertise_collation(&mut virtual_overseer, peer_b, relay_parent, None).await;
+
+		// Now disconnect peer_b. This should clean up their entry from the waiting queue.
+		overseer_send(
+			&mut virtual_overseer,
+			CollatorProtocolMessage::NetworkBridgeUpdate(NetworkBridgeEvent::PeerDisconnected(
+				peer_b,
+			)),
+		)
+		.await;
+
+		// Peer_b reconnects and declares again (but does NOT re-advertise yet).
+		overseer_send(
+			&mut virtual_overseer,
+			CollatorProtocolMessage::NetworkBridgeUpdate(NetworkBridgeEvent::PeerConnected(
+				peer_b,
+				ObservedRole::Full,
+				CollationVersion::V1.into(),
+				None,
+			)),
+		)
+		.await;
+
+		overseer_send(
+			&mut virtual_overseer,
+			CollatorProtocolMessage::NetworkBridgeUpdate(NetworkBridgeEvent::PeerMessage(
+				peer_b,
+				CollationProtocols::V1(protocol_v1::CollatorProtocolMessage::Declare(
+					collator_b.public(),
+					test_state.chain_ids[0],
+					collator_b.sign(&protocol_v1::declare_signature_payload(&peer_b)),
+				)),
+			)),
+		)
+		.await;
+
+		// Complete the first fetch from peer_a.
+		let pov = PoV { block_data: BlockData(vec![]) };
+		let mut candidate_a =
+			dummy_candidate_receipt_bad_sig(dummy_hash(), Some(Default::default()));
+		candidate_a.descriptor.para_id = test_state.chain_ids[0];
+		candidate_a.descriptor.relay_parent = relay_parent;
+		candidate_a.descriptor.persisted_validation_data_hash = dummy_pvd().hash();
+
+		response_channel_a
+			.send(Ok((
+				request_v1::CollationFetchingResponse::Collation(
+					candidate_a.clone().into(),
+					pov.clone(),
+				)
+				.encode(),
+				ProtocolName::from(""),
+			)))
+			.expect("Sending response should succeed");
+
+		// This triggers candidate backing.
+		assert_candidate_backing_second(
+			&mut virtual_overseer,
+			relay_parent,
+			test_state.chain_ids[0],
+			&pov,
+			CollationVersion::V1,
+		)
+		.await;
+
+		// Ensure the subsystem is polled.
+		test_helpers::Yield::new().await;
+
+		// The key assertion: after completing the first fetch, the subsystem should NOT
+		// attempt to fetch from peer_b because their waiting queue entry was cleaned up
+		// on disconnect. Without the fix, we would see a fetch request here that would
+		// fail with "NotAdvertised" because peer_b's advertisement state was cleared
+		// when they disconnected.
+		assert!(
+			overseer_recv_with_timeout(&mut virtual_overseer, Duration::from_millis(100))
+				.await
+				.is_none(),
+			"There should be no fetch request for peer_b - their entry was cleaned from waiting queue on disconnect"
+		);
+
+		virtual_overseer
+	})
+}

prdoc/pr_10906.prdoc

@@ -0,0 +1,16 @@
+title: 'collator-protocol: Remove stale pending collations from the waiting queue'
+doc:
+- audience: Node Dev
+  description: |-
+    This PR removes the stale pending collations from the waiting queue when the peer that advertised the collation disconnects.
+
+    When the peer reconnects, the peer data is freshly created without any prior information about advertised collations.
+    Then the state-pending collation is picked from the queue. The network request will not be emitted since the `fn fetch_collation`  sees no prior advertisement via `peer_data.has_advertised` and returns `Err(FetchError::NotAdvertised)`.
+
+    To avoid this, remove the stale entries immediately when the peer disconnects.
+
+    Part of the stabilization of:
+    - https://github.com/paritytech/polkadot-sdk/issues/10425
+crates:
+- name: polkadot-collator-protocol
+  bump: patch