Upgrade litep2p to v0.12.0 and handle Kademlia publish success events (#9685)

litep2p v0.12.0 adds ability to track whether publishing a DHT record or
provider was successful. This PR brings this functionality to substrate.
Particularly, this fixes authority-discovery unnecessarily republishing
DHT records due to litep2p not emitting
`KademliaEvent::PutRecordSuccess` before v0.12.0.

---------

Co-authored-by: cmd[bot] <41898282+github-actions[bot]@users.noreply.github.com>

Author: dmitry-markin

Cargo.lock

@@ -908,7 +908,7 @@ linked-hash-map = { version = "0.5.4" }
 linked_hash_set = { version = "0.1.4" }
 linregress = { version = "0.5.1" }
 lite-json = { version = "0.2.0", default-features = false }
-litep2p = { version = "0.11.0", features = ["rsa", "websocket"] }
+litep2p = { version = "0.12.0", features = ["rsa", "websocket"] }
 log = { version = "0.4.22", default-features = false }
 macro_magic = { version = "0.5.1" }
 maplit = { version = "1.0.2" }

Cargo.toml

@@ -0,0 +1,12 @@
+title: Upgrade litep2p to v0.12.0
+doc:
+- audience: Node Dev
+  description: litep2p v0.12.0 adds ability to track whether publishing a DHT record
+    or provider was successful. This PR brings this functionality to substrate. Particularly,
+    this fixes authority-discovery unnecessarily republishing DHT records due to litep2p
+    not emitting `KademliaEvent::PutRecordSuccess` before v0.12.0.
+crates:
+- name: sc-network
+  bump: major
+- name: sc-network-types
+  bump: major

prdoc/pr_9685.prdoc

@@ -48,14 +48,12 @@ pub enum DhtEvent {
 	ValueNotFound(Key),
 
 	/// The record has been successfully inserted into the DHT.
-	// TODO: this is not implemented with litep2p network backend.
 	ValuePut(Key),
 
 	/// An error has occurred while putting a record into the DHT.
 	ValuePutFailed(Key),
 
 	/// Successfully started providing the given key.
-	// TODO: this is not implemented with litep2p network backend.
 	StartedProviding(Key),
 
 	/// An error occured while registering as a content provider on the DHT.

substrate/client/network/src/event.rs

@@ -80,6 +80,16 @@ const MAX_EXTERNAL_ADDRESSES: u32 = 32;
 /// external.
 const MIN_ADDRESS_CONFIRMATIONS: usize = 3;
 
+/// Quorum threshold to interpret `PUT_VALUE` & `ADD_PROVIDER` as successful.
+///
+/// As opposed to libp2p, litep2p does not finish the query as soon as the required number of
+/// peers have reached. Instead, it tries to put the record to all target peers (typically 20) and
+/// uses the quorum setting only to determine the success of the query.
+///
+/// We set the threshold to 50% of the target peers to account for unreachable peers. The actual
+/// number of stored records may be higher.
+const QUORUM_THRESHOLD: NonZeroUsize = NonZeroUsize::new(10).expect("10 > 0; qed");
+
 /// Discovery events.
 #[derive(Debug)]
 pub enum DiscoveryEvent {
@@ -174,6 +184,14 @@ pub enum DiscoveryEvent {
 		providers: Vec<ContentProvider>,
 	},
 
+	/// Provider was successfully published.
+	AddProviderSuccess {
+		/// Query ID.
+		query_id: QueryId,
+		/// Provided key.
+		provided_key: RecordKey,
+	},
+
 	/// Query failed.
 	QueryFailed {
 		/// Query ID.
@@ -401,7 +419,10 @@ impl Discovery {
 	/// Publish value on the DHT using Kademlia `PUT_VALUE`.
 	pub async fn put_value(&mut self, key: KademliaKey, value: Vec<u8>) -> QueryId {
 		self.kademlia_handle
-			.put_record(Record::new(RecordKey::new(&key.to_vec()), value))
+			.put_record(
+				Record::new(RecordKey::new(&key.to_vec()), value),
+				Quorum::N(QUORUM_THRESHOLD),
+			)
 			.await
 	}
 
@@ -417,6 +438,9 @@ impl Discovery {
 				record,
 				peers.into_iter().map(|peer| peer.into()).collect(),
 				update_local_storage,
+				// These are the peers that just returned the record to us in authority-discovery,
+				// so we assume they are all reachable.
+				Quorum::All,
 			)
 			.await
 	}
@@ -446,8 +470,10 @@ impl Discovery {
 	}
 
 	/// Start providing `key`.
-	pub async fn start_providing(&mut self, key: KademliaKey) {
-		self.kademlia_handle.start_providing(key.into()).await;
+	pub async fn start_providing(&mut self, key: KademliaKey) -> QueryId {
+		self.kademlia_handle
+			.start_providing(key.into(), Quorum::N(QUORUM_THRESHOLD))
+			.await
 	}
 
 	/// Stop providing `key`.
@@ -680,6 +706,17 @@ impl Stream for Discovery {
 					providers,
 				}))
 			},
+			Poll::Ready(Some(KademliaEvent::AddProviderSuccess { query_id, provided_key })) => {
+				log::trace!(
+					target: LOG_TARGET,
+					"`ADD_PROVIDER` for {query_id:?} with {provided_key:?} succeeded",
+				);
+
+				return Poll::Ready(Some(DiscoveryEvent::AddProviderSuccess {
+					query_id,
+					provided_key,
+				}))
+			},
 			// We do not validate incoming providers.
 			Poll::Ready(Some(KademliaEvent::IncomingProvider { .. })) => {},
 		}