node: add reconcile

logs

reconcile

reconcile

easytier/src/dns/config/mod.rs

@@ -22,6 +22,7 @@ pub static DNS_SERVER_RPC_ADDR: LazyLock<Url> =
 pub const DNS_NODE_TTI: Duration = Duration::from_secs(5);
 
 pub const DNS_NODE_RR_INTERVAL: Duration = Duration::from_secs(4);
+pub const DNS_NODE_RECONCILE_INTERVAL: Duration = Duration::from_secs(10);
 pub const DNS_SERVER_ELECTION_INTERVAL: Duration = Duration::from_secs(5);
 pub const DNS_PEER_TTI: Duration = Duration::from_secs(3);
 pub const DNS_PEER_REFRESH_ATTEMPTS: usize = 3;

easytier/src/dns/node.rs

@@ -1,7 +1,7 @@
 use crate::common::global_ctx::{ArcGlobalCtx, GlobalCtxEvent};
 use crate::dns::config::{
-    DNS_NODE_RR_INTERVAL, DNS_PEER_REFRESH_ATTEMPTS, DNS_PEER_REFRESH_BACKOFF,
-    DNS_SERVER_ELECTION_INTERVAL, DNS_SERVER_RPC_ADDR,
+    DNS_NODE_RECONCILE_INTERVAL, DNS_NODE_RR_INTERVAL, DNS_PEER_REFRESH_ATTEMPTS,
+    DNS_PEER_REFRESH_BACKOFF, DNS_SERVER_ELECTION_INTERVAL, DNS_SERVER_RPC_ADDR,
 };
 use crate::dns::peer_mgr::DnsPeerMgr;
 use crate::dns::server::DnsServer;
@@ -17,7 +17,7 @@ use std::io;
 use std::sync::Arc;
 use tokio::sync::{broadcast, Notify};
 use tokio::task::JoinSet;
-use tokio::time::{sleep, sleep_until, Instant};
+use tokio::time::{interval, sleep, sleep_until, Instant, MissedTickBehavior};
 use tokio_util::sync::CancellationToken;
 use tracing::instrument;
 use uuid::Uuid;
@@ -93,6 +93,9 @@ impl DnsNodeRuntime {
         let timer = sleep_until(last_heartbeat);
         tokio::pin!(timer);
 
+        let mut reconcile_interval = interval(DNS_NODE_RECONCILE_INTERVAL);
+        reconcile_interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
         let mut subscriber = self.global_ctx.subscribe();
         let mut tasks = JoinSet::new();
 
@@ -123,6 +126,13 @@ impl DnsNodeRuntime {
                     last_heartbeat = Instant::now();
                 }
 
+                _ = reconcile_interval.tick() => {
+                    let mgr = self.mgr.clone();
+                    tasks.spawn(async move {
+                        mgr.reconcile().await;
+                    });
+                }
+
                 _ = self.mgr.dirty.wait() => {}
 
                 event = subscriber.recv() => {

easytier/src/dns/peer_mgr.rs

@@ -75,6 +75,7 @@ impl DnsPeerMgrInner {
         }
     }
 
+    #[instrument(skip(self), level = "trace", ret)]
     pub async fn refresh(
         &self,
         peer_id: PeerId,
@@ -85,7 +86,10 @@ impl DnsPeerMgrInner {
             attempts = attempts.saturating_sub(1);
             let result = self.try_refresh(peer_id).await;
             match &result {
-                Ok(_) => return result,
+                Ok(_) => {
+                    tracing::trace!(?peer_id, "peer info refreshed");
+                    return result;
+                }
                 Err(_) if attempts == 0 => return result,
                 Err(error) => {
                     tracing::error!(
@@ -203,6 +207,26 @@ impl DnsPeerMgr {
                 &self.global_ctx.get_network_name(),
             )
     }
+
+    #[instrument(skip(self), level = "trace")]
+    pub async fn reconcile(&self) {
+        stream::iter(self.peer_mgr.list_routes().await.into_iter())
+            .map(|route| {
+                let peer_id = route.peer_id;
+                let this = self.clone();
+                async move {
+                    if let Err(error) = this
+                        .refresh(peer_id, DNS_PEER_REFRESH_ATTEMPTS, DNS_PEER_REFRESH_BACKOFF)
+                        .await
+                    {
+                        tracing::error!(?error, ?peer_id, "failed to refresh peer info");
+                    }
+                }
+            })
+            .buffer_unordered(32)
+            .collect::<Vec<_>>()
+            .await;
+    }
 }
 
 impl Deref for DnsPeerMgr {