diff --git a/pr/524/docs/noq_proto/n0_nat_traversal/index.html b/pr/524/docs/noq_proto/n0_nat_traversal/index.html index 77ed28999..0de012673 100644 --- a/pr/524/docs/noq_proto/n0_nat_traversal/index.html +++ b/pr/524/docs/noq_proto/n0_nat_traversal/index.html @@ -1,2 +1,2 @@ -noq_proto::n0_nat_traversal - Rust

Module n0_nat_traversal

noq_proto

Module n0_nat_traversal 

Source
Expand description

n0’s (https://n0.computer) NAT Traversal protocol implementation.

+noq_proto::n0_nat_traversal - Rust

Module n0_nat_traversal

noq_proto

Module n0_nat_traversal 

Source
Expand description

n0’s (https://n0.computer) NAT Traversal protocol implementation.

Enums§

Error
Errors that the nat traversal state might encounter.
Event
Event emitted when the client receives ADD_ADDRESS or REMOVE_ADDRESS frames.
\ No newline at end of file diff --git a/pr/524/docs/src/noq_proto/connection/mod.rs.html b/pr/524/docs/src/noq_proto/connection/mod.rs.html index 4fc7899f8..b832aecd3 100644 --- a/pr/524/docs/src/noq_proto/connection/mod.rs.html +++ b/pr/524/docs/src/noq_proto/connection/mod.rs.html @@ -2352,11 +2352,11 @@ 2352 } 2353 } 2354 ConnTimer::OffPathProbeRetry => { -2355 // Re-queue off-path probes for retransmission. -2356 if let Ok(server_state) = self.n0_nat_traversal.server_side_mut() -2357 && server_state.queue_retries() -2358 { -2359 trace!("off-path probe retry timer fired, re-queued probes"); +2355 if let Ok(server_state) = self.n0_nat_traversal.server_side_mut() { +2356 let (requeued, _expired_cids) = server_state.queue_retries(); +2357 if requeued { +2358 trace!("off-path probe retry timer fired, re-queued probes"); +2359 } 2360 } 2361 } 2362 }, diff --git a/pr/524/docs/src/noq_proto/n0_nat_traversal.rs.html b/pr/524/docs/src/noq_proto/n0_nat_traversal.rs.html index 9f93ca29b..927252698 100644 --- a/pr/524/docs/src/noq_proto/n0_nat_traversal.rs.html +++ b/pr/524/docs/src/noq_proto/n0_nat_traversal.rs.html @@ -411,286 +411,290 @@ 411 /// Re-queue all sent probes that haven't exceeded [`MAX_OFF_PATH_PROBE_ATTEMPTS`] 412 /// for retransmission. Called when the off-path probe retry timer fires. 413 /// -414 /// Returns `true` if any probes were re-queued. -415 pub(crate) fn queue_retries(&mut self) -> bool { -416 let mut any_requeued = false; -417 self.pending_probes.retain(|_, state| { -418 if state.attempts > 0 && state.attempts < MAX_OFF_PATH_PROBE_ATTEMPTS { -419 state.ready_to_send = true; -420 any_requeued = true; -421 true -422 } else if state.attempts >= MAX_OFF_PATH_PROBE_ATTEMPTS { -423 // Max attempts reached, remove -424 false -425 } else { -426 // Not yet sent, keep as-is -427 true -428 } -429 }); -430 any_requeued -431 } -432 -433 /// Returns whether there are any probes that have been sent but are waiting -434 /// for retry (i.e., sent at least once but under the max attempt limit). -435 pub(crate) fn has_pending_retries(&self) -> bool { -436 self.pending_probes -437 .values() -438 .any(|state| state.attempts > 0 && state.attempts < MAX_OFF_PATH_PROBE_ATTEMPTS) -439 } -440 -441 /// Returns the next probe's address and previous CID without holding a borrow. -442 pub(crate) fn next_probe_info(&self) -> Option<(SocketAddr, Option<ConnectionId>)> { -443 self.pending_probes -444 .iter() -445 .find(|(_, state)| state.ready_to_send) -446 .map(|(addr, state)| ((*addr).into(), state.cid)) -447 } -448 -449 /// Mark a probe as sent by address. -450 pub(crate) fn mark_probe_sent(&mut self, remote: IpPort, cid: ConnectionId) { -451 if let Some(state) = self.pending_probes.get_mut(&remote) { -452 if state.cid.is_none() { -453 state.cid = Some(cid); -454 } -455 state.attempts += 1; -456 state.ready_to_send = false; -457 } -458 } -459} -460 -461pub(crate) struct ServerProbing<'a> { -462 remote: IpPort, -463 pending_probes: &'a mut FxHashMap<IpPort, ProbeState>, -464} -465 -466impl<'a> ServerProbing<'a> { -467 /// Mark this probe as sent with the given CID. -468 pub(crate) fn mark_as_sent(self, cid: ConnectionId) { -469 if let Some(state) = self.pending_probes.get_mut(&self.remote) { -470 if state.cid.is_none() { -471 state.cid = Some(cid); -472 } -473 state.attempts += 1; -474 state.ready_to_send = false; -475 } -476 } -477 -478 /// Returns the CID previously used for this probe address, if any. -479 /// On retries, reuse this CID instead of consuming a new one. -480 pub(crate) fn previous_cid(&self) -> Option<ConnectionId> { -481 self.pending_probes.get(&self.remote).and_then(|s| s.cid) -482 } -483 -484 pub(crate) fn remote(&self) -> SocketAddr { -485 self.remote.into() +414 /// Returns `(any_requeued, expired_cids)` where `expired_cids` are CIDs from +415 /// probes that exceeded max attempts and should be retired. +416 pub(crate) fn queue_retries(&mut self) -> (bool, Vec<ConnectionId>) { +417 let mut any_requeued = false; +418 let mut expired_cids = Vec::new(); +419 self.pending_probes.retain(|_, state| { +420 if state.attempts > 0 && state.attempts < MAX_OFF_PATH_PROBE_ATTEMPTS { +421 state.ready_to_send = true; +422 any_requeued = true; +423 true +424 } else if state.attempts >= MAX_OFF_PATH_PROBE_ATTEMPTS { +425 if let Some(cid) = state.cid { +426 expired_cids.push(cid); +427 } +428 false +429 } else { +430 // Not yet sent, keep as-is +431 true +432 } +433 }); +434 (any_requeued, expired_cids) +435 } +436 +437 /// Returns whether there are any probes that have been sent but are waiting +438 /// for retry (i.e., sent at least once but under the max attempt limit). +439 pub(crate) fn has_pending_retries(&self) -> bool { +440 self.pending_probes +441 .values() +442 .any(|state| state.attempts > 0 && state.attempts < MAX_OFF_PATH_PROBE_ATTEMPTS) +443 } +444 +445 /// Returns the next probe's address and previous CID without holding a borrow. +446 pub(crate) fn next_probe_info(&self) -> Option<(SocketAddr, Option<ConnectionId>)> { +447 self.pending_probes +448 .iter() +449 .find(|(_, state)| state.ready_to_send) +450 .map(|(addr, state)| ((*addr).into(), state.cid)) +451 } +452 +453 /// Mark a probe as sent by address. +454 pub(crate) fn mark_probe_sent(&mut self, remote: IpPort, cid: ConnectionId) { +455 if let Some(state) = self.pending_probes.get_mut(&remote) { +456 if state.cid.is_none() { +457 state.cid = Some(cid); +458 } +459 state.attempts += 1; +460 state.ready_to_send = false; +461 } +462 } +463} +464 +465pub(crate) struct ServerProbing<'a> { +466 remote: IpPort, +467 pending_probes: &'a mut FxHashMap<IpPort, ProbeState>, +468} +469 +470impl<'a> ServerProbing<'a> { +471 /// Mark this probe as sent with the given CID. +472 pub(crate) fn mark_as_sent(self, cid: ConnectionId) { +473 if let Some(state) = self.pending_probes.get_mut(&self.remote) { +474 if state.cid.is_none() { +475 state.cid = Some(cid); +476 } +477 state.attempts += 1; +478 state.ready_to_send = false; +479 } +480 } +481 +482 /// Returns the CID previously used for this probe address, if any. +483 /// On retries, reuse this CID instead of consuming a new one. +484 pub(crate) fn previous_cid(&self) -> Option<ConnectionId> { +485 self.pending_probes.get(&self.remote).and_then(|s| s.cid) 486 } -487} -488 -489impl State { -490 pub(crate) fn new(max_remote_addresses: u8, max_local_addresses: u8, side: Side) -> Self { -491 match side { -492 Side::Client => Self::ClientSide(ClientState::new( -493 max_remote_addresses.into(), -494 max_local_addresses.into(), -495 )), -496 Side::Server => Self::ServerSide(ServerState::new( +487 +488 pub(crate) fn remote(&self) -> SocketAddr { +489 self.remote.into() +490 } +491} +492 +493impl State { +494 pub(crate) fn new(max_remote_addresses: u8, max_local_addresses: u8, side: Side) -> Self { +495 match side { +496 Side::Client => Self::ClientSide(ClientState::new( 497 max_remote_addresses.into(), 498 max_local_addresses.into(), 499 )), -500 } -501 } -502 -503 pub(crate) fn client_side(&self) -> Result<&ClientState, Error> { -504 match self { -505 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), -506 Self::ClientSide(client_side) => Ok(client_side), -507 Self::ServerSide(_) => Err(Error::WrongConnectionSide), -508 } -509 } -510 -511 pub(crate) fn client_side_mut(&mut self) -> Result<&mut ClientState, Error> { -512 match self { -513 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), -514 Self::ClientSide(client_side) => Ok(client_side), -515 Self::ServerSide(_) => Err(Error::WrongConnectionSide), -516 } -517 } -518 -519 pub(crate) fn server_side_mut(&mut self) -> Result<&mut ServerState, Error> { -520 match self { -521 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), -522 Self::ClientSide(_) => Err(Error::WrongConnectionSide), -523 Self::ServerSide(server_side) => Ok(server_side), -524 } -525 } -526 -527 /// Adds a local address to use for nat traversal. -528 /// -529 /// When this endpoint is the server within the connection, these addresses will be sent to the -530 /// client in add address frames. For clients, these addresses will be sent in reach out frames -531 /// when nat traversal attempts are initiated. +500 Side::Server => Self::ServerSide(ServerState::new( +501 max_remote_addresses.into(), +502 max_local_addresses.into(), +503 )), +504 } +505 } +506 +507 pub(crate) fn client_side(&self) -> Result<&ClientState, Error> { +508 match self { +509 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), +510 Self::ClientSide(client_side) => Ok(client_side), +511 Self::ServerSide(_) => Err(Error::WrongConnectionSide), +512 } +513 } +514 +515 pub(crate) fn client_side_mut(&mut self) -> Result<&mut ClientState, Error> { +516 match self { +517 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), +518 Self::ClientSide(client_side) => Ok(client_side), +519 Self::ServerSide(_) => Err(Error::WrongConnectionSide), +520 } +521 } +522 +523 pub(crate) fn server_side_mut(&mut self) -> Result<&mut ServerState, Error> { +524 match self { +525 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), +526 Self::ClientSide(_) => Err(Error::WrongConnectionSide), +527 Self::ServerSide(server_side) => Ok(server_side), +528 } +529 } +530 +531 /// Adds a local address to use for nat traversal. 532 /// -533 /// If a frame should be sent, it is returned. -534 pub(crate) fn add_local_address( -535 &mut self, -536 address: SocketAddr, -537 ) -> Result<Option<AddAddress>, Error> { -538 let ip_port = IpPort::from((address.ip(), address.port())); -539 match self { -540 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), -541 Self::ClientSide(client_state) => { -542 client_state.add_local_address(ip_port)?; -543 Ok(None) -544 } -545 Self::ServerSide(server_state) => server_state.add_local_address(ip_port), -546 } -547 } -548 -549 /// Removes a local address from the advertised set for nat traversal. -550 /// -551 /// When this endpoint is the server, removed addresses must be reported with remove address -552 /// frames. Clients will simply stop reporting these addresses in reach out frames. -553 /// -554 /// If a frame should be sent, it is returned. -555 pub(crate) fn remove_local_address( -556 &mut self, -557 address: SocketAddr, -558 ) -> Result<Option<RemoveAddress>, Error> { -559 let address = IpPort::from((address.ip(), address.port())); -560 match self { -561 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), -562 Self::ClientSide(client_state) => { -563 client_state.remove_local_address(&address); -564 Ok(None) -565 } -566 Self::ServerSide(server_state) => Ok(server_state.remove_local_address(&address)), -567 } -568 } -569 -570 pub(crate) fn get_local_nat_traversal_addresses(&self) -> Result<Vec<SocketAddr>, Error> { -571 match self { -572 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), -573 Self::ClientSide(client_state) => Ok(client_state -574 .local_addresses -575 .iter() -576 .copied() -577 .map(Into::into) -578 .collect()), -579 Self::ServerSide(server_state) => Ok(server_state -580 .local_addresses -581 .keys() -582 .copied() -583 .map(Into::into) -584 .collect()), -585 } -586 } -587} -588 -589/// Returns the given address as canonicalized IP address. -590/// -591/// This checks that the address family is supported by our local socket. -592/// If it is supported, then the address is mapped to the respective IP address. -593/// If the given address is an IPv6 address, but our local socket doesn't support -594/// IPv6, then this returns `None`. -595pub(crate) fn map_to_local_socket_family(address: IpAddr, ipv6: bool) -> Option<IpAddr> { -596 let ip = match address { -597 IpAddr::V4(addr) if ipv6 => IpAddr::V6(addr.to_ipv6_mapped()), -598 IpAddr::V4(_) => address, -599 IpAddr::V6(_) if ipv6 => address, -600 IpAddr::V6(addr) => IpAddr::V4(addr.to_ipv4_mapped()?), -601 }; -602 Some(ip) -603} -604 -605#[cfg(test)] -606mod tests { -607 use super::*; +533 /// When this endpoint is the server within the connection, these addresses will be sent to the +534 /// client in add address frames. For clients, these addresses will be sent in reach out frames +535 /// when nat traversal attempts are initiated. +536 /// +537 /// If a frame should be sent, it is returned. +538 pub(crate) fn add_local_address( +539 &mut self, +540 address: SocketAddr, +541 ) -> Result<Option<AddAddress>, Error> { +542 let ip_port = IpPort::from((address.ip(), address.port())); +543 match self { +544 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), +545 Self::ClientSide(client_state) => { +546 client_state.add_local_address(ip_port)?; +547 Ok(None) +548 } +549 Self::ServerSide(server_state) => server_state.add_local_address(ip_port), +550 } +551 } +552 +553 /// Removes a local address from the advertised set for nat traversal. +554 /// +555 /// When this endpoint is the server, removed addresses must be reported with remove address +556 /// frames. Clients will simply stop reporting these addresses in reach out frames. +557 /// +558 /// If a frame should be sent, it is returned. +559 pub(crate) fn remove_local_address( +560 &mut self, +561 address: SocketAddr, +562 ) -> Result<Option<RemoveAddress>, Error> { +563 let address = IpPort::from((address.ip(), address.port())); +564 match self { +565 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), +566 Self::ClientSide(client_state) => { +567 client_state.remove_local_address(&address); +568 Ok(None) +569 } +570 Self::ServerSide(server_state) => Ok(server_state.remove_local_address(&address)), +571 } +572 } +573 +574 pub(crate) fn get_local_nat_traversal_addresses(&self) -> Result<Vec<SocketAddr>, Error> { +575 match self { +576 Self::NotNegotiated => Err(Error::ExtensionNotNegotiated), +577 Self::ClientSide(client_state) => Ok(client_state +578 .local_addresses +579 .iter() +580 .copied() +581 .map(Into::into) +582 .collect()), +583 Self::ServerSide(server_state) => Ok(server_state +584 .local_addresses +585 .keys() +586 .copied() +587 .map(Into::into) +588 .collect()), +589 } +590 } +591} +592 +593/// Returns the given address as canonicalized IP address. +594/// +595/// This checks that the address family is supported by our local socket. +596/// If it is supported, then the address is mapped to the respective IP address. +597/// If the given address is an IPv6 address, but our local socket doesn't support +598/// IPv6, then this returns `None`. +599pub(crate) fn map_to_local_socket_family(address: IpAddr, ipv6: bool) -> Option<IpAddr> { +600 let ip = match address { +601 IpAddr::V4(addr) if ipv6 => IpAddr::V6(addr.to_ipv6_mapped()), +602 IpAddr::V4(_) => address, +603 IpAddr::V6(_) if ipv6 => address, +604 IpAddr::V6(addr) => IpAddr::V4(addr.to_ipv4_mapped()?), +605 }; +606 Some(ip) +607} 608 -609 #[test] -610 fn test_basic_server_state() { -611 let mut state = ServerState::new(2, 2); +609#[cfg(test)] +610mod tests { +611 use super::*; 612 -613 state -614 .handle_reach_out( -615 ReachOut { -616 round: 1u32.into(), -617 ip: std::net::Ipv4Addr::LOCALHOST.into(), -618 port: 1, -619 }, -620 true, -621 ) -622 .unwrap(); -623 -624 state -625 .handle_reach_out( -626 ReachOut { -627 round: 1u32.into(), -628 ip: "1.1.1.1".parse().unwrap(), //std::net::Ipv4Addr::LOCALHOST.into(), -629 port: 2, -630 }, -631 true, -632 ) -633 .unwrap(); -634 -635 dbg!(&state); -636 assert_eq!(state.pending_probes.len(), 2); -637 -638 let dummy_cid = ConnectionId::new(&[1, 2, 3, 4]); -639 let probe = state.next_probe().unwrap(); -640 probe.mark_as_sent(dummy_cid); -641 let probe = state.next_probe().unwrap(); -642 probe.mark_as_sent(dummy_cid); -643 -644 // After sending both probes, next_probe() returns None (no ready probes) -645 // but the probes are still tracked for potential retry. -646 assert!(state.next_probe().is_none()); -647 assert_eq!(state.pending_probes.len(), 2); -648 assert!(state.has_pending_retries()); -649 -650 // After queuing retries, probes become available again -651 assert!(state.queue_retries()); -652 state.next_probe().unwrap().mark_as_sent(dummy_cid); -653 state.next_probe().unwrap().mark_as_sent(dummy_cid); -654 -655 // After 2 attempts each, retries still available (max is 10) -656 assert!(state.queue_retries()); +613 #[test] +614 fn test_basic_server_state() { +615 let mut state = ServerState::new(2, 2); +616 +617 state +618 .handle_reach_out( +619 ReachOut { +620 round: 1u32.into(), +621 ip: std::net::Ipv4Addr::LOCALHOST.into(), +622 port: 1, +623 }, +624 true, +625 ) +626 .unwrap(); +627 +628 state +629 .handle_reach_out( +630 ReachOut { +631 round: 1u32.into(), +632 ip: "1.1.1.1".parse().unwrap(), //std::net::Ipv4Addr::LOCALHOST.into(), +633 port: 2, +634 }, +635 true, +636 ) +637 .unwrap(); +638 +639 dbg!(&state); +640 assert_eq!(state.pending_probes.len(), 2); +641 +642 let dummy_cid = ConnectionId::new(&[1, 2, 3, 4]); +643 let probe = state.next_probe().unwrap(); +644 probe.mark_as_sent(dummy_cid); +645 let probe = state.next_probe().unwrap(); +646 probe.mark_as_sent(dummy_cid); +647 +648 // After sending both probes, next_probe() returns None (no ready probes) +649 // but the probes are still tracked for potential retry. +650 assert!(state.next_probe().is_none()); +651 assert_eq!(state.pending_probes.len(), 2); +652 assert!(state.has_pending_retries()); +653 +654 // After queuing retries, probes become available again +655 assert!(state.queue_retries().0); +656 state.next_probe().unwrap().mark_as_sent(dummy_cid); 657 state.next_probe().unwrap().mark_as_sent(dummy_cid); -658 state.next_probe().unwrap().mark_as_sent(dummy_cid); -659 -660 // Exhaust remaining attempts -661 for _ in 3..MAX_OFF_PATH_PROBE_ATTEMPTS { -662 assert!(state.queue_retries()); -663 state.next_probe().unwrap().mark_as_sent(dummy_cid); -664 state.next_probe().unwrap().mark_as_sent(dummy_cid); -665 } -666 -667 // After max attempts, probes are removed -668 assert!(!state.queue_retries()); -669 assert!(state.next_probe().is_none()); -670 assert_eq!(state.pending_probes.len(), 0); -671 } -672 -673 #[test] -674 fn test_map_to_local_socket() { -675 assert_eq!( -676 map_to_local_socket_family("1.1.1.1".parse().unwrap(), false), -677 Some("1.1.1.1".parse().unwrap()) -678 ); +658 +659 // After 2 attempts each, retries still available (max is 10) +660 assert!(state.queue_retries().0); +661 state.next_probe().unwrap().mark_as_sent(dummy_cid); +662 state.next_probe().unwrap().mark_as_sent(dummy_cid); +663 +664 // Exhaust remaining attempts +665 for _ in 3..MAX_OFF_PATH_PROBE_ATTEMPTS { +666 assert!(state.queue_retries().0); +667 state.next_probe().unwrap().mark_as_sent(dummy_cid); +668 state.next_probe().unwrap().mark_as_sent(dummy_cid); +669 } +670 +671 // After max attempts, probes are removed +672 assert!(!state.queue_retries().0); +673 assert!(state.next_probe().is_none()); +674 assert_eq!(state.pending_probes.len(), 0); +675 } +676 +677 #[test] +678 fn test_map_to_local_socket() { 679 assert_eq!( -680 map_to_local_socket_family("1.1.1.1".parse().unwrap(), true), -681 Some("::ffff:1.1.1.1".parse().unwrap()) +680 map_to_local_socket_family("1.1.1.1".parse().unwrap(), false), +681 Some("1.1.1.1".parse().unwrap()) 682 ); 683 assert_eq!( -684 map_to_local_socket_family("::1".parse().unwrap(), true), -685 Some("::1".parse().unwrap()) +684 map_to_local_socket_family("1.1.1.1".parse().unwrap(), true), +685 Some("::ffff:1.1.1.1".parse().unwrap()) 686 ); 687 assert_eq!( -688 map_to_local_socket_family("::1".parse().unwrap(), false), -689 None -690 ); +688 map_to_local_socket_family("::1".parse().unwrap(), true), +689 Some("::1".parse().unwrap()) +690 ); 691 assert_eq!( -692 map_to_local_socket_family("::ffff:1.1.1.1".parse().unwrap(), false), -693 Some("1.1.1.1".parse().unwrap()) -694 ) -695 } -696}
\ No newline at end of file +692 map_to_local_socket_family("::1".parse().unwrap(), false), +693 None +694 ); +695 assert_eq!( +696 map_to_local_socket_family("::ffff:1.1.1.1".parse().unwrap(), false), +697 Some("1.1.1.1".parse().unwrap()) +698 ) +699 } +700} \ No newline at end of file