handle any incoming PoVs and redistribute

node/network/pov-distribution/src/lib.rs

@@ -267,6 +267,7 @@ async fn report_peer(
 	ctx.send_message(AllMessages::NetworkBridge(NetworkBridgeMessage::ReportPeer(peer, rep))).await
 }
 
+/// Handle a notification from a peer that they are awaiting some PoVs.
 async fn handle_awaiting(
 	state: &mut State,
 	ctx: &mut impl SubsystemContext<Message = PoVDistributionMessage>,
@@ -319,6 +320,9 @@ async fn handle_awaiting(
 	Ok(())
 }
 
+/// Handle an incoming PoV from our peer. Reports them if unexpected, rewards them if not.
+///
+/// Completes any requests awaiting that PoV.
 async fn handle_incoming_pov(
 	state: &mut State,
 	ctx: &mut impl SubsystemContext<Message = PoVDistributionMessage>,
@@ -327,8 +331,61 @@ async fn handle_incoming_pov(
 	pov_hash: Hash,
 	pov: PoV,
 ) -> SubsystemResult<()> {
-	// TODO [now]
-	Ok(())
+	let relay_parent_state = match state.relay_parent_state.get_mut(&relay_parent) {
+		None =>	{
+			report_peer(ctx, peer, COST_UNEXPECTED_POV).await?;
+			return Ok(());
+		},
+		Some(r) => r,
+	};
+
+	let pov = {
+		// Do validity checks and complete all senders awaiting this PoV.
+		let fetching = match relay_parent_state.fetching.get_mut(&pov_hash) {
+			None => {
+				report_peer(ctx, peer, COST_UNEXPECTED_POV).await?;
+				return Ok(());
+			}
+			Some(f) => f,
+		};
+
+		let hash = pov.hash();
+		if hash != pov_hash {
+			report_peer(ctx, peer, COST_UNEXPECTED_POV).await?;
+			return Ok(());
+		}
+
+		let pov = Arc::new(pov);
+
+		if fetching.is_empty() {
+			// fetching is empty whenever we were awaiting something and
+			// it was completed afterwards.
+			report_peer(ctx, peer.clone(), BENEFIT_LATE_POV).await?;
+		} else {
+			// fetching is non-empty when the peer just provided us with data we needed.
+			report_peer(ctx, peer.clone(), BENEFIT_FRESH_POV).await?;
+		}
+
+		for response_sender in fetching.drain(..) {
+			let _ = response_sender.send(pov.clone());
+		}
+
+		pov
+	};
+
+	// make sure we don't consider this peer as awaiting that PoV anymore.
+	if let Some(peer_state) = state.peer_state.get_mut(&peer) {
+		peer_state.awaited.remove(&pov_hash);
+	}
+
+	// distribute the PoV to all other peers who are awaiting it.
+	distribute_to_awaiting(
+		&mut state.peer_state,
+		ctx,
+		relay_parent,
+		pov_hash,
+		&*pov,
+	).await
 }
 
 /// Handles a network bridge update.

roadmap/implementors-guide/src/node/backing/pov-distribution.md

@@ -105,10 +105,11 @@ Here is the logic of the state machine:
 		- For each new `pov_hash` in `pov_hashes`, if there is a `pov` under `pov_hash` in the `known` map, send the peer a `NetworkMessage::SendPoV(relay_parent, pov_hash, pov)`.
 		- Otherwise, add the `pov_hash` to the `awaited` map
 	- If this is `NetworkMessage::SendPoV(relay_parent, pov_hash, pov)`:
-		- If there is no entry under `relay_parent` in `relay_parent_state` or no entry under `pov_hash` in our `awaited` map for that `relay_parent`, report and ignore.
+		- If there is no entry under `relay_parent` in `relay_parent_state` or no entry under `pov_hash` in our `fetching` map for that `relay_parent`, report and ignore.
 		- If the blake2-256 hash of the pov doesn't equal `pov_hash`, report and ignore.
 		- Complete and remove any listeners in the `fetching` map under `pov_hash`. However, leave an empty set of listeners in the `fetching` map to denote that this was something we once awaited. This will allow us to recognize peers who have sent us something we were expecting, but just a little late.
 		- Add to `known` map.
+		- Remove the `pov_hash` from the `peer.awaited` map, if any.
 		- Send `NetworkMessage::SendPoV(relay_parent, descriptor.pov_hash, PoV)` to all peers who have the `descriptor.pov_hash` in the set under `relay_parent` in the `peer.awaited` map and remove the entry from `peer.awaited`.
 - On `PeerViewChange(peer_id, view)`
 	- If Peer is unknown, ignore.