feat: add trace offer for a series of content keys (#1830)

Author: KolbyML

crates/ethportal-api/src/types/portal_wire.rs

@@ -473,9 +473,9 @@ impl From<PopulatedOffer> for Offer {
 #[derive(Debug, Clone)]
 pub struct PopulatedOfferWithResult {
     /// The offered content key & value
-    pub content_item: (RawContentKey, RawContentValue),
+    pub content_items: Vec<(RawContentKey, RawContentValue)>,
     /// The channel to send the result of the offer to
-    pub result_tx: tokio::sync::mpsc::UnboundedSender<OfferTrace>,
+    pub result_tx: tokio::sync::mpsc::UnboundedSender<OfferTraceMultipleItems>,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
@@ -486,11 +486,22 @@ pub enum OfferTrace {
     Failed,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub enum OfferTraceMultipleItems {
+    /// Offer was successful sent and received a response
+    Success(AcceptCodeList),
+    /// This offer failed, perhaps locally or from a timeout or transfer failure
+    Failed,
+}
+
 impl From<PopulatedOfferWithResult> for Offer {
-    fn from(val: PopulatedOfferWithResult) -> Self {
-        Self {
-            content_keys: vec![val.content_item.0],
-        }
+    fn from(value: PopulatedOfferWithResult) -> Self {
+        let content_keys = value
+            .content_items
+            .into_iter()
+            .map(|(key, _val)| key)
+            .collect();
+        Self { content_keys }
     }
 }
 

crates/portalnet/src/overlay/protocol.rs

@@ -31,8 +31,8 @@ use ethportal_api::{
         },
         portal::PutContentInfo,
         portal_wire::{
-            Content, FindContent, FindNodes, Message, Nodes, OfferTrace, Ping, Pong,
-            PopulatedOffer, PopulatedOfferWithResult, Request, Response,
+            Content, FindContent, FindNodes, Message, Nodes, OfferTrace, OfferTraceMultipleItems,
+            Ping, Pong, PopulatedOffer, PopulatedOfferWithResult, Request, Response,
         },
     },
     utils::bytes::hex_encode,
@@ -609,10 +609,34 @@ impl<
         content_key: RawContentKey,
         content_value: RawContentValue,
     ) -> Result<OfferTrace, OverlayRequestError> {
+        match self
+            .send_offer_trace_with_multiple_items(enr, vec![(content_key, content_value)])
+            .await?
+        {
+            OfferTraceMultipleItems::Success(accept_code_list) => {
+                if accept_code_list.len() != 1 {
+                    return Err(OverlayRequestError::Failure(format!(
+                        "Expected exactly one AcceptCode in the response. Got {}",
+                        accept_code_list.len()
+                    )));
+                }
+
+                Ok(OfferTrace::Success(accept_code_list[0]))
+            }
+            OfferTraceMultipleItems::Failed => Ok(OfferTrace::Failed),
+        }
+    }
+
+    /// Send Offer request with trace, without storing the content into db, with multiple items
+    pub async fn send_offer_trace_with_multiple_items(
+        &self,
+        enr: Enr,
+        content_items: Vec<(RawContentKey, RawContentValue)>,
+    ) -> Result<OfferTraceMultipleItems, OverlayRequestError> {
         // Construct the request.
         let (result_tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
         let request = Request::PopulatedOfferWithResult(PopulatedOfferWithResult {
-            content_item: (content_key, content_value),
+            content_items,
             result_tx,
         });
 

crates/portalnet/src/overlay/service/offer.rs

@@ -9,7 +9,9 @@ use ethportal_api::{
         distance::Metric,
         enr::Enr,
         network_spec::network_spec,
-        portal_wire::{Accept, Content, FindContent, Offer, OfferTrace, Request, Response},
+        portal_wire::{
+            Accept, Content, FindContent, Offer, OfferTraceMultipleItems, Request, Response,
+        },
         protocol_versions::ProtocolVersion,
     },
     OverlayContentKey, RawContentKey, RawContentValue,
@@ -358,13 +360,9 @@ impl<
         // keys
         if content_keys.all_declined() {
             if let Some(tx) = gossip_result_tx {
-                if content_keys.len() != 1 {
-                    error!(
-                        "OfferTrace only supports one content key at a time, got {}",
-                        content_keys.len()
-                    );
+                if let Err(err) = tx.send(OfferTraceMultipleItems::Success(content_keys)) {
+                    warn!(%err, "Unable to send OfferTrace result all keys declined");
                 }
-                let _ = tx.send(OfferTrace::Success(content_keys[0]));
             }
             return Ok(());
         }
@@ -392,7 +390,7 @@ impl<
                     .collect()),
                 Request::PopulatedOfferWithResult(offer) => Ok(content_keys
                     .iter()
-                    .zip(vec![offer.content_item])
+                    .zip(offer.content_items)
                     .filter(|(is_accepted, _item)| **is_accepted == AcceptCode::Accepted)
                     .map(|(_is_accepted, (_key, val))| val)
                     .collect()),
@@ -411,7 +409,9 @@ impl<
                         "Error decoding previously offered content items"
                     );
                     if let Some(tx) = gossip_result_tx {
-                        let _ = tx.send(OfferTrace::Failed);
+                        if let Err(err) = tx.send(OfferTraceMultipleItems::Failed) {
+                            warn!(%err, "Unable to send OfferTrace Failed result for decoding offered content items");
+                        }
                     }
                     return;
                 }
@@ -422,7 +422,9 @@ impl<
                 Err(err) => {
                     warn!(%err, "Unable to build content payload");
                     if let Some(tx) = gossip_result_tx {
-                        let _ = tx.send(OfferTrace::Failed);
+                        if let Err(err) = tx.send(OfferTraceMultipleItems::Failed) {
+                            warn!(%err, "Unable to send OfferTrace Failed result failed to build content payload");
+                        }
                     }
                     return;
                 }
@@ -431,16 +433,13 @@ impl<
                 .connect_outbound_stream(cid, peer, &content_payload)
                 .await;
             if let Some(tx) = gossip_result_tx {
-                if result {
-                    if content_keys.len() != 1 {
-                        error!(
-                            "OfferTrace only supports one content key at a time, got {}",
-                            content_keys.len()
-                        );
-                    }
-                    let _ = tx.send(OfferTrace::Success(content_keys[0]));
+                let result = if result {
+                    OfferTraceMultipleItems::Success(content_keys)
                 } else {
-                    let _ = tx.send(OfferTrace::Failed);
+                    OfferTraceMultipleItems::Failed
+                };
+                if let Err(err) = tx.send(result) {
+                    warn!(%err, "Unable to send OfferTrace result");
                 }
             }
             // explicitly drop permit in the thread so the permit is included in the thread