feat(snownet): timeout connections if we don't receive a candidate within 10s (#3790)

Previously, we had a dedicated timer for this within the tunnel
implementation. Now that we have control over the internals of our
connection via `snownet`, we can timeout the connection if we don't
receive a candidate from the remote within 10s.

rust/Cargo.lock

@@ -5650,6 +5650,7 @@ dependencies = [
  "stun_codec",
  "thiserror",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]

rust/connlib/snownet/Cargo.toml

@@ -20,3 +20,4 @@ bytes = "1.4.0"
 once_cell = "1.17.1"
 backoff = "0.4.0"
 hex = "0.4.0"
+tracing-subscriber = { workspace = true }

rust/connlib/snownet/src/node.rs

@@ -35,6 +35,9 @@ use tracing::{field, Span};
 // Note: Taken from boringtun
 const HANDSHAKE_RATE_LIMIT: u64 = 100;
 
+/// How long we will at most wait for a candidate from the remote.
+const CANDIDATE_TIMEOUT: Duration = Duration::from_secs(10);
+
 const MAX_UDP_SIZE: usize = (1 << 16) - 1;
 
 /// Manages a set of wireguard connections for a server.
@@ -469,16 +472,20 @@ where
         self.last_now = now;
 
         let mut expired_connections = vec![];
+        let mut no_candidates_received = vec![];
 
         for (id, c) in self.connections.iter_established_mut() {
             match c.handle_timeout(now, &mut self.allocations) {
                 Ok(Some(transmit)) => {
                     self.buffered_transmits.push_back(transmit);
                 }
-                Err(WireGuardError::ConnectionExpired) => {
+                Err(ConnectionError::Wireguard(WireGuardError::ConnectionExpired)) => {
                     expired_connections.push(id);
                 }
-                Err(e) => {
+                Err(ConnectionError::CandidateTimeout) => {
+                    no_candidates_received.push(id);
+                }
+                Err(ConnectionError::Wireguard(e)) => {
                     tracing::warn!(%id, ?e);
                 }
                 _ => {}
@@ -492,6 +499,13 @@ where
             self.pending_events.push_back(Event::ConnectionFailed(conn))
         }
 
+        for conn in no_candidates_received {
+            tracing::info!(id = %conn, "Connection failed (no candidates received)");
+
+            self.connections.established.remove(&conn);
+            self.pending_events.push_back(Event::ConnectionFailed(conn))
+        }
+
         for binding in self.bindings.values_mut() {
             binding.handle_timeout(now);
         }
@@ -560,6 +574,7 @@ where
         key: [u8; 32],
         allowed_stun_servers: HashSet<SocketAddr>,
         allowed_turn_servers: HashSet<SocketAddr>,
+        now: Instant,
     ) -> Connection {
         agent.handle_timeout(self.last_now);
 
@@ -584,6 +599,7 @@ where
             peer_socket: None,
             possible_sockets: HashSet::default(),
             stats: Default::default(),
+            signalling_completed_at: now,
         }
     }
 
@@ -846,9 +862,14 @@ where
         params
     }
 
+    /// Whether we have sent an [`Offer`] for this connection and are currently expecting an [`Answer`].
+    pub fn is_expecting_answer(&self, id: TId) -> bool {
+        self.connections.initial.contains_key(&id)
+    }
+
     /// Accept an [`Answer`] from the remote for a connection previously created via [`Node::new_connection`].
     #[tracing::instrument(level = "info", skip_all, fields(%id))]
-    pub fn accept_answer(&mut self, id: TId, remote: PublicKey, answer: Answer) {
+    pub fn accept_answer(&mut self, id: TId, remote: PublicKey, answer: Answer, now: Instant) {
         let Some(initial) = self.connections.initial.remove(&id) else {
             tracing::debug!("No initial connection state, ignoring answer"); // This can happen if the connection setup timed out.
             return;
@@ -873,6 +894,7 @@ where
             *initial.session_key.expose_secret(),
             initial.stun_servers,
             initial.turn_servers,
+            now,
         );
 
         let existing = self.connections.established.insert(id, connection);
@@ -900,6 +922,7 @@ where
         remote: PublicKey,
         allowed_stun_servers: HashSet<SocketAddr>,
         allowed_turn_servers: HashSet<(SocketAddr, String, String, String)>,
+        now: Instant,
     ) -> Answer {
         debug_assert!(
             !self.connections.initial.contains_key(&id),
@@ -945,6 +968,7 @@ where
             *offer.session_key.expose_secret(),
             allowed_stun_servers,
             allowed_turn_servers,
+            now,
         );
         let existing = self.connections.established.insert(id, connection);
 
@@ -1287,6 +1311,8 @@ struct Connection {
     turn_servers: HashSet<SocketAddr>,
 
     stats: ConnectionStats,
+
+    signalling_completed_at: Instant,
 }
 
 /// The socket of the peer we are connected to.
@@ -1354,8 +1380,17 @@ impl Connection {
     fn poll_timeout(&mut self) -> Option<Instant> {
         let agent_timeout = self.agent.poll_timeout();
         let next_wg_timer = Some(self.next_timer_update);
+        let candidate_timeout = self.candidate_timeout();
 
-        earliest(agent_timeout, next_wg_timer)
+        earliest(agent_timeout, earliest(next_wg_timer, candidate_timeout))
+    }
+
+    fn candidate_timeout(&self) -> Option<Instant> {
+        if !self.agent.remote_candidates().is_empty() {
+            return None;
+        }
+
+        Some(self.signalling_completed_at + CANDIDATE_TIMEOUT)
     }
 
     #[must_use]
@@ -1407,9 +1442,16 @@ impl Connection {
         &mut self,
         now: Instant,
         allocations: &mut HashMap<SocketAddr, Allocation>,
-    ) -> Result<Option<Transmit<'static>>, WireGuardError> {
+    ) -> Result<Option<Transmit<'static>>, ConnectionError> {
         self.agent.handle_timeout(now);
 
+        if self
+            .candidate_timeout()
+            .is_some_and(|timeout| now >= timeout)
+        {
+            return Err(ConnectionError::CandidateTimeout);
+        }
+
         // TODO: `boringtun` is impure because it calls `Instant::now`.
 
         if now >= self.next_timer_update {
@@ -1429,7 +1471,7 @@ impl Connection {
 
             match self.tunnel.update_timers(&mut buf) {
                 TunnResult::Done => {}
-                TunnResult::Err(e) => return Err(e),
+                TunnResult::Err(e) => return Err(ConnectionError::Wireguard(e)),
                 TunnResult::WriteToNetwork(b) => {
                     let Some(transmit) = self.encapsulate(b, allocations, now) else {
                         return Ok(None);
@@ -1524,3 +1566,9 @@ impl Connection {
             .find(|c| c.addr() == source)
     }
 }
+
+#[derive(Debug)]
+enum ConnectionError {
+    Wireguard(WireGuardError),
+    CandidateTimeout,
+}

rust/connlib/snownet/tests/lib.rs

@@ -1,5 +1,5 @@
 use boringtun::x25519::StaticSecret;
-use snownet::{ClientNode, Event, ServerNode};
+use snownet::{Answer, ClientNode, Event, ServerNode};
 use std::{
     collections::HashSet,
     iter,
@@ -9,11 +9,10 @@ use std::{
 use str0m::{net::Protocol, Candidate};
 
 #[test]
-fn connection_times_out_after_10_seconds() {
+fn connection_times_out_after_20_seconds() {
     let start = Instant::now();
 
-    let mut alice =
-        ClientNode::<u64>::new(StaticSecret::random_from_rng(rand::thread_rng()), start);
+    let (mut alice, _) = alice_and_bob(start);
 
     let _ = alice.new_connection(1, HashSet::new(), HashSet::new());
     alice.handle_timeout(start + Duration::from_secs(20));
@@ -21,21 +20,56 @@ fn connection_times_out_after_10_seconds() {
     assert_eq!(alice.poll_event().unwrap(), Event::ConnectionFailed(1));
 }
 
+#[test]
+fn connection_without_candidates_times_out_after_10_seconds() {
+    let _ = tracing_subscriber::fmt().with_test_writer().try_init();
+
+    let start = Instant::now();
+
+    let (mut alice, mut bob) = alice_and_bob(start);
+    let answer = send_offer(&mut alice, &mut bob, start);
+
+    let accepted_at = start + Duration::from_secs(1);
+    alice.accept_answer(1, bob.public_key(), answer, accepted_at);
+
+    alice.handle_timeout(accepted_at + Duration::from_secs(10));
+
+    assert_eq!(alice.poll_event().unwrap(), Event::ConnectionFailed(1));
+}
+
+#[test]
+fn connection_with_candidates_does_not_time_out_after_10_seconds() {
+    let _ = tracing_subscriber::fmt().with_test_writer().try_init();
+
+    let start = Instant::now();
+
+    let (mut alice, mut bob) = alice_and_bob(start);
+    let answer = send_offer(&mut alice, &mut bob, start);
+
+    let accepted_at = start + Duration::from_secs(1);
+    alice.accept_answer(1, bob.public_key(), answer, accepted_at);
+    alice.add_local_host_candidate(s("10.0.0.2:4444")).unwrap();
+    alice.add_remote_candidate(1, host("10.0.0.1:4444"));
+
+    alice.handle_timeout(accepted_at + Duration::from_secs(10));
+
+    let any_failed =
+        iter::from_fn(|| alice.poll_event()).any(|e| matches!(e, Event::ConnectionFailed(_)));
+
+    assert!(!any_failed);
+}
+
 #[test]
 fn answer_after_stale_connection_does_not_panic() {
     let start = Instant::now();
 
-    let mut alice =
-        ClientNode::<u64>::new(StaticSecret::random_from_rng(rand::thread_rng()), start);
-    let mut bob = ServerNode::<u64>::new(StaticSecret::random_from_rng(rand::thread_rng()), start);
+    let (mut alice, mut bob) = alice_and_bob(start);
+    let answer = send_offer(&mut alice, &mut bob, start);
 
-    let offer = alice.new_connection(1, HashSet::new(), HashSet::new());
-    let answer =
-        bob.accept_connection(1, offer, alice.public_key(), HashSet::new(), HashSet::new());
+    let now = start + Duration::from_secs(10);
+    alice.handle_timeout(now);
 
-    alice.handle_timeout(start + Duration::from_secs(10));
-
-    alice.accept_answer(1, bob.public_key(), answer);
+    alice.accept_answer(1, bob.public_key(), answer, now + Duration::from_secs(1));
 }
 
 #[test]
@@ -62,10 +96,16 @@ fn only_generate_candidate_event_after_answer() {
         "no event to be emitted before accepting the answer"
     );
 
-    let answer =
-        bob.accept_connection(1, offer, alice.public_key(), HashSet::new(), HashSet::new());
+    let answer = bob.accept_connection(
+        1,
+        offer,
+        alice.public_key(),
+        HashSet::new(),
+        HashSet::new(),
+        Instant::now(),
+    );
 
-    alice.accept_answer(1, bob.public_key(), answer);
+    alice.accept_answer(1, bob.public_key(), answer, Instant::now());
 
     assert!(iter::from_fn(|| alice.poll_event()).any(|ev| ev
         == Event::SignalIceCandidate {
@@ -102,6 +142,26 @@ fn second_connection_with_same_relay_reuses_allocation() {
     assert!(alice.poll_transmit().is_none());
 }
 
+fn alice_and_bob(start: Instant) -> (ClientNode<u64>, ServerNode<u64>) {
+    let alice = ClientNode::<u64>::new(StaticSecret::random_from_rng(rand::thread_rng()), start);
+    let bob = ServerNode::<u64>::new(StaticSecret::random_from_rng(rand::thread_rng()), start);
+
+    (alice, bob)
+}
+
+fn send_offer(alice: &mut ClientNode<u64>, bob: &mut ServerNode<u64>, now: Instant) -> Answer {
+    let offer = alice.new_connection(1, HashSet::new(), HashSet::new());
+
+    bob.accept_connection(
+        1,
+        offer,
+        alice.public_key(),
+        HashSet::new(),
+        HashSet::new(),
+        now,
+    )
+}
+
 fn relay(username: &str, pass: &str, realm: &str) -> (SocketAddr, String, String, String) {
     (
         RELAY,
@@ -111,4 +171,14 @@ fn relay(username: &str, pass: &str, realm: &str) -> (SocketAddr, String, String
     )
 }
 
+fn host(socket: &str) -> String {
+    Candidate::host(s(socket), Protocol::Udp)
+        .unwrap()
+        .to_sdp_string()
+}
+
+fn s(socket: &str) -> SocketAddr {
+    socket.parse().unwrap()
+}
+
 const RELAY: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 10000));

rust/connlib/tunnel/src/client.rs

@@ -11,7 +11,7 @@ use connlib_shared::messages::{
 };
 use connlib_shared::{Callbacks, Dname, IpProvider};
 use domain::base::Rtype;
-use futures_bounded::{FuturesMap, FuturesTupleSet, PushError};
+use futures_bounded::FuturesTupleSet;
 use ip_network::IpNetwork;
 use ip_network_table::IpNetworkTable;
 use itertools::Itertools;
@@ -29,7 +29,6 @@ use tokio::time::{Interval, MissedTickBehavior};
 // Using str here because Ipv4/6Network doesn't support `const` 🙃
 const IPV4_RESOURCES: &str = "100.96.0.0/11";
 const IPV6_RESOURCES: &str = "fd00:2021:1111:8000::/107";
-const MAX_CONNECTION_REQUEST_DELAY: Duration = Duration::from_secs(10);
 
 #[derive(Debug, Clone, Hash, PartialEq, Eq)]
 pub struct DnsResource {
@@ -247,18 +246,6 @@ where
 /// [`Tunnel`] state specific to clients.
 pub struct ClientState {
     awaiting_connection: HashMap<ResourceId, AwaitingConnectionDetails>,
-
-    pub gateway_awaiting_connection: HashSet<GatewayId>,
-    // This timer exist for an unlikely case, on unreliable connections where the RequestConnection message
-    // or the response is lost:
-    // This would remove the "PendingConnection" message and be able to try the connection again.
-    // There are some edge cases that come with this:
-    // * a gateway in a VERY unlikely case could receive the connection request twice. This will stop any connection attempt and make the whole thing start again.
-    // if this would happen often the UX would be awful but this is only in cases where messages are delayed for more than 10 seconds, it's enough that it doesn't break correctness.
-    // * even more unlikely a tunnel could be established in a sort of race condition when this timer goes off. Again a similar behavior to the one above will happen, the webrtc connection will be forcefully terminated from the gateway.
-    // then the old peer will expire, this might take ~180 seconds. This is an even worse experience but the likelihood of this happen is infinitesimaly small, again correctness is the only important part.
-    gateway_awaiting_connection_timers: FuturesMap<GatewayId, ()>,
-
     resources_gateways: HashMap<ResourceId, GatewayId>,
 
     pub dns_resources_internal_ips: HashMap<DnsResource, HashSet<IpAddr>>,
@@ -396,31 +383,9 @@ impl ClientState {
             .get_mut(&resource)
             .ok_or(Error::UnexpectedConnectionDetails)?;
 
-        if self.gateway_awaiting_connection.contains(&gateway) {
-            self.awaiting_connection.remove(&resource);
-            return Err(Error::PendingConnection);
-        }
-
         self.resources_gateways.insert(resource, gateway);
 
         if self.peers.get(&gateway).is_none() {
-            match self
-                .gateway_awaiting_connection_timers
-                // Note: we don't need to set a timer here because
-                // the FutureMap already expires things, it seems redundant
-                // to also have timer that expires.
-                .try_push(gateway, std::future::pending())
-            {
-                Ok(_) => {}
-                Err(PushError::BeyondCapacity(_)) => {
-                    tracing::warn!(%gateway, "Too many concurrent connection attempts");
-                    return Err(Error::TooManyConnectionRequests);
-                }
-                Err(PushError::Replaced(_)) => {
-                    // The timers are equivalent for our purpose so we don't really care about this one.
-                }
-            };
-            self.gateway_awaiting_connection.insert(gateway);
             return Ok(None);
         };
 
@@ -438,13 +403,7 @@ impl ClientState {
 
     pub fn on_connection_failed(&mut self, resource: ResourceId) {
         self.awaiting_connection.remove(&resource);
-
-        let Some(gateway) = self.resources_gateways.remove(&resource) else {
-            return;
-        };
-
-        self.gateway_awaiting_connection.remove(&gateway);
-        self.gateway_awaiting_connection_timers.remove(gateway);
+        self.resources_gateways.remove(&resource);
     }
 
     #[tracing::instrument(level = "debug", skip_all, fields(resource_address = %resource.address, resource_id = %resource.id))]
@@ -483,9 +442,8 @@ impl ClientState {
         debug_assert!(self.resource_ids.contains_key(&resource));
 
         let gateways = self
-            .gateway_awaiting_connection
-            .iter()
-            .chain(self.resources_gateways.values())
+            .resources_gateways
+            .values()
             .copied()
             .collect::<HashSet<_>>();
 
@@ -521,7 +479,6 @@ impl ClientState {
     pub fn create_peer_config_for_new_connection(
         &mut self,
         resource: ResourceId,
-        gateway: GatewayId,
         domain: &Option<Dname>,
     ) -> Result<Vec<IpNetwork>, ConnlibError> {
         let desc = self
@@ -532,8 +489,6 @@ impl ClientState {
         let ips = self.get_resource_ip(desc, domain);
 
         // Tidy up state once everything succeeded.
-        self.gateway_awaiting_connection.remove(&gateway);
-        self.gateway_awaiting_connection_timers.remove(gateway);
         self.awaiting_connection.remove(&resource);
 
         Ok(ips)
@@ -633,12 +588,6 @@ impl ClientState {
                 return Poll::Ready(event);
             }
 
-            if let Poll::Ready((gateway_id, _)) =
-                self.gateway_awaiting_connection_timers.poll_unpin(cx)
-            {
-                self.gateway_awaiting_connection.remove(&gateway_id);
-            }
-
             if self.refresh_dns_timer.poll_tick(cx).is_ready() {
                 let mut connections = Vec::new();
 
@@ -713,10 +662,6 @@ impl Default for ClientState {
 
         Self {
             awaiting_connection: Default::default(),
-
-            gateway_awaiting_connection: Default::default(),
-            gateway_awaiting_connection_timers: FuturesMap::new(MAX_CONNECTION_REQUEST_DELAY, 100),
-
             resources_gateways: Default::default(),
             forwarded_dns_queries: FuturesTupleSet::new(
                 Duration::from_secs(60),

rust/connlib/tunnel/src/control_protocol/client.rs

@@ -1,4 +1,4 @@
-use std::{collections::HashSet, net::IpAddr};
+use std::{collections::HashSet, net::IpAddr, time::Instant};
 
 use boringtun::x25519::PublicKey;
 use connlib_shared::{
@@ -64,6 +64,10 @@ where
             return Ok(Request::ReuseConnection(connection));
         }
 
+        if self.connections_state.node.is_expecting_answer(gateway_id) {
+            return Err(Error::PendingConnection);
+        }
+
         let domain = self
             .role_state
             .get_awaiting_connection_domain(&resource_id)?
@@ -101,7 +105,6 @@ where
     ) -> Result<()> {
         let ips = self.role_state.create_peer_config_for_new_connection(
             resource_id,
-            gateway_id,
             &domain_response.as_ref().map(|d| d.domain.clone()),
         )?;
 
@@ -151,6 +154,7 @@ where
                     password: rtc_ice_params.password,
                 },
             },
+            Instant::now(),
         );
 
         self.new_peer(resource_id, gateway_id, domain_response)?;

rust/connlib/tunnel/src/control_protocol/gateway.rs

@@ -0,0 +1,183 @@
+use crate::{
+    dns::is_subdomain,
+    peer::{PacketTransformGateway, Peer},
+    Error, GatewayState, Tunnel,
+};
+
+use super::{stun, turn};
+use boringtun::x25519::PublicKey;
+use chrono::{DateTime, Utc};
+use connlib_shared::{
+    messages::{
+        Answer, ClientId, ConnectionAccepted, DomainResponse, Key, Offer, Relay, ResourceId,
+    },
+    Callbacks, Dname, Result,
+};
+use ip_network::IpNetwork;
+use secrecy::{ExposeSecret as _, Secret};
+use snownet::{Credentials, Server};
+use std::time::Instant;
+
+/// Description of a resource that maps to a DNS record which had its domain already resolved.
+#[derive(Debug, Clone, PartialEq, Eq, Hash)]
+pub struct ResolvedResourceDescriptionDns {
+    pub id: ResourceId,
+    /// Internal resource's domain name.
+    pub domain: String,
+    /// Name of the resource.
+    ///
+    /// Used only for display.
+    pub name: String,
+
+    pub addresses: Vec<IpNetwork>,
+}
+
+pub type ResourceDescription =
+    connlib_shared::messages::ResourceDescription<ResolvedResourceDescriptionDns>;
+
+impl<CB> Tunnel<CB, GatewayState, Server, ClientId>
+where
+    CB: Callbacks + 'static,
+{
+    /// Accept a connection request from a client.
+    ///
+    /// Sets a connection to a remote SDP, creates the local SDP
+    /// and returns it.
+    ///
+    /// # Returns
+    /// The connection details
+    #[allow(clippy::too_many_arguments)]
+    pub fn set_peer_connection_request(
+        &mut self,
+        client: ClientId,
+        key: Secret<Key>,
+        offer: Offer,
+        gateway: PublicKey,
+        ips: Vec<IpNetwork>,
+        relays: Vec<Relay>,
+        domain: Option<Dname>,
+        expires_at: Option<DateTime<Utc>>,
+        resource: ResourceDescription,
+    ) -> Result<ConnectionAccepted> {
+        let resource_addresses = match &resource {
+            ResourceDescription::Dns(r) => {
+                let Some(domain) = domain.clone() else {
+                    return Err(Error::ControlProtocolError);
+                };
+
+                if !is_subdomain(&domain, &r.domain) {
+                    return Err(Error::InvalidResource);
+                }
+
+                r.addresses.clone()
+            }
+            ResourceDescription::Cidr(ref cidr) => vec![cidr.address],
+        };
+
+        let answer = self.connections_state.node.accept_connection(
+            client,
+            snownet::Offer {
+                session_key: key.expose_secret().0.into(),
+                credentials: Credentials {
+                    username: offer.username,
+                    password: offer.password,
+                },
+            },
+            gateway,
+            stun(&relays, |addr| {
+                self.connections_state.sockets.can_handle(addr)
+            }),
+            turn(&relays, |addr| {
+                self.connections_state.sockets.can_handle(addr)
+            }),
+            Instant::now(),
+        );
+
+        self.new_peer(
+            ips,
+            client,
+            resource,
+            expires_at,
+            resource_addresses.clone(),
+        )?;
+
+        tracing::info!(%client, gateway = %hex::encode(gateway.as_bytes()), "Connection is ready");
+
+        Ok(ConnectionAccepted {
+            ice_parameters: Answer {
+                username: answer.credentials.username,
+                password: answer.credentials.password,
+            },
+            domain_response: domain.map(|domain| DomainResponse {
+                domain,
+                address: resource_addresses
+                    .into_iter()
+                    .map(|ip| ip.network_address())
+                    .collect(),
+            }),
+        })
+    }
+
+    pub fn allow_access(
+        &mut self,
+        resource: ResourceDescription,
+        client: ClientId,
+        expires_at: Option<DateTime<Utc>>,
+        domain: Option<Dname>,
+    ) -> Option<DomainResponse> {
+        let peer = self.role_state.peers.get_mut(&client)?;
+
+        let (addresses, resource_id) = match &resource {
+            ResourceDescription::Dns(r) => {
+                let Some(domain) = domain.clone() else {
+                    return None;
+                };
+
+                if !is_subdomain(&domain, &r.domain) {
+                    return None;
+                }
+
+                (r.addresses.clone(), r.id)
+            }
+            ResourceDescription::Cidr(cidr) => (vec![cidr.address], cidr.id),
+        };
+
+        for address in &addresses {
+            peer.transform
+                .add_resource(*address, resource.clone(), expires_at);
+        }
+
+        tracing::info!(%client, resource = %resource_id, expires = ?expires_at.map(|e| e.to_rfc3339()), "Allowing access to resource");
+
+        if let Some(domain) = domain {
+            return Some(DomainResponse {
+                domain,
+                address: addresses.iter().map(|i| i.network_address()).collect(),
+            });
+        }
+
+        None
+    }
+
+    fn new_peer(
+        &mut self,
+        ips: Vec<IpNetwork>,
+        client_id: ClientId,
+        resource: ResourceDescription,
+        expires_at: Option<DateTime<Utc>>,
+        resource_addresses: Vec<IpNetwork>,
+    ) -> Result<()> {
+        tracing::trace!(?ips, "new_data_channel_open");
+
+        let mut peer = Peer::new(client_id, PacketTransformGateway::default(), &ips, ());
+
+        for address in resource_addresses {
+            peer.transform
+                .add_resource(address, resource.clone(), expires_at);
+        }
+
+        self.role_state.peers.insert(peer, &ips);
+
+        Ok(())
+    }
+}

rust/connlib/tunnel/src/gateway.rs

@@ -15,7 +15,7 @@ use ip_network::IpNetwork;
 use secrecy::{ExposeSecret as _, Secret};
 use snownet::Server;
 use std::task::{ready, Context, Poll};
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tokio::time::{interval, Interval, MissedTickBehavior};
 
 const PEERS_IPV4: &str = "100.64.0.0/11";
@@ -113,6 +113,7 @@ where
             turn(&relays, |addr| {
                 self.connections_state.sockets.can_handle(addr)
             }),
+            Instant::now(),
         );
 
         self.new_peer(

rust/connlib/tunnel/src/lib.rs

@@ -335,6 +335,17 @@ where
     }
 
     fn poll_next_event(&mut self, cx: &mut Context<'_>) -> Poll<Event<TId>> {
+        if let Poll::Ready(prev_timeout) = self.connection_pool_timeout.poll_unpin(cx) {
+            self.node.handle_timeout(prev_timeout);
+            if let Some(new_timeout) = self.node.poll_timeout() {
+                debug_assert_ne!(prev_timeout, new_timeout, "Timer busy loop!");
+
+                self.connection_pool_timeout = sleep_until(new_timeout).boxed();
+            }
+
+            cx.waker().wake_by_ref();
+        }
+
         if self.stats_timer.poll_tick(cx).is_ready() {
             let (node_stats, conn_stats) = self.node.stats();
 
@@ -373,15 +384,6 @@ where
             _ => {}
         }
 
-        if let Poll::Ready(instant) = self.connection_pool_timeout.poll_unpin(cx) {
-            self.node.handle_timeout(instant);
-            if let Some(timeout) = self.node.poll_timeout() {
-                self.connection_pool_timeout = sleep_until(timeout).boxed();
-            }
-
-            cx.waker().wake_by_ref();
-        }
-
         Poll::Pending
     }
 }

rust/snownet-tests/src/main.rs

@@ -112,6 +112,7 @@ async fn main() -> Result<()> {
                         password: answer.password,
                     },
                 },
+                Instant::now(),
             );
 
             let rx = spawn_candidate_task(redis_connection.clone(), "listener_candidates");
@@ -178,6 +179,7 @@ async fn main() -> Result<()> {
                 offer.public_key.into(),
                 stun_server.into_iter().collect(),
                 turn_server.into_iter().collect(),
+                Instant::now(),
             );
 
             redis_connection

scripts/nix/flake.nix

@@ -51,7 +51,7 @@
           };
 
           devShell = pkgs.mkShell {
-            packages = [ pkgs.cargo-tauri ];
+            packages = [ pkgs.cargo-tauri pkgs.iptables ];
             buildInputs = [
               (pkgs.rust-bin.fromRustupToolchainFile ../../rust/rust-toolchain.toml)
             ] ++ packages;