Merge branch 'feature/basic-overlay'

network/Cargo.toml

@@ -22,6 +22,7 @@ everscale-crypto = { version = "0.2", features = ["tl-proto"] }
 futures-util = { version = "0.3", features = ["sink"] }
 hex = "0.4"
 moka = { version = "0.12", features = ["sync"] }
+parking_lot = "0.12"
 pin-project-lite = "0.2"
 pkcs8 = "0.10"
 quinn = { version = "0.10", default-features = false, features = ["runtime-tokio", "tls-rustls"] }

network/src/dht/mod.rs

@@ -509,7 +509,7 @@ impl DhtInner {
         *self.local_peer_info.lock().unwrap() = Some(peer_info);
     }
 
-    #[tracing::instrument(level = "debug", skip_all, fields(local_id = % self.local_id))]
+    #[tracing::instrument(level = "debug", skip_all, fields(local_id = %self.local_id))]
     async fn announce_local_peer_info(&self, network: &Network) -> Result<()> {
         let data = tl_proto::serialize(&[network.local_addr().into()] as &[Address]);
 
@@ -524,7 +524,7 @@ impl DhtInner {
         self.store_value(network, ValueRef::Peer(value), true).await
     }
 
-    #[tracing::instrument(level = "debug", skip_all, fields(local_id = % self.local_id))]
+    #[tracing::instrument(level = "debug", skip_all, fields(local_id = %self.local_id))]
     async fn refresh_routing_table(&self, network: &Network) {
         const PARALLEL_QUERIES: usize = 3;
         const MAX_DISTANCE: usize = 15;

network/src/dht/routing.rs

@@ -203,9 +203,9 @@ mod tests {
 
     #[test]
     fn buckets_are_sets() {
-        let mut table = RoutingTable::new(PeerId::random());
+        let mut table = RoutingTable::new(rand::random());
 
-        let peer = PeerId::random();
+        let peer = rand::random();
         assert!(table.add(
             make_node(peer),
             MAX_K,
@@ -223,7 +223,7 @@ mod tests {
 
     #[test]
     fn sould_not_add_seld() {
-        let local_id = PeerId::random();
+        let local_id = rand::random();
         let mut table = RoutingTable::new(local_id);
 
         assert!(!table.add(
@@ -243,14 +243,14 @@ mod tests {
 
         for _ in 0..k {
             assert!(bucket.insert(
-                make_node(PeerId::random()),
+                make_node(rand::random()),
                 k,
                 &timeout,
                 RoutingTableSource::Trusted
             ));
         }
         assert!(!bucket.insert(
-            make_node(PeerId::random()),
+            make_node(rand::random()),
             k,
             &timeout,
             RoutingTableSource::Trusted

network/src/lib.rs

@@ -1,3 +1,9 @@
+pub use self::overlay::{
+    OverlayConfig, OverlayId, OverlayService, OverlayServiceBackgroundTasks, OverlayServiceBuilder,
+    PrivateOverlay, PrivateOverlayBuilder, PrivateOverlayEntries, PrivateOverlayEntriesReadGuard,
+    PrivateOverlayEntriesWriteGuard, PublicOverlay, PublicOverlayBuilder, PublicOverlayEntries,
+    PublicOverlayEntriesReadGuard,
+};
 pub use self::util::{check_peer_signature, NetworkExt, Routable, Router, RouterBuilder};
 pub use dht::{
     xor_distance, DhtClient, DhtClientBuilder, DhtConfig, DhtQueryBuilder, DhtQueryWithDataBuilder,
@@ -18,11 +24,13 @@ pub use quinn;
 
 mod dht;
 mod network;
+mod overlay;
 mod types;
 mod util;
 
 pub mod proto {
     pub mod dht;
+    pub mod overlay;
 }
 
 #[doc(hidden)]
@@ -39,17 +47,30 @@ mod tests {
     #[tokio::test]
     async fn init_works() {
         let keypair = everscale_crypto::ed25519::KeyPair::generate(&mut rand::thread_rng());
+        let peer_id: PeerId = keypair.public_key.into();
+
+        let private_overlay = PrivateOverlay::builder(rand::random())
+            .build(service_message_fn(|_| futures_util::future::ready(())));
+
+        let public_overlay = PublicOverlay::builder(rand::random())
+            .build(service_message_fn(|_| futures_util::future::ready(())));
+
+        let (overlay_tasks, overlay_service) = OverlayService::builder(peer_id)
+            .with_private_overlay(&private_overlay)
+            .with_public_overlay(&public_overlay)
+            .build();
 
-        let (dht_client, dht) = DhtService::builder(keypair.public_key.into()).build();
+        let (dht_client, dht) = DhtService::builder(peer_id).build();
 
-        let router = Router::builder().route(dht).build();
+        let router = Router::builder().route(dht).route(overlay_service).build();
 
         let network = Network::builder()
             .with_random_private_key()
             .with_service_name("test-service")
             .build((Ipv4Addr::LOCALHOST, 0), router)
             .unwrap();
 
-        let _dht_client = dht_client.build(network);
+        let _dht_client = dht_client.build(network.clone());
+        overlay_tasks.spawn(network);
     }
 }

network/src/overlay/config.rs

@@ -0,0 +1,41 @@
+use std::time::Duration;
+
+use serde::{Deserialize, Serialize};
+use tycho_util::serde_helpers;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(default)]
+pub struct OverlayConfig {
+    /// Maximum time to live for public overlay peer entries.
+    ///
+    /// Default: 1 hour.
+    #[serde(with = "serde_helpers::humantime")]
+    pub max_public_entry_tll: Duration,
+
+    /// A period of exchanging public overlay peers.
+    ///
+    /// Default: 3 minutes.
+    #[serde(with = "serde_helpers::humantime")]
+    pub public_overlay_peer_exchange_period: Duration,
+
+    /// A maximum value of a random jitter for the peer exchange period.
+    ///
+    /// Default: 30 seconds.
+    pub public_overlay_peer_exchange_max_jitter: Duration,
+
+    /// Number of peers to send during entries exchange request.
+    ///
+    /// Default: 20.
+    pub exchange_public_entries_batch: usize,
+}
+
+impl Default for OverlayConfig {
+    fn default() -> Self {
+        Self {
+            max_public_entry_tll: Duration::from_secs(3600),
+            public_overlay_peer_exchange_period: Duration::from_secs(3 * 60),
+            public_overlay_peer_exchange_max_jitter: Duration::from_secs(30),
+            exchange_public_entries_batch: 20,
+        }
+    }
+}