feat: validate content before ending query

There are quite a few moving parts here, so the goal was to minimize the
number of functional changes in this refactor.

Overall: add a new query state for while the content is being validated.
Don't exit the query until validation is complete.

Some other changes along the way:

* Moving the bigendian conversion of connection ID

Note that when handling the FindContentQueryResponse::ConnectionId in
findcontent.rs, we don't do the bigendian conversion anymore. That has
been moved to earlier in the process, when handling the Content message,
to reduce the number of places that the conversion needs to be handled.

* The query trace has slightly different end-stage behavior

The received_from peer and the cancelled peers are not set in the query
until the content gets validated.

* Naturally, tests need to reflect the new query structure

For example:
- the query trace test needs to have the content be marked validated to
  progress
- update overlay service tests to use new Validating query step
- immediately mark content as verified, after marking it as received,
  during the test

* bugfix: the content search continues even if many peers don't have it

Formerly, the content query would exit out if too many peers replied
successfully (even if the responses were just more nodes to look up).

Best we can tell, the limit on the number of maximum results is
copy-pasta from the recursive-find-nodes logic. (Confirmation from
Mike)

Now, the query continues until the content is found. It doesn't track
the number of successful responses at all.

ethportal-api/src/types/query_trace.rs

@@ -15,7 +15,7 @@ type ContentId = [u8; 32];
 #[derive(Debug, Serialize, Deserialize, Clone, Eq, PartialEq)]
 #[serde(rename_all = "camelCase")]
 pub struct QueryTrace {
-    /// Node ID from which the content was received. None if the content was not found.
+    /// Node ID from which the content was received. None if the content was not found & verified.
     pub received_from: Option<NodeId>,
     /// The local node.
     pub origin: NodeId,
@@ -71,10 +71,13 @@ impl QueryTrace {
         self.add_metadata(enr, true);
     }
 
-    /// Mark the node that responded with the content, and when it was received.
+    /// Mark that a node responded with content, and when it was received.
+    /// Multiple nodes might respond, for a variety of reasons, including:
+    /// - they sent content that was later invalidated
+    /// - they claimed to have content for utp transfer, but disappeared before transmission
+    /// - they sent content too slowly, and we found it elsewhere
     pub fn node_responded_with_content(&mut self, enr: &Enr) {
         let node_id = enr.into();
-        self.received_from = Some(node_id);
         let timestamp_u64 = QueryTrace::timestamp_millis_u64(self.started_at_ms);
         self.responses.insert(
             node_id,
@@ -86,6 +89,13 @@ impl QueryTrace {
         self.add_metadata(enr, true);
     }
 
+    /// Mark the node that sent the content that was finally verified.
+    pub fn content_validated(&mut self, node_id: NodeId) {
+        if self.received_from.is_none() {
+            self.received_from = Some(node_id);
+        }
+    }
+
     /// Returns milliseconds since the time provided.
     fn timestamp_millis_u64(since: SystemTime) -> u64 {
         let timestamp_millis_u128 = SystemTime::now()
@@ -177,6 +187,7 @@ mod tests {
         tracer.node_responded_with(&enr_a, vec![]);
         tracer.node_responded_with(&enr_b, vec![&enr_d]);
         tracer.node_responded_with_content(&enr_c);
+        tracer.content_validated(*node_id_c);
 
         let origin_entry = tracer.responses.get(local_node_id).unwrap();
 

portalnet/Cargo.toml

@@ -15,6 +15,7 @@ alloy-primitives.workspace = true
 anyhow.workspace = true
 async-trait.workspace = true
 bytes.workspace = true
+crossbeam-channel = "0.5.13"
 delay_map.workspace = true
 directories.workspace = true
 discv5.workspace = true