feat: return error for unpopulated offer if content not available loc…

…ally

ethportal-peertest/src/scenarios/offer_accept.rs

@@ -3,11 +3,10 @@ use std::{
     str::FromStr,
 };
 
-use tokio::time::{sleep, Duration};
 use tracing::info;
 
 use crate::{
-    utils::{fixture_block_body, fixture_header_with_proof, wait_for_history_content},
+    utils::{fixture_header_with_proof, wait_for_history_content},
     Peertest,
 };
 use ethportal_api::{
@@ -51,6 +50,25 @@ pub async fn test_unpopulated_offer(peertest: &Peertest, target: &Client) {
     );
 }
 
+pub async fn test_unpopulated_offer_fails_with_missing_content(
+    peertest: &Peertest,
+    target: &Client,
+) {
+    info!("Testing Unpopulated OFFER/ACCEPT flow with missing content");
+
+    let (content_key, _content_value) = fixture_header_with_proof();
+
+    // validate that unpopulated offer fails if content not available locally
+    assert!(target
+        .offer(
+            Enr::from_str(&peertest.bootnode.enr.to_base64()).unwrap(),
+            content_key.clone(),
+            None,
+        )
+        .await
+        .is_err());
+}
+
 pub async fn test_populated_offer(peertest: &Peertest, target: &Client) {
     info!("Testing Populated Offer/ACCEPT flow");
 
@@ -77,7 +95,7 @@ pub async fn test_populated_offer(peertest: &Peertest, target: &Client) {
 }
 
 pub async fn test_offer_propagates_gossip(peertest: &Peertest, target: &Client) {
-    info!("Testing poke xxx");
+    info!("Testing populated offer propagates gossip");
 
     // connect target to network
     let _ = target.ping(peertest.bootnode.enr.clone()).await.unwrap();
@@ -118,7 +136,6 @@ pub async fn test_offer_propagates_gossip(peertest: &Peertest, target: &Client)
     // connect target to network
     let _ = target.ping(fresh_enr.clone()).await.unwrap();
 
-    // offer header to validate block body later
     let (content_key, content_value) = fixture_header_with_proof();
     // use populated offer which means content will *not* be stored in the target's local db
     target
@@ -130,45 +147,7 @@ pub async fn test_offer_propagates_gossip(peertest: &Peertest, target: &Client)
         .await
         .unwrap();
 
-    sleep(Duration::from_secs(3)).await;
-
-    // check that the target has been offered accumulator_1 after fresh target dropped it
-    // idk do we expect client to store locally after populated offer?
-    assert!(
-        HistoryNetworkApiClient::local_content(target, content_key.clone())
-            .await
-            .is_ok()
-    );
-    assert!(
-        HistoryNetworkApiClient::local_content(&fresh_target, content_key.clone())
-            .await
-            .is_ok()
-    );
-    assert!(HistoryNetworkApiClient::local_content(
-        &peertest.nodes[0].ipc_client,
-        content_key.clone()
-    )
-    .await
-    .is_ok());
-    assert!(HistoryNetworkApiClient::local_content(
-        &peertest.bootnode.ipc_client,
-        content_key.clone()
-    )
-    .await
-    .is_ok());
-
-    // use block body to test larger content over utp
-    let (content_key, content_value) = fixture_block_body();
-    // use populated offer which means content will *not* be stored in the target's local db
-    target
-        .offer(fresh_enr, content_key.clone(), Some(content_value.clone()))
-        .await
-        .unwrap();
-
-    sleep(Duration::from_secs(3)).await;
-
-    // check that the target has been offered accumulator_1 after fresh target dropped it
-    // idk do we expect client to store locally after populated offer?
+    // validate that every node in the network now has a local copy of the header
     assert!(
         HistoryNetworkApiClient::local_content(target, content_key.clone())
             .await

portalnet/src/gossip.rs

@@ -209,7 +209,7 @@ pub async fn trace_propagate_gossip_cross_thread<TContentKey: OverlayContentKey>
 }
 
 /// Filter all nodes from overlay routing table where XOR_distance(content_id, nodeId) < node radius
-pub fn calculate_interested_enrs<TContentKey: OverlayContentKey>(
+fn calculate_interested_enrs<TContentKey: OverlayContentKey>(
     content_key: &TContentKey,
     all_nodes: &[&kbucket::Node<NodeId, Node>],
 ) -> Vec<Enr> {

portalnet/src/overlay/protocol.rs

@@ -531,6 +531,32 @@ where
             destination: enr.clone(),
         };
 
+        // Validate that the content keys are available in the local store, before sending the
+        // offer
+        for content_key in content_keys.into_iter() {
+            let content_key = TContentKey::try_from(content_key.clone()).map_err(|err| {
+                OverlayRequestError::ContentNotFound {
+                    message: format!(
+                        "Error decoding content key for content key: {content_key:02X?} - {err}"
+                    ),
+                    utp: false,
+                    trace: None,
+                }
+            })?;
+            match self.store.read().get(&content_key) {
+                Ok(Some(_)) => {}
+                _ => {
+                    return Err(OverlayRequestError::ContentNotFound {
+                        message: format!(
+                            "Content key not found in local store: {content_key:02X?}"
+                        ),
+                        utp: false,
+                        trace: None,
+                    });
+                }
+            }
+        }
+
         // Send the request and wait on the response.
         match self
             .send_overlay_request(Request::Offer(request), direction)

portalnet/src/overlay/service.rs

@@ -1717,7 +1717,6 @@ where
             }
         };
 
-        println!("xxx overlay propagate_validated_content 2");
         let _ = Self::propagate_validated_content(vec![validated_content], utp_processing).await;
         Ok(())
     }

rpc/src/history_rpc.rs

@@ -170,6 +170,9 @@ impl HistoryNetworkApiServer for HistoryNetworkApi {
     }
 
     /// Send an OFFER request with given ContentKey, to the designated peer and wait for a response.
+    /// If the content value is provided, a "populated" offer is used, which will not store the
+    /// content locally. Otherwise a regular offer is sent, after validating that the content is
+    /// available locally.
     /// Returns the content keys bitlist upon successful content transmission or empty bitlist
     /// receive.
     async fn offer(

tests/self_peertest.rs

@@ -74,6 +74,18 @@ async fn peertest_unpopulated_offer() {
     handle.stop().unwrap();
 }
 
+#[tokio::test(flavor = "multi_thread")]
+#[serial]
+async fn peertest_unpopulated_offer_fails_with_missing_content() {
+    let (peertest, target, handle) = setup_peertest("mainnet").await;
+    peertest::scenarios::offer_accept::test_unpopulated_offer_fails_with_missing_content(
+        &peertest, &target,
+    )
+    .await;
+    peertest.exit_all_nodes();
+    handle.stop().unwrap();
+}
+
 #[tokio::test(flavor = "multi_thread")]
 #[serial]
 async fn peertest_gossip_with_trace() {