Emit global uTP listener events when uTP stream is closed or reset

Author: ogenev

Cargo.lock

@@ -1 +1,6 @@
-Process all closed uTP streams in UtpListener and pass the payload to overlay service.
+- Rename `UtpSocket` to `UtpStream`.
+- Refactor the way we are storing the received payload (DATA packets) in the uTP stream. 
+- Add a new AddActiveConnection UtpListener request and move the initialization of a uTP stream inside UtpListener.
+- Add UtpStream -> UtpListener event channel and emit event inside UtpStream when stream state changes to Closed or Reset.
+- Emit a global uTP listener event containing a uTP payload when a stream is closed.
+- Remove redundant and dead code.

newsfragments/325.added.md

@@ -69,3 +69,4 @@ features = ["bundled"]
 
 [dev-dependencies]
 quickcheck = "1.0.3"
+ntest = "0.8.0"

trin-core/Cargo.toml

@@ -14,6 +14,7 @@ use rand::seq::SliceRandom;
 use serde_json::{json, Value};
 use std::{
     convert::TryFrom,
+    fmt,
     net::{IpAddr, SocketAddr},
     sync::Arc,
     time::Duration,
@@ -54,6 +55,18 @@ pub struct Discovery {
     pub listen_socket: SocketAddr,
 }
 
+impl fmt::Debug for Discovery {
+    fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
+        write!(
+            f,
+            "Discovery: ( enr: {}, started: {}, listen_socket: {} )",
+            self.discv5.local_enr(),
+            self.started,
+            self.listen_socket
+        )
+    }
+}
+
 impl Discovery {
     pub fn new(portal_config: PortalnetConfig) -> Result<Self, String> {
         let listen_all_ips = SocketAddr::new("0.0.0.0".parse().unwrap(), portal_config.listen_port);

trin-core/src/portalnet/discovery.rs

@@ -18,7 +18,7 @@ use crate::portalnet::{
 use crate::{
     portalnet::types::content_key::RawContentKey,
     utp::{
-        stream::{UtpListenerRequest, UtpSocket, BUF_SIZE},
+        stream::{UtpListenerRequest, UtpStream, BUF_SIZE},
         trin_helpers::{UtpAccept, UtpMessage, UtpStreamId},
     },
 };
@@ -281,49 +281,41 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         conn_id: u16,
     ) -> Result<Content, OverlayRequestError> {
         // initiate the connection to the acceptor
-        let (tx, rx) = tokio::sync::oneshot::channel::<anyhow::Result<UtpSocket>>();
-        self.utp_listener_tx
-            .send(UtpListenerRequest::Connect(
-                conn_id,
-                enr.node_id(),
-                self.protocol.clone(),
-                UtpStreamId::FindContentStream,
-                tx,
+        let (tx, rx) = tokio::sync::oneshot::channel::<UtpStream>();
+        let utp_request = UtpListenerRequest::Connect(
+            conn_id,
+            enr,
+            self.protocol.clone(),
+            UtpStreamId::FindContentStream,
+            tx,
+        );
+
+        self.utp_listener_tx.send(utp_request).map_err(|err| {
+            OverlayRequestError::UtpError(format!(
+                "Unable to send Connect request with FindContent stream to UtpListener: {err}"
             ))
-            .map_err(|err| {
-                OverlayRequestError::UtpError(format!(
-                    "Unable to send Connect request with FindContent stream to UtpListener: {err}"
-                ))
-            })?;
+        })?;
 
         match rx.await {
-            Ok(conn) => {
-                match conn {
-                    Ok(mut conn) => {
-                        let mut result = Vec::new();
-                        // Loop and receive all DATA packets, similar to `read_to_end`
-                        loop {
-                            let mut buf = [0; BUF_SIZE];
-                            match conn.recv_from(&mut buf).await {
-                                Ok((0, _)) => {
-                                    break;
-                                }
-                                Ok((bytes, _)) => {
-                                    result.extend_from_slice(&mut buf[..bytes]);
-                                }
-                                Err(err) => {
-                                    warn!("Unable to receive content via uTP: {err}");
-                                    return Err(OverlayRequestError::UtpError(err.to_string()));
-                                }
-                            }
+            Ok(mut conn) => {
+                let mut result = Vec::new();
+                // Loop and receive all DATA packets, similar to `read_to_end`
+                loop {
+                    let mut buf = [0; BUF_SIZE];
+                    match conn.recv_from(&mut buf).await {
+                        Ok((0, _)) => {
+                            break;
+                        }
+                        Ok((bytes, _)) => {
+                            result.extend_from_slice(&mut buf[..bytes]);
+                        }
+                        Err(err) => {
+                            warn!("Unable to receive content via uTP: {err}");
+                            return Err(OverlayRequestError::UtpError(err.to_string()));
                         }
-                        Ok(Content::Content(VariableList::from(result)))
-                    }
-                    Err(err) => {
-                        warn!("Unable to initiate uTP stream with remote node. Error initializing uTP socket: {err}");
-                        Err(OverlayRequestError::UtpError(err.to_string()))
                     }
                 }
+                Ok(Content::Content(VariableList::from(result)))
             }
             Err(err) => {
                 warn!("Unable to receive from uTP listener channel: {err}");
@@ -390,47 +382,43 @@ impl<TContentKey: OverlayContentKey + Send, TMetric: Metric + Send>
         }
 
         // initiate the connection to the acceptor
-        let (tx, rx) = tokio::sync::oneshot::channel::<anyhow::Result<UtpSocket>>();
-
-        self.utp_listener_tx.send(UtpListenerRequest::Connect(
+        let (tx, rx) = tokio::sync::oneshot::channel::<UtpStream>();
+        let utp_request = UtpListenerRequest::Connect(
             conn_id,
-            enr.node_id(),
+            enr,
             self.protocol.clone(),
             UtpStreamId::OfferStream,
             tx,
-        )).map_err(|err| anyhow!("Unable to send Connect request to UtpListener when processing ACCEPT message: {err}"))?;
+        );
 
-        match rx.await? {
-            Ok(mut conn) => {
-                // Handle STATE packet for SYN
-                let mut buf = [0; BUF_SIZE];
-                conn.recv(&mut buf).await?;
-
-                let content_items = self.provide_requested_content(&response, content_keys_offered);
-
-                let content_message = UtpAccept {
-                    message: content_items,
-                };
-
-                tokio::spawn(async move {
-                    // send the content to the acceptor over a uTP stream
-                    if let Err(err) = conn
-                        .send_to(&UtpMessage::new(content_message.as_ssz_bytes()).encode()[..])
-                        .await
-                    {
-                        warn!("Error sending content {err}");
-                    };
-                    // Close uTP connection
-                    if let Err(err) = conn.close().await {
-                        warn!("Unable to close uTP connection!: {err}")
-                    };
-                });
-                Ok(response)
-            }
-            Err(err) => Err(anyhow!(
-                "Unable to initialize Offer uTP stream with remote node: {err}"
-            )),
-        }
+        self.utp_listener_tx
+            .send(utp_request).map_err(|err| anyhow!("Unable to send Connect request to UtpListener when processing ACCEPT message: {err}"))?;
+
+        let mut conn = rx.await?;
+        // Handle STATE packet for SYN
+        let mut buf = [0; BUF_SIZE];
+        conn.recv(&mut buf).await?;
+
+        let content_items = self.provide_requested_content(&response, content_keys_offered);
+
+        let content_message = UtpAccept {
+            message: content_items,
+        };
+
+        tokio::spawn(async move {
+            // send the content to the acceptor over a uTP stream
+            if let Err(err) = conn
+                .send_to(&UtpMessage::new(content_message.as_ssz_bytes()).encode()[..])
+                .await
+            {
+                warn!("Error sending content {err}");
+            };
+            // Close uTP connection
+            if let Err(err) = conn.close().await {
+                warn!("Unable to close uTP connection!: {err}")
+            };
+        });
+        Ok(response)
     }
 
     /// Provide the requested content key and content value for the acceptor