fix: fix keepalive not working

Author: ICKelin

examples/read_async.rs

@@ -0,0 +1,55 @@
+//            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
+//                    Version 2, December 2004
+//
+// Copyleft (ↄ) meh. <meh@schizofreni.co> | http://meh.schizofreni.co
+//
+// Everyone is permitted to copy and distribute verbatim or modified
+// copies of this license document, and changing it is allowed as long
+// as the name is changed.
+//
+//            DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
+//   TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
+//
+//  0. You just DO WHAT THE FUCK YOU WANT TO.
+
+use std::thread::{sleep, Builder};
+use std::time::Duration;
+use tokio::io::AsyncReadExt;
+use tokio_util::sync::CancellationToken;
+use tun::{AbstractDevice, BoxError};
+
+#[tokio::main]
+async fn main() -> Result<(), BoxError> {
+    main_entry().await?;
+    loop {
+        tokio::time::sleep(Duration::from_secs(10)).await;
+    }
+}
+
+async fn main_entry() -> Result<(), BoxError> {
+    let mut config = tun::Configuration::default();
+
+    config
+        .address((10, 0, 0, 9))
+        .netmask((255, 255, 255, 0))
+        .destination((10, 0, 0, 1))
+        .mtu(tun::DEFAULT_MTU)
+        .up();
+
+    #[cfg(target_os = "linux")]
+    config.platform_config(|config| {
+        config.ensure_root_privileges(true);
+    });
+
+    let mut dev = tun::create_as_async(&config)?;
+    let size = dev.mtu()? as usize + tun::PACKET_INFORMATION_LENGTH;
+    let mut buf = vec![0; size];
+    loop {
+        tokio::select! {
+            len = dev.read(&mut buf) => {
+                println!("pkt: {:?}", &buf[..len?]);
+            }
+        };
+    }
+    Ok(())
+}

src/client/mod.rs

@@ -34,7 +34,7 @@ pub struct Args {
     pub keepalive_interval: u64,
 
     /// Keep-alive threshold (reconnect after this many failures)
-    #[arg(long, default_value = "5")]
+    #[arg(long, default_value = "3")]
     pub keepalive_threshold: u8,
 
     /// Enable P2P direct connection (disabled by default, uses relay only)

src/client/relay.rs

@@ -59,9 +59,18 @@ impl RelayClient {
         let port = self.cfg.port;
         let stun_ip = self.cfg.stun_ip.clone();
         let stun_port = self.cfg.stun_port;
+
+        let mut last_active = Instant::now();
+        let timeout_secs = (self.cfg.keep_alive_thresh - 1) as u64 * self.cfg.keepalive_interval.as_secs();
         loop {
             tokio::select! {
                 _ = keepalive_ticker.tick() => {
+                    if last_active.elapsed().as_secs() > timeout_secs {
+                        tracing::warn!("keepalive threshold {:?} exceeded", last_active.elapsed());
+                        break;
+                    }
+
+                    tracing::debug!("sending keepalive frame");
                     let keepalive_frame = Frame::KeepAlive(KeepAliveFrame {
                         identity: self.cfg.identity.clone(),
                         ipv6: current_ipv6.clone(),
@@ -87,6 +96,7 @@ impl RelayClient {
 
                 // Periodic IPv6 address update check
                 _ = ipv6_update_ticker.tick() => {
+                    tracing::debug!("ipv6 update tick");
                     if let Some(new_ipv6) = utils::get_ipv6() {
                         tracing::info!("IPv6 address updated: {} -> {}", current_ipv6, new_ipv6);
                         current_ipv6 = new_ipv6.clone();
@@ -98,6 +108,7 @@ impl RelayClient {
 
                 // inbound
                 result = conn.read_frame() => {
+                    last_active = Instant::now();
                     match result {
                         Ok(frame) => {
                             tracing::debug!("received frame {}", frame);

src/server/server.rs

@@ -194,6 +194,7 @@ impl Handler {
                         };
                     }
                 }
+                // TODO: check last_active
             }
         }