Merge pull request #324 from ogenev/handle-offer-refactor

Refactor handling `OFFER` req and update `should_store` db method

Author: ogenev

Cargo.lock

@@ -14,7 +14,7 @@ futures = "0.3.21"
 hyper = { version = "0.14", features = ["full"] }
 hex = "0.4.3"
 log = "0.4.14"
-rocksdb = "0.16.0"
+rocksdb = "0.18.0"
 serde_json = "1.0.59"
 structopt = "0.3"
 tokio = {version = "1.14.0", features = ["full"]}

ethportal-peertest/Cargo.toml

@@ -0,0 +1 @@
+Remove the dummy should_store method when handling OFFER requests and accept only content we are interested in.

newsfragments/324.fixed.md

@@ -32,7 +32,7 @@ parking_lot = "0.11.2"
 prometheus_exporter = "0.8.4"
 rand = "0.8.4"
 rlp = "0.5.0"
-rocksdb = "0.16.0"
+rocksdb = "0.18.0"
 rstest = "0.11.0"
 r2d2 = "0.8.9"
 r2d2_sqlite = "0.19.0"

trin-core/Cargo.toml

@@ -81,15 +81,15 @@ pub enum OverlayRequestError {
     Discv5Error(discv5::RequestError),
 
     /// Error types resulting from building ACCEPT message
-    #[error("Error while building accept message")]
-    AcceptError(ssz_types::Error),
+    #[error("Error while building accept message: {0}")]
+    AcceptError(String),
 
     /// Error types resulting from building ACCEPT message
-    #[error("Error while sending offer message")]
+    #[error("Error while sending offer message: {0}")]
     OfferError(String),
 
     /// uTP request error
-    #[error("uTP request error")]
+    #[error("uTP request error: {0}")]
     UtpError(String),
 }
 
@@ -643,27 +643,46 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         }
     }
 
-    /// Attempts to build a `Accept` response for a `Offer` request.
+    /// Attempts to build an `Accept` response for an `Offer` request.
     fn handle_offer(&self, request: Offer) -> Result<Accept, OverlayRequestError> {
         self.metrics
             .as_ref()
             .and_then(|m| Some(m.report_inbound_offer()));
-        let mut requested_keys = BitList::with_capacity(request.content_keys.len())
-            .map_err(|e| OverlayRequestError::AcceptError(e))?;
-        let conn_id: u16 = crate::utp::stream::rand();
 
-        // TODO: Pipe this with overlay DB and request only not available keys.
+        let mut requested_keys =
+            BitList::with_capacity(request.content_keys.len()).map_err(|_| {
+                OverlayRequestError::AcceptError(
+                    "Unable to initialize bitlist for requested keys.".to_owned(),
+                )
+            })?;
+
         let accept_keys = request.content_keys.clone();
 
-        for (i, key) in request.content_keys.iter().enumerate() {
-            // should_store is currently a dummy function
-            // the actual function will take ContentKey type, so we'll  have to decode keys here
+        for (i, key) in request.content_keys.into_iter().enumerate() {
+            let key = (TContentKey::try_from)(key).map_err(|_| {
+                OverlayRequestError::AcceptError(format!(
+                    "Unable to build content key from OFFER request."
+                ))
+            })?;
+
             requested_keys
-                .set(i, should_store(key))
-                .map_err(|e| OverlayRequestError::AcceptError(e))?;
+                .set(
+                    i,
+                    self.storage.read().should_store(&key).map_err(|err| {
+                        OverlayRequestError::AcceptError(format!(
+                            "Unable to check content availability: {err}"
+                        ))
+                    })?,
+                )
+                .map_err(|err| {
+                    OverlayRequestError::AcceptError(format!(
+                        "Unable to set requested keys bits: {err:?}"
+                    ))
+                })?;
         }
 
         // listen for incoming connection request on conn_id, as part of utp handshake
+        let conn_id: u16 = crate::utp::stream::rand();
         let utp_request = UtpListenerRequest::OfferStream(conn_id);
         if let Err(err) = self.utp_listener_tx.send(utp_request) {
             return Err(OverlayRequestError::UtpError(format!(
@@ -1809,7 +1828,3 @@ mod tests {
         assert_eq!(target_bucket_idx, expected_index as u8);
     }
 }
-
-fn should_store(_key: &Vec<u8>) -> bool {
-    return true;
-}

trin-core/src/portalnet/overlay_service.rs

@@ -122,7 +122,7 @@ impl PortalStorage {
     pub fn should_store(&self, key: &impl OverlayContentKey) -> Result<bool, PortalStorageError> {
         let content_id = key.content_id();
         // Don't store if we already have the data
-        match self.db.get(&content_id) {
+        match self.db.get_pinned(&content_id) {
             Ok(Some(_)) => return Ok(false),
             Err(e) => return Err(PortalStorageError::RocksDB(e)),
             _ => (),

trin-core/src/portalnet/storage.rs

@@ -1,3 +1,4 @@
+use anyhow::anyhow;
 use ethereum_types::{U256, U512};
 use sha2::{Digest as Sha2Digest, Sha256};
 use sha3::{Digest, Keccak256};
@@ -28,12 +29,12 @@ impl IdentityContentKey {
 }
 
 impl TryFrom<Vec<u8>> for IdentityContentKey {
-    type Error = String;
+    type Error = anyhow::Error;
 
     fn try_from(value: Vec<u8>) -> Result<Self, Self::Error> {
         // Require that length of input is equal to 32.
         if value.len() != 32 {
-            return Err(String::from("Input Vec has invalid length"));
+            return Err(anyhow!("Input Vec has invalid length"));
         }
 
         // The following will not panic because of the length check above.

trin-core/src/portalnet/types/content_key.rs

@@ -16,7 +16,7 @@ keccak-hash = "0.8.0"
 log = "0.4.14"
 parking_lot = "0.11.2"
 rlp = "0.5.0"
-rocksdb = "0.16.0"
+rocksdb = "0.18.0"
 serde_json = "1.0.59"
 serial_test = "0.5.1"
 tempdir = "0.3.7"

trin-history/Cargo.toml

@@ -22,4 +22,4 @@ tracing-subscriber = "0.2.18"
 tracing-futures = "0.2.5"
 tokio = {version = "1.8.0", features = ["full"]}
 trin-core = { path = "../trin-core" }
-rocksdb = "0.16.0"
+rocksdb = "0.18.0"