ab/tsunagi
1
0 Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity
Files
8a6216bccb392ee9e9d625bc23a7eaf541506c81
tsunagi/vendor/iroh/src/socket.rs
T
Aleksandr Bogomiakov 2b0c938487 Bound iroh path retries to prevent multi-connection OOM 
A long-running rc.33 daemon reached OOM with roughly 22 GiB resident and
24 GiB swapped. Live inspection found 40 GiB and 10/20 GiB anonymous
container mappings; the active worker stack and tracing metadata identified
iroh's scheduled path-open retry loop. In iroh 1.2.0 every failed connection
requeues the same path, while each retry visits every connection again.
Separate control and data connections multiply the queue geometrically.

Vendor the published iroh 1.2.0 source and select it with a workspace Cargo
patch. Deduplicate retry admission and cap distinct pending paths at 64,
evicting the oldest before inserting a new candidate. Keep the current and
next retry batches bounded without combining control/data connections or
changing wire protocols, identity derivation, authentication or state.
Document source provenance, the live evidence and removal conditions.
Upstream PR 4551 merged on October 6, but inspection of the official 1.3.0
crate still shows the unconditional enqueue, so that upgrade alone is not
a fix. No version bump, release tag or installed-service change.

The default workspace suite includes the exact vendored admission module.
Three regressions failed with the original enqueue and pass with the fix:
repeated failures across connections, duplicate admission at capacity, and
10,000 distinct candidates with bounded length/capacity and FIFO eviction.
Transport integration tests continue using real endpoints and encryption.

Checks on Linux in nix develop:
- cargo fmt --all -- --check
- cargo clippy --locked --workspace --all-targets -- -D warnings
- cargo test --locked --workspace --all-targets, with only the inherited
  TSUNAGI_CONTROL_SOCKET removed from the test environment
- git diff --cached --check

Initial full-suite runs failed six DNS readiness checks because the inherited
system socket path overrode their temporary state directories. Clearing that
environment variable resolves those failures without changing application code.
Root/network-dependent ignored tests and other operating systems were not run.
2026-10-14 23:28:53 +01:00

2867 lines
106 KiB
Rust
Raw Blame History

//! Implements a socket that can change its communication path while in use, actively searching for the best way to communicate.
//!
//!
//! ### `RelayOnly` path selection:
//! When set this will force all packets to be sent over
//! the relay connection, regardless of whether or
//! not we have a direct UDP address for the given endpoint.
//!
//! The intended use is for testing the relay protocol inside the Socket
//! to ensure that we can rely on the relay to send packets when two endpoints
//! are unable to find direct UDP connections to each other.
//!
//! This also prevent this endpoint from attempting to hole punch and prevents it
//! from responding to any hole punching attempts. This endpoint will still,
//! however, read any packets that come off the UDP sockets.
use std::{
collections::{BTreeMap, BTreeSet},
fmt::Display,
io,
net::{IpAddr, SocketAddr},
sync::{
Arc, Mutex, RwLock,
atomic::{AtomicBool, Ordering},
},
};
use iroh_base::{EndpointAddr, EndpointId, RelayUrl, SecretKey, TransportAddr};
use iroh_relay::{RelayConfig, RelayMap};
use mapped_addrs::MultipathMappedAddr;
use n0_error::{AnyError, anyerr, bail, e, stack_error};
use n0_future::{
MaybeFuture,
task::{self, AbortOnDropHandle},
time::{self, Duration, Instant},
};
use n0_watcher::{self, Watchable, Watcher};
use netwatch::netmon;
#[cfg(not(wasm_browser))]
use netwatch::{
interfaces::{IpNet, Ipv6AddrFlags},
ip::LocalAddresses,
};
use noq::{
NetworkChangeHint, TokenStore,
crypto::rustls::{QuicClientConfig, QuicServerConfig},
};
use rand::RngExt;
use rustc_hash::FxHashSet;
use tokio::sync::{
Mutex as AsyncMutex,
mpsc::{self},
oneshot,
};
use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned};
use tracing::{Instrument, Level, Span, debug, error, event, info_span, instrument, trace, warn};
use transports::{LocalAddrsWatch, Transport, TransportConfig};
use url::Url;
use self::{
remote_map::{RemoteMap, RemoteStateMessage},
transports::{RelayActorConfig, Transports},
};
#[cfg(not(wasm_browser))]
use crate::dns::DnsResolver;
#[cfg(not(wasm_browser))]
use crate::net_report::QuicConfig;
use crate::{
address_lookup::{self, AddressLookupFailed, EndpointData, UserData},
defaults::timeouts::NET_REPORT_TIMEOUT,
endpoint::{
LocalTransportAddr, RelayStatus, hooks::EndpointHooksList, quic::QuicTransportConfig,
},
metrics::EndpointMetrics,
net_report::{self, IfStateDetails, Report},
portmapper,
runtime::Runtime,
socket::{
concurrent_read_map::ReadOnlyMap,
remote_map::{MappedAddrs, PathSelector, PathStateReceiver, RemoteInfo},
transports::{HomeRelayWatch, HomeRelayWatcher},
},
tls::{
self,
misc::{Blake3HmacKey, RustlsTokenKey},
},
};
mod metrics;
pub(crate) mod biased_rtt_path_selector;
pub(crate) mod concurrent_read_map;
pub(crate) mod mapped_addrs;
pub(crate) mod remote_map;
pub(crate) mod transports;
use self::mapped_addrs::{EndpointIdMappedAddr, MappedAddr};
pub use self::metrics::Metrics;
// TODO: Use this
// /// How long we consider a QAD-derived endpoint valid for. UDP NAT mappings typically
// /// expire at 30 seconds, so this is a few seconds shy of that.
// const ENDPOINTS_FRESH_ENOUGH_DURATION: Duration = Duration::from_secs(27);
/// The duration in which we send keep-alives.
///
/// If a path is idle for this long, a PING frame will be sent to keep the connection
/// alive.
pub(crate) const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(5);
/// The maximum time a path can stay idle before being closed.
///
/// 15s gives 3x [`HEARTBEAT_INTERVAL`] (5s) for multiple retry chances, and enough
/// margin for real-world outages (WiFi reconnect 2-5s, cellular handoff 2-10s).
/// iroh 0.35 used 10s at the QUIC level; tailscale uses 45s at the WireGuard session
/// level with 3s heartbeats.
pub(crate) const PATH_MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(15);
/// The maximum time a relay path can stay idle before being closed.
///
/// Relay paths need a longer idle timeout than direct paths because the relay actor
/// manages the WebSocket connection and transparently reconnects after network changes
/// or relay server restarts. During network outages the interface may be down for
/// 5-15s, during which no relay traffic flows. Once the interface recovers, the relay
/// actor reconnects (DNS + TCP + TLS + WebSocket upgrade), which adds another 1-2s.
///
/// Set to match the connection-level idle timeout (30s) so the relay path survives
/// as long as the connection itself.
pub(crate) const RELAY_PATH_MAX_IDLE_TIMEOUT: Duration = Duration::from_secs(30);
/// Maximum number of concurrent QUIC multipath paths per connection.
///
/// We expect 1 relay path, and then leave space for ~3 IP and custom transport paths.
/// On top of that, when we expect a network change, we might be closing these paths
/// (except for the relay path) and open new ones, and give us 3 more paths to spare.
/// And finally we round that up to 8 for good measure.
pub(crate) const MAX_MULTIPATH_PATHS: u32 = 8;
/// Maximum number of n0 QUIC NAT Traversal addresses that the QUIC stack should allow.
///
/// This needs to be big enough to accommodate for machines which have lots of network
/// interfaces enabled. We've seen MacOS machines with >25 interfaces in the wild
/// (mostly due to VPN TUN and docket interfaces), so this seems like a reasonable
/// value.
pub(crate) const MAX_QNT_ADDRESSES: u8 = 32;
/// Error returned when the endpoint state actor stopped while waiting for a reply.
#[stack_error(add_meta, derive)]
#[error("endpoint state actor stopped")]
#[derive(Clone)]
pub(crate) struct RemoteStateActorStoppedError;
impl From<mpsc::error::SendError<RemoteStateMessage>> for RemoteStateActorStoppedError {
#[track_caller]
fn from(_value: mpsc::error::SendError<RemoteStateMessage>) -> Self {
Self::new()
}
}
/// Contains options for `Socket::listen`.
#[derive(derive_more::Debug)]
pub(crate) struct Options {
/// The configuration for the different transports.
pub(crate) transports: Vec<TransportConfig>,
/// Secret key for this endpoint.
pub(crate) secret_key: SecretKey,
/// Optional user-defined Address Lookup data.
pub(crate) address_lookup_user_data: Option<UserData>,
/// A DNS resolver to use for resolving relay URLs.
///
/// You can use [`crate::dns::DnsResolver::new`] for a resolver
/// that uses the system's DNS configuration.
#[cfg(not(wasm_browser))]
pub(crate) dns_resolver: DnsResolver,
/// Proxy configuration.
pub(crate) proxy_url: Option<Url>,
/// TLS configuration for HTTPS and non-iroh-QUIC connections.
pub(crate) tls_config: rustls::ClientConfig,
/// ServerConfig for the internal QUIC endpoint
pub(crate) server_config: noq_proto::ServerConfig,
pub(crate) metrics: EndpointMetrics,
pub(crate) hooks: EndpointHooksList,
pub(crate) path_selector: Arc<dyn PathSelector>,
pub(crate) portmapper_config: portmapper::PortmapperConfig,
pub(crate) net_report_config: crate::net_report::NetReportConfig,
/// Static configuration for the endpoint.
pub(crate) static_config: StaticConfig,
/// Explicitly configured external addresses to advertise.
pub(crate) configured_addrs: BTreeSet<SocketAddr>,
}
/// Inner state for an iroh [`crate::Endpoint`].
///
/// Dereferences to [`Socket`], and handles closing.
#[derive(Debug, derive_more::Deref)]
pub(crate) struct EndpointInner {
#[deref(forward)]
sock: Arc<Socket>,
// empty when shutdown
actor_task: Mutex<Option<AbortOnDropHandle<()>>>,
/// Channel to send to the internal actor.
actor_sender: mpsc::Sender<ActorMessage>,
// noq endpoint
endpoint: noq::Endpoint,
// Runtime used by noq
runtime: Arc<Runtime>,
/// Static configuration for the endpoint.
pub(crate) static_config: StaticConfig,
}
impl Drop for EndpointInner {
fn drop(&mut self) {
if self.sock.is_closed() {
return;
}
tracing::error!(
"Endpoint dropped without calling `Endpoint::close`. Aborting ungracefully."
);
self.abort();
}
}
/// Configuration for a [`noq::Endpoint`] that cannot be changed at runtime.
#[derive(derive_more::Debug)]
pub(crate) struct StaticConfig {
pub(crate) tls_config: tls::TlsConfig,
#[debug("QuicServerConifg")]
pub(crate) server_config: QuicServerConfig,
#[debug("QuicClientConfig")]
pub(crate) client_config: QuicClientConfig,
#[debug("Arc<RustlsTokenKey>")]
pub(crate) token_key: Arc<RustlsTokenKey>,
#[debug("Arc<dyn TokenStore>")]
pub(crate) token_store: Arc<dyn TokenStore>,
pub(crate) transport_config: QuicTransportConfig,
}
impl StaticConfig {
/// Create a [`noq_proto::ServerConfig`] with the specified ALPN protocols.
pub(crate) fn create_server_config(
&self,
alpn_protocols: Vec<Vec<u8>>,
) -> noq_proto::ServerConfig {
let mut quic_server_config = self.server_config.clone();
quic_server_config.set_alpn_protocols(alpn_protocols);
let mut inner =
noq::ServerConfig::new(Arc::new(quic_server_config), self.token_key.clone());
inner.transport_config(self.transport_config.to_inner_arc());
inner
}
/// Create a [`noq_proto::ClientConfig`] with the specified ALPN protocols.
pub(crate) fn create_client_config(
&self,
alpn_protocols: Vec<Vec<u8>>,
transport_config: Arc<noq::TransportConfig>,
) -> noq_proto::ClientConfig {
let mut quic_client_config = self.client_config.clone();
quic_client_config.set_alpn_protocols(alpn_protocols);
let mut inner = noq::ClientConfig::new(Arc::new(quic_client_config));
inner.transport_config(transport_config);
inner.token_store(self.token_store.clone());
inner
}
}
/// This coordinates the shutdown of the [`Socket`] and all its tasks.
///
/// It also tightly binds to the [`EndpointInner`] and [`Actor`] closing as that is where
/// most of the logic lives.
#[derive(Debug)]
struct ShutdownState {
/// Token that is cancelled at the moment [`crate::Endpoint::close`] is called.
///
/// Currently cancelled from [`EndpointInner::close`].
at_close_start: CancellationToken,
/// Token that is cancelled once the [`noq::Endpoint`] is drained.
///
/// Only 100ms after this is cancelled will the [`Actor`] task be cancelled, it should
/// have exited already by then as it is considered an error if it was still running.
at_endpoint_closed: CancellationToken,
/// Set if the endpoint is closed and all tasks are stopped.
///
/// This is only set once both [`Self::at_close_start`] and [`Self::at_endpoint_closed`]
/// are cancelled **and** the [`Actor`] task is no longer running.
closed: AtomicBool,
}
impl Default for ShutdownState {
fn default() -> Self {
Self {
at_close_start: CancellationToken::new(),
at_endpoint_closed: CancellationToken::new(),
closed: AtomicBool::new(false),
}
}
}
impl ShutdownState {
/// Whether the endpoint has started closing, or is already closed.
///
/// This is true once [`crate::Endpoint::close`] is called, and remains true forever
/// after. Tasks might still be shutting down.
fn is_closing(&self) -> bool {
self.at_close_start.is_cancelled()
}
/// Whether the endpoint is fully closed and all tasks stopped.
///
/// The endpoint will be drained, all transports and sockets will be closed.
fn is_closed(&self) -> bool {
self.closed.load(Ordering::Relaxed)
}
}
/// Iroh connectivity layer.
///
/// This is responsible for routing packets to endpoints based on endpoint IDs, it will initially
/// route packets via a relay and transparently try and establish an endpoint-to-endpoint
/// connection and upgrade to it. It will also keep looking for better connections as the
/// network details of both endpoints change.
///
/// It is usually only necessary to use a single [`Socket`] instance in an application, it
/// means any QUIC endpoints on top will be sharing as much information about endpoints as
/// possible.
#[derive(Debug)]
pub(crate) struct Socket {
/// Read-only view of the per-remote `RemoteStateActor` inboxes.
///
/// Lets callers send to an existing `RemoteStateActor` without going through
/// the socket actor.
///
/// A missing entry means no actor is running for that remote. Spawning new
/// `RemoteStateActor`s must go through the socket actor channel.
remote_actors: ReadOnlyMap<EndpointId, mpsc::Sender<RemoteStateMessage>>,
// - Shutdown Management
shutdown: ShutdownState,
// - Networking Info
/// Our discovered direct addresses.
direct_addrs: DiscoveredDirectAddrs,
/// Our latest net-report
net_report: Watchable<(Option<Report>, UpdateReason)>,
/// If the last net_report report, reports IPv6 to be available.
ipv6_reported: Arc<AtomicBool>,
/// Maps for resolving mapped addrs to/from IP and relay addresses.
mapped_addrs: MappedAddrs,
/// Local addresses
local_addrs_watch: LocalAddrsWatch,
home_relay_watch: HomeRelayWatcher,
/// Currently bound IP addresses of all sockets
#[cfg(not(wasm_browser))]
ip_bind_addrs: Vec<SocketAddr>,
/// The DNS resolver to be used in this socket.
#[cfg(not(wasm_browser))]
dns_resolver: DnsResolver,
relay_map: RelayMap,
/// Optional Address Lookup
address_lookup: address_lookup::AddressLookupServices,
/// Optional user-defined discover data.
address_lookup_user_data: RwLock<Option<UserData>>,
/// Explicitly configured external addresses to advertise.
configured_addrs: RwLock<BTreeSet<SocketAddr>>,
pub(crate) tls_config: rustls::ClientConfig,
/// Metrics
pub(crate) metrics: EndpointMetrics,
pub(crate) hooks: EndpointHooksList,
/// Tracing span for this endpoint.
pub(crate) span: Span,
}
impl Socket {
/// Returns the relay endpoint we are connected to, that has the best latency.
///
/// If `None`, then we are not connected to any relay endpoints.
pub(crate) fn my_relay(&self) -> Option<RelayUrl> {
self.local_addr().into_iter().find_map(|a| {
if let transports::Addr::Relay(url, _) = a {
Some(url)
} else {
None
}
})
}
/// Whether the iroh endpoint is closed and all its actors stopped.
pub(crate) fn is_closed(&self) -> bool {
self.shutdown.is_closed()
}
/// Whether [`crate::Endpoint::close`] has been called.
fn is_closing(&self) -> bool {
self.shutdown.is_closing()
}
/// Returns a future that resolves once endpoint shutdown has started.
pub(crate) fn closed(&self) -> WaitForCancellationFutureOwned {
self.shutdown.at_close_start.clone().cancelled_owned()
}
/// Get the cached version of addresses.
pub(crate) fn local_addr(&self) -> Vec<transports::Addr> {
self.local_addrs_watch.clone().get()
}
#[cfg(not(wasm_browser))]
fn ip_bind_addrs(&self) -> &[SocketAddr] {
&self.ip_bind_addrs
}
fn ip_local_addrs(&self) -> impl Iterator<Item = SocketAddr> + use<> {
self.local_addr()
.into_iter()
.filter_map(|addr| addr.into_socket_addr())
}
/// Tries to send a [`RemoteStateMessage`] to the `RemoteStateActor` for given [`EndpointId`].
///
/// Returns an error if there currently is no remote state actor running for this, or when it
/// is currently shutting down.
pub(crate) fn try_send_remote_state_msg(
&self,
endpoint_id: EndpointId,
message: RemoteStateMessage,
) -> Result<(), RemoteStateMessage> {
let Some(sender) = self.remote_actors.get(&endpoint_id) else {
return Err(message);
};
sender.try_send(message).map_err(|err| err.into_inner())
}
/// Returns a [`Watcher`] for this socket's direct addresses.
///
/// The [`Socket`] continuously monitors the direct addresses, the network addresses
/// it might be able to be contacted on, for changes. Whenever changes are detected
/// this [`Watcher`] will yield a new list of addresses.
///
/// Upon the first creation on the [`Socket`] it may not yet have completed a first
/// net report to discover IP addresses, in this case the current item in this [`Watcher`] will be
/// [`None`]. Once the first set of ip addresses are discovered the [`Watcher`] will
/// store [`Some`] set of addresses.
///
/// To get the current direct addresses, use [`Watcher::initialized`].
///
/// [`Watcher`]: n0_watcher::Watcher
/// [`Watcher::initialized`]: n0_watcher::Watcher::initialized
pub(crate) fn ip_addrs(&self) -> n0_watcher::Direct<BTreeSet<DirectAddr>> {
self.direct_addrs.addrs.watch()
}
/// Returns a [`Watcher`] for this socket's net-report.
///
/// The [`Socket`] continuously monitors the network conditions for changes.
/// Whenever changes are detected this [`Watcher`] will yield a new report.
///
/// Upon the first creation on the [`Socket`] it may not yet have completed
/// a first net-report. In this case, the current item in this [`Watcher`] will
/// be [`None`]. Once the first report has been run, the [`Watcher`] will
/// store [`Some`] report.
///
/// To get the current `net-report`, use [`Watcher::initialized`].
///
/// [`Watcher`]: n0_watcher::Watcher
/// [`Watcher::initialized`]: n0_watcher::Watcher::initialized
#[cfg(feature = "unstable-net-report")]
pub(crate) fn net_report(&self) -> impl Watcher<Value = Option<Report>> + use<> {
self.net_report.watch().map(|(r, _)| r)
}
/// Watch for changes to the home relay.
///
/// Note that this can be used to wait for the initial home relay to be known using
/// [`Watcher::initialized`].
pub(crate) fn home_relay(&self) -> impl Watcher<Value = Vec<RelayUrl>> + use<> {
self.local_addrs_watch.clone().map(|addrs| {
addrs
.into_iter()
.filter_map(|addr| {
if let transports::Addr::Relay(url, _) = addr {
Some(url)
} else {
None
}
})
.collect()
})
}
pub(crate) fn home_relay_status(&self) -> impl Watcher<Value = Vec<RelayStatus>> + use<> {
self.home_relay_watch.clone()
}
/// Stores a new set of direct addresses.
///
/// If the direct addresses have changed from the previous set, they are published to
/// the address lookup system.
fn store_direct_addresses(&self, addrs: BTreeSet<DirectAddr>) {
let updated = self.direct_addrs.update(addrs);
if updated {
self.publish_my_addr();
}
}
/// Get a reference to the DNS resolver used in this [`Socket`].
#[cfg(not(wasm_browser))]
pub(crate) fn dns_resolver(&self) -> &DnsResolver {
&self.dns_resolver
}
/// Translates a possible IP-mapped [`SocketAddr`] into a [`transports::Addr`].
///
/// For regular IP addresses this returns `Addr::Ip`. For mapped addresses this performs
/// a reverse lookup.
pub(crate) fn to_transport_addr(&self, addr: SocketAddr) -> Option<transports::Addr> {
self.mapped_addrs.to_transport_addr(addr)
}
pub(crate) fn to_local_transport_addr(
&self,
local_ip: Option<IpAddr>,
remote_addr: SocketAddr,
) -> LocalTransportAddr {
let remote_addr = self.to_transport_addr(remote_addr).unwrap_or_else(|| {
error!(
mapped_addr = ?remote_addr,
"Socket::to_local_transport_addr: invalid mapped address",
);
transports::Addr::Ip(remote_addr)
});
LocalTransportAddr::from_noq_local_ip(
local_ip,
&remote_addr,
&self.mapped_addrs.custom_addrs,
)
}
/// Reference to the internal Address Lookup
pub(crate) fn address_lookup(&self) -> &address_lookup::AddressLookupServices {
&self.address_lookup
}
/// Updates the user-defined Address Lookup data for this endpoint.
pub(crate) fn set_user_data_for_address_lookup(&self, user_data: Option<UserData>) {
let mut guard = self
.address_lookup_user_data
.write()
.expect("lock poisened");
if *guard != user_data {
*guard = user_data;
drop(guard);
self.publish_my_addr();
}
}
/// Process datagrams received from all the transports.
///
/// All the `bufs` and `metas` should have initialized packets in them.
///
/// This fixes up the datagrams to use the correct [`MultipathMappedAddr`] and extracts
/// DISCO packets, processing them inside the socket.
///
/// [`MultipathMappedAddr`]: mapped_addrs::MultipathMappedAddr
fn process_datagrams(
&self,
bufs: &mut [io::IoSliceMut<'_>],
metas: &mut [noq_udp::RecvMeta],
recv_infos: &[transports::RecvInfo],
) {
assert_eq!(bufs.len(), metas.len(), "non matching bufs & metas");
assert_eq!(
bufs.len(),
recv_infos.len(),
"non matching bufs & recv_infos"
);
// zip is slow :(
for i in 0..metas.len() {
let noq_meta = &mut metas[i];
let recv_info = &recv_infos[i];
let remote_addr = recv_info.remote();
let datagram_count = if noq_meta.stride == 0 {
if noq_meta.len > 0 {
warn!(
src = ?remote_addr,
len = noq_meta.len,
"received datagram with stride=0 but len>0",
);
// fix the weird len
noq_meta.len = 0;
}
// one empty datagram
1
} else {
noq_meta.len.div_ceil(noq_meta.stride)
};
self.metrics
.socket
.recv_datagrams
.inc_by(datagram_count as _);
if noq_meta.len > noq_meta.stride {
trace!(
src = ?remote_addr,
len = noq_meta.len,
stride = %noq_meta.stride,
datagram_count,
"GRO datagram received",
);
self.metrics.socket.recv_gro_datagrams.inc();
} else {
trace!(src = ?remote_addr, len = noq_meta.len, "datagram received");
}
match remote_addr {
transports::Addr::Ip(SocketAddr::V4(..)) => {
self.metrics.socket.recv_data_ipv4.inc_by(noq_meta.len as _);
}
transports::Addr::Ip(SocketAddr::V6(..)) => {
self.metrics.socket.recv_data_ipv6.inc_by(noq_meta.len as _);
}
transports::Addr::Relay(src_url, src_node) => {
self.metrics
.socket
.recv_data_relay
.inc_by(noq_meta.len as _);
// Fill in the correct mapped address
let mapped_addr = self
.mapped_addrs
.relay_addrs
.get(&(src_url.clone(), *src_node));
noq_meta.addr = mapped_addr.private_socket_addr();
}
transports::Addr::Custom(remote) => {
self.metrics
.socket
.recv_data_custom
.inc_by(noq_meta.len as _);
// Fill in the correct mapped address
let mapped_addr = self.mapped_addrs.custom_addrs.get(remote);
noq_meta.addr = mapped_addr.private_socket_addr();
if let Some(local) = recv_info.local() {
let local_mapped = self.mapped_addrs.custom_addrs.get(local);
noq_meta.dst_ip = Some(local_mapped.private_socket_addr().ip());
}
}
}
}
}
/// Publishes our address to an address lookup service, if configured.
///
/// Called whenever our addresses or home relay endpoint changes.
fn publish_my_addr(&self) {
let relay_url = self.my_relay();
let mut addrs: Vec<_> = self
.direct_addrs
.sockaddrs()
.map(TransportAddr::Ip)
.collect();
let user_data = self
.address_lookup_user_data
.read()
.expect("lock poisened")
.clone();
if relay_url.is_none() && addrs.is_empty() && user_data.is_none() {
// do not bother publishing if we don't have any information
return;
}
if let Some(url) = relay_url {
addrs.push(TransportAddr::Relay(url));
}
let mut data = EndpointData::new(addrs);
data.set_user_data(user_data);
self.address_lookup.publish(&data);
}
}
/// Manages currently running net reports to learn this endpoint's IP addresses.
///
/// Invariants:
/// - only one direct addr update must be running at a time
/// - if an update is scheduled while another one is running, remember that
/// and start a new one when the current one has finished
#[derive(Debug)]
struct DirectAddrUpdateState {
/// If set, start a new update as soon as the current one is finished.
want_update: Option<UpdateReason>,
sock: Arc<Socket>,
port_mapper: portmapper::Client,
/// The prober that discovers local network conditions, including the closest relay relay and NAT mappings.
net_reporter: Arc<AsyncMutex<net_report::Client>>,
relay_map: RelayMap,
run_done: mpsc::Sender<()>,
shutdown_token: CancellationToken,
}
#[derive(Default, Debug, PartialEq, Eq, Clone, Copy)]
enum UpdateReason {
/// Initial state
#[default]
None,
Periodic,
PortmapUpdated,
LinkChangeMajor,
LinkChangeMinor,
RelayMapChange,
}
impl UpdateReason {
fn is_major(self) -> bool {
matches!(self, Self::LinkChangeMajor | Self::RelayMapChange)
}
}
impl DirectAddrUpdateState {
fn new(
sock: Arc<Socket>,
port_mapper: portmapper::Client,
net_reporter: Arc<AsyncMutex<net_report::Client>>,
relay_map: RelayMap,
run_done: mpsc::Sender<()>,
shutdown_token: CancellationToken,
) -> Self {
DirectAddrUpdateState {
want_update: Default::default(),
port_mapper,
net_reporter,
sock,
relay_map,
run_done,
shutdown_token,
}
}
/// Schedules a new run, either starting it immediately if none is running or
/// scheduling it for later.
fn schedule_run(&mut self, why: UpdateReason, if_state: IfStateDetails) {
match self.net_reporter.clone().try_lock_owned() {
Ok(net_reporter) => {
self.run(why, if_state, net_reporter);
}
Err(_) => {
let _ = self.want_update.insert(why);
}
}
}
/// If another run is needed, triggers this run, otherwise does nothing.
fn try_run(&mut self, if_state: IfStateDetails) {
match self.net_reporter.clone().try_lock_owned() {
Ok(net_reporter) => {
if let Some(why) = self.want_update.take() {
self.run(why, if_state, net_reporter);
}
}
Err(_) => {
// do nothing
}
}
}
/// Trigger a new run.
fn run(
&mut self,
why: UpdateReason,
if_state: IfStateDetails,
mut net_reporter: tokio::sync::OwnedMutexGuard<net_report::Client>,
) {
debug!("starting direct addr update ({:?})", why);
// Don't start a net report probe if we know
// we are shutting down
if self.shutdown_token.is_cancelled() {
debug!("skipping net_report, socket is shutting down");
// deactivate portmapper
self.port_mapper.deactivate();
return;
}
if self.relay_map.is_empty() {
debug!("skipping net_report, empty RelayMap");
self.sock.net_report.set((None, why)).ok();
return;
}
self.sock.metrics.net_report.portmap_attempts.inc();
self.port_mapper.procure_mapping();
trace!("requesting net_report report");
let sock = self.sock.clone();
let run_done = self.run_done.clone();
// Ensure that reports are cancelled when we shutdown
let token = self.shutdown_token.child_token();
let inner_token = token.child_token();
task::spawn(
async move {
let fut = token.run_until_cancelled(time::timeout(
NET_REPORT_TIMEOUT,
net_reporter.get_report(if_state, why.is_major(), inner_token),
));
match fut.await {
Some(Ok(report)) => {
sock.net_report.set((Some(report), why)).ok();
}
Some(Err(time::Elapsed { .. })) => {
warn!("net_report report timed out");
}
None => {
trace!("net_report cancelled");
}
}
// mark run as finished
debug!("direct addr update done ({:?})", why);
run_done.send(()).await.ok();
}
.instrument(tracing::Span::current()),
);
}
}
#[allow(missing_docs)]
#[stack_error(derive, add_meta)]
#[non_exhaustive]
pub enum BindError {
#[error("Failed to bind sockets")]
Sockets { source: io::Error },
#[error("Failed to create internal QUIC endpoint")]
CreateQuicEndpoint { source: io::Error },
#[error("Failed to create netmon monitor")]
CreateNetmonMonitor { source: AnyError },
#[error("Invalid transport configuration")]
InvalidTransportConfig,
#[error("Invalid CA root configuration")]
InvalidCaRootConfig { source: io::Error },
#[error("Failed to create an address lookup service")]
AddressLookup {
#[error(from)]
source: crate::address_lookup::AddressLookupBuilderError,
},
#[error("Missing or incompatible rustls crypto provider configured")]
InvalidCryptoProvider,
#[error("Error constructing TLS configuration")]
TlsConfigError {
#[error(from)]
source: tls::TlsConfigError,
},
}
impl EndpointInner {
/// Creates a [`EndpointInner`].
pub(crate) async fn bind(opts: Options) -> Result<Self, BindError> {
// Use the current span as the main span for all tasks spawned in this endpoint.
// `EndpointInner::bind` is not public and only called from `crate::endpoint::Builder::bind`,
// which instruments the call with a span created for this purpose.
let span = tracing::Span::current();
let Options {
secret_key,
transports: transport_configs,
address_lookup_user_data,
#[cfg(not(wasm_browser))]
dns_resolver,
proxy_url,
server_config,
tls_config,
metrics,
hooks,
path_selector,
portmapper_config,
net_report_config,
static_config,
configured_addrs,
} = opts;
let address_lookup = address_lookup::AddressLookupServices::default();
let port_mapper = portmapper::create_client(&portmapper_config);
let relay_transport_configs: Vec<_> = transport_configs
.iter()
.filter(|t| matches!(t, TransportConfig::Relay { .. }))
.collect();
// Currently we only support a single relay transport
if relay_transport_configs.len() > 1 {
bail!(BindError::InvalidTransportConfig);
}
let relay_map = relay_transport_configs
.iter()
.filter_map(|t| {
#[allow(irrefutable_let_patterns)]
if let TransportConfig::Relay { relay_map, .. } = t {
Some(relay_map.clone())
} else {
None
}
})
.next()
.unwrap_or_else(RelayMap::empty);
let ipv6_reported = Arc::new(AtomicBool::new(false));
let relay_actor_config = RelayActorConfig {
my_relay: HomeRelayWatch::default(),
secret_key: secret_key.clone(),
#[cfg(not(wasm_browser))]
dns_resolver: dns_resolver.clone(),
proxy_url: proxy_url.clone(),
ipv6_reported: ipv6_reported.clone(),
tls_config: tls_config.clone(),
metrics: metrics.socket.clone(),
relay_map: relay_map.clone(),
};
let shutdown_state = ShutdownState::default();
let shutdown_token = shutdown_state.at_endpoint_closed.child_token();
let transports = Transports::bind(
&transport_configs,
relay_actor_config,
&metrics,
shutdown_token.child_token(),
)
.map_err(|err| e!(BindError::Sockets, err))?;
if let Some(v4_port) = transports.local_addrs().into_iter().find_map(|t| {
if let transports::Addr::Ip(SocketAddr::V4(addr)) = t {
Some(addr.port())
} else {
None
}
}) {
// NOTE: we can end up with a zero port if `netwatch::UdpSocket::socket_addr` fails
match v4_port.try_into() {
Ok(non_zero_port) => {
port_mapper.update_local_port(non_zero_port);
}
Err(_zero_port) => debug!("Skipping port mapping with zero local port"),
}
}
let (actor_sender, actor_receiver) = mpsc::channel(256);
#[cfg(not(wasm_browser))]
let has_ipv6_transport = transports
.ip_bind_addrs()
.into_iter()
.any(|addr| addr.is_ipv6());
#[cfg(not(wasm_browser))]
let has_ip_transports = !transports.ip_bind_addrs().is_empty();
let direct_addrs = DiscoveredDirectAddrs::default();
let remote_map = {
RemoteMap::new(
metrics.socket.clone(),
direct_addrs.addrs.watch(),
address_lookup.clone(),
shutdown_token.child_token(),
path_selector,
span.clone(),
)
};
let home_relay_watch = transports.home_relay_watch();
let sock = Arc::new(Socket {
remote_actors: remote_map.senders(),
shutdown: shutdown_state,
ipv6_reported,
mapped_addrs: remote_map.mapped_addrs.clone(),
address_lookup,
relay_map: relay_map.clone(),
address_lookup_user_data: RwLock::new(address_lookup_user_data),
configured_addrs: RwLock::new(configured_addrs),
direct_addrs,
net_report: Watchable::new((None, UpdateReason::None)),
#[cfg(not(wasm_browser))]
dns_resolver: dns_resolver.clone(),
metrics: metrics.clone(),
local_addrs_watch: transports.local_addrs_watch(),
home_relay_watch,
#[cfg(not(wasm_browser))]
ip_bind_addrs: transports.ip_bind_addrs(),
tls_config: tls_config.clone(),
hooks,
span: span.clone(),
});
let mut endpoint_config =
noq::EndpointConfig::new(Arc::new(Blake3HmacKey::new(&mut rand::rng())));
// Setting this to false means that noq will ignore packets that have the QUIC fixed bit
// set to 0. The fixed bit is the 3rd bit of the first byte of a packet.
// For performance reasons and to not rewrite buffers we pass non-QUIC UDP packets straight
// through to noq. We set the first byte of the packet to zero, which makes noq ignore
// the packet if grease_quic_bit is set to false.
endpoint_config.grease_quic_bit(false);
let local_addrs_watch = transports.local_addrs_watch();
let transports_network_change = transports.create_network_change_sender();
let runtime = Arc::new(Runtime::new(secret_key.public()));
let endpoint = noq::Endpoint::new_with_abstract_socket(
endpoint_config,
Some(server_config),
Box::new(Transport::new(sock.clone(), transports)),
runtime.clone(),
)
.map_err(|err| e!(BindError::CreateQuicEndpoint, err))?;
let network_monitor = netmon::Monitor::new()
.await
.map_err(|err| e!(BindError::CreateNetmonMonitor, anyerr!(err)))?;
#[cfg(not(wasm_browser))]
let net_report_config = {
// Set a `QuicConfig` for address discovery (QAD), but only if we have IP transports.
//
// If there are no IP transports configured, then we don't set a QuicConfig.
// If we would, the `noq::Endpoint` passed along will not have IP connectivity,
// and the QAD probes that connect to the relay's QUIC endpoints would time out
// because all outgoing packets to IP destinations would be dropped.
let qad_config = has_ip_transports.then(|| QuicConfig {
ep: endpoint.clone(),
client_config: tls_config.clone(),
ipv4: true,
ipv6: has_ipv6_transport,
});
net_report::Options::new(tls_config.clone())
.quic_config(qad_config)
.proxy_url(proxy_url.clone())
.net_report_config(net_report_config)
};
#[cfg(wasm_browser)]
let net_report_config = net_report::Options::default().net_report_config(net_report_config);
let net_reporter = net_report::Client::new(
#[cfg(not(wasm_browser))]
dns_resolver,
relay_map.clone(),
net_report_config,
metrics.net_report.clone(),
);
let (direct_addr_done_tx, direct_addr_done_rx) = mpsc::channel(8);
let direct_addr_update_state = DirectAddrUpdateState::new(
sock.clone(),
port_mapper,
Arc::new(AsyncMutex::new(net_reporter)),
relay_map,
direct_addr_done_tx,
sock.shutdown.at_close_start.child_token(),
);
let local_interfaces_watcher = network_monitor.interface_state();
#[cfg_attr(not(wasm_browser), allow(unused_mut))]
let mut actor = Actor {
endpoint: endpoint.clone(),
sock: sock.clone(),
remote_map,
periodic_re_stun_timer: new_re_stun_timer(false),
network_monitor,
local_interfaces_watcher,
direct_addr_update_state,
transports_network_change,
direct_addr_done_rx,
call_notify_quic_network_change: None,
};
// Initialize addresses
#[cfg(not(wasm_browser))]
actor.update_direct_addresses(None);
let actor_task = task::spawn(
actor
.run(
actor_receiver,
shutdown_token.child_token(),
local_addrs_watch,
)
.instrument(info_span!(parent: span, "actor")),
);
let actor_task = Mutex::new(Some(AbortOnDropHandle::new(actor_task)));
Ok(EndpointInner {
sock,
actor_sender,
actor_task,
endpoint,
runtime,
static_config,
})
}
/// Returns a reference to the underlying [`noq::Endpoint`].
pub(crate) fn noq_endpoint(&self) -> &noq::Endpoint {
&self.endpoint
}
/// Closes the iroh endpoint.
///
/// Only the first close does anything. Any later closes return nil. Polling the socket
/// ([`noq::AsyncUdpSocket::poll_recv`]) will return [`Poll::Pending`] indefinitely
/// after this call.
///
/// [`Poll::Pending`]: std::task::Poll::Pending
#[instrument(skip_all, parent = self.sock.span.clone())]
pub(crate) async fn close(&self) {
if self.sock.is_closed() || self.sock.is_closing() {
return;
}
trace!("socket closing...");
// Cancel at_close_start token, which cancels running netreports.
self.sock.shutdown.at_close_start.cancel();
// Remove address lookup services
self.sock.address_lookup().clear();
// Initiate closing all connections, and refuse future connections.
self.noq_endpoint().close(0u16.into(), b"");
// In the history of this code, this call had been
// - removed: https://github.com/n0-computer/iroh/pull/1753
// - then added back in: https://github.com/n0-computer/iroh/pull/2227/files#diff-ba27e40e2986a3919b20f6b412ad4fe63154af648610ea5d9ed0b5d5b0e2d780R573
// - then removed again: https://github.com/n0-computer/iroh/pull/3165
// and finally added back in together with this comment.
// So before removing this call, please consider carefully.
// Among other things, this call tries its best to make sure that any queued close frames
// (e.g. via the call to `endpoint.close(...)` above), are flushed out to the sockets
// *and acknowledged* (or time out with the "probe timeout" of usually 3 seconds).
// This allows the other endpoints for these connections to be notified to release
// their resources, or - depending on the protocol - that all data was received.
// With the current noq API, this is the only way to ensure protocol code can use
// connection close codes, and close the endpoint properly.
// If this call is skipped, then connections that protocols close just shortly before the
// call to `Endpoint::close` will in most cases cause connection time-outs on remote ends.
trace!("wait_all_draining start");
self.noq_endpoint().wait_all_draining().await;
trace!("wait_all_draining done");
// Start cancellation of all actors.
self.sock.shutdown.at_endpoint_closed.cancel();
// MutexGuard is not held across await points
let task = self.actor_task.lock().expect("poisoned").take();
if let Some(task) = task {
// give the tasks a moment to shutdown cleanly
let shutdown_done = time::timeout(Duration::from_millis(100), async move {
if let Err(err) = task.await {
warn!("unexpected error in task shutdown: {:?}", err);
}
})
.await;
match shutdown_done {
Ok(_) => trace!("tasks finished in time, shutdown complete"),
Err(time::Elapsed { .. }) => {
// Dropping the task will abort it
warn!("tasks didn't finish in time, aborting");
}
}
}
// Waits for the EndpointDriver and all ConnectionDrivers to shut down
// Expects that the `noq::Endpoint` has been closed before this call,
// otherwise, the runtime will never shutdown.
self.runtime.shutdown().await;
self.sock.shutdown.closed.store(true, Ordering::SeqCst);
trace!("socket closed");
}
/// Aborts the endpoint ungracefully:
///
/// - Calls cancellation token that stops running net reports
/// - Removes all address lookup services
/// - Calls cancellation token that stops all the Socket actors
/// - Aborts the runtime
/// - Drops the actor task
/// - Sets the `Socket::is_closed` state to true
///
/// This does not wait for any current connections or tasks to close gracefully.
///
/// This should only be called in the `iroh::Endpoint` `Drop` impl when the
/// `iroh::Endpoint` is dropped without first calling `Endpoint::close`.
#[instrument(skip_all, parent = self.sock.span.clone())]
pub(crate) fn abort(&self) {
if self.sock.is_closed() || self.sock.is_closing() {
return;
}
trace!("socket aborting...");
// Cancel at_close_start token, which cancels running netreports.
self.sock.shutdown.at_close_start.cancel();
self.sock.address_lookup().clear();
// Cancel all actors.
self.sock.shutdown.at_endpoint_closed.cancel();
// Aborts all tasks, not waiting for any to close gracefully.
self.runtime.abort();
self.actor_task.lock().expect("poisoned").take();
self.sock.shutdown.closed.store(true, Ordering::SeqCst);
trace!("socket closed");
}
pub(crate) async fn insert_relay(
&self,
relay: RelayUrl,
endpoint: Arc<RelayConfig>,
) -> Option<Arc<RelayConfig>> {
let res = self.relay_map.insert(relay, endpoint);
self.actor_sender
.send(ActorMessage::RelayMapChange)
.await
.ok();
res
}
pub(crate) async fn remove_relay(&self, relay: &RelayUrl) -> Option<Arc<RelayConfig>> {
let res = self.relay_map.remove(relay);
self.actor_sender
.send(ActorMessage::RelayMapChange)
.await
.ok();
res
}
/// Adds an external address to advertise to peers.
pub(crate) async fn add_external_addr(&self, addr: SocketAddr) {
self.sock
.configured_addrs
.write()
.expect("poisoned")
.insert(addr);
self.actor_sender
.send(ActorMessage::DirectAddrRefresh)
.await
.ok();
}
/// Removes a configured external address. Returns `true` if it was present.
pub(crate) async fn remove_external_addr(&self, addr: &SocketAddr) -> bool {
let removed = self
.sock
.configured_addrs
.write()
.expect("poisoned")
.remove(addr);
if removed {
self.actor_sender
.send(ActorMessage::DirectAddrRefresh)
.await
.ok();
}
removed
}
/// Call to notify the system of potential network changes.
pub(crate) async fn network_change(&self) {
self.actor_sender
.send(ActorMessage::NetworkChange)
.await
.ok();
}
#[cfg(all(test, with_crypto_provider))]
async fn force_network_change(&self, is_major: bool) {
self.actor_sender
.send(ActorMessage::ForceNetworkChange(is_major))
.await
.ok();
}
/// Resolves an [`EndpointAddr`] to an [`EndpointIdMappedAddr`] to connect to via [`EndpointInner::endpoint`].
///
/// This starts a `RemoteStateActor` for the remote if not running already, and then checks
/// if the actor has any known paths to the remote. If not, it starts address lookup and waits for
/// at least one result to arrive.
///
/// Returns `Ok(Ok(EndpointIdMappedAddr))` if there is a known path or Address Lookup produced
/// at least one result. This does not mean there is a working path, only that we have at least
/// one transport address we can try to connect to.
///
/// Returns `Ok(Err(address_lookup_error))` if there are no known paths to the remote and Address Lookup
/// failed or produced no results. This means that we don't have any transport address for
/// the remote, thus there is no point in trying to connect over the noq endpoint.
///
/// Returns `Err(RemoteStateActorStoppedError)` if the `RemoteStateActor` for the remote has stopped,
/// which may never happen and thus is a bug if it does.
pub(crate) async fn resolve_remote(
&self,
addr: EndpointAddr,
) -> Result<Result<EndpointIdMappedAddr, AddressLookupFailed>, RemoteStateActorStoppedError>
{
let (tx, rx) = oneshot::channel();
let remote_id = addr.id;
self.actor_sender
.send(ActorMessage::ResolveRemote(addr, tx))
.await
.ok();
let reply = rx.await.map_err(|_| RemoteStateActorStoppedError::new())?;
match reply {
Ok(()) => Ok(Ok(self.mapped_addrs.endpoint_addrs.get(&remote_id))),
Err(err) => Ok(Err(err)),
}
}
/// Fetches the [`RemoteInfo`] about a remote from the `RemoteStateActor`.
///
/// Returns `None` if no actor is running for the remote.
pub(crate) async fn remote_info(&self, id: EndpointId) -> Option<RemoteInfo> {
let (tx, rx) = oneshot::channel();
self.remote_actors
.get(&id)?
.send(RemoteStateMessage::RemoteInfo(tx))
.await
.ok()?;
rx.await.ok()
}
/// Registers the connection in the `RemoteStateActor`.
///
/// The actor is responsible for holepunching and opening additional paths to this
/// connection.
///
/// Returns a future that resolves to a [`PathStateReceiver`] for the new connection.
///
/// The returned future is `'static`, so it can be stored without being lifetime-bound to `&self`.
pub(crate) fn register_connection(
&self,
remote: EndpointId,
conn: noq::Connection,
) -> impl Future<Output = Result<PathStateReceiver, RemoteStateActorStoppedError>> + Send + 'static
{
let (tx, rx) = oneshot::channel();
let sender = self.actor_sender.clone();
async move {
sender
.send(ActorMessage::AddConnection(remote, conn, tx))
.await
.map_err(|_| RemoteStateActorStoppedError::new())?;
rx.await.map_err(|_| RemoteStateActorStoppedError::new())
}
}
}
#[derive(derive_more::Debug)]
#[allow(clippy::enum_variant_names)]
enum ActorMessage {
NetworkChange,
RelayMapChange,
#[debug("ResolveRemote(..)")]
ResolveRemote(
EndpointAddr,
oneshot::Sender<Result<(), AddressLookupFailed>>,
),
#[debug("AddConnection(..)")]
AddConnection(
EndpointId,
noq::Connection,
oneshot::Sender<PathStateReceiver>,
),
/// Re-evaluate direct addresses, e.g. after configured external addresses changed.
DirectAddrRefresh,
#[cfg(all(test, with_crypto_provider))]
ForceNetworkChange(bool),
}
/// State for polling until a default route is available after a network change.
///
/// When a network change is detected but no default route exists yet (e.g.,
/// interface just came up but gateway not assigned), we poll with exponential
/// backoff until the gateway appears. This avoids the fixed 2s delay that was
/// too slow for interface recovery scenarios.
struct PendingNetworkChangeNotify {
/// Next time to check for default route.
next_check: Instant,
/// Current backoff interval.
interval: Duration,
/// Whether this was a major change.
is_major: bool,
/// When we started polling (to enforce a max wait).
started: Instant,
}
impl PendingNetworkChangeNotify {
const INITIAL_INTERVAL: Duration = Duration::from_millis(100);
const MAX_INTERVAL: Duration = Duration::from_secs(1);
const MAX_WAIT: Duration = Duration::from_secs(5);
fn new(is_major: bool) -> Self {
Self {
next_check: Instant::now() + Self::INITIAL_INTERVAL,
interval: Self::INITIAL_INTERVAL,
is_major,
started: Instant::now(),
}
}
/// Advance to the next check interval (exponential backoff, capped).
fn advance(&mut self) {
self.interval = (self.interval * 2).min(Self::MAX_INTERVAL);
self.next_check = Instant::now() + self.interval;
}
/// Whether we've exceeded the maximum wait time.
fn expired(&self) -> bool {
self.started.elapsed() >= Self::MAX_WAIT
}
}
struct Actor {
/// A clone of the quinn Endpoint.
///
/// The task of this actor is currently owned by the [`crate::Endpoint`] and wrapped in
/// an [`AbortOnDropHandle`]. When [`crate::Endpoint::close`] is called various
/// subsystems are being stopped. Then, when [`ShutdownState::at_endpoint_closed`] is
/// called by [`crate::Endpoint::close`], this actor itself is stopped via it's
/// [`CancellationToken`] and we will drop this clone of the endpoint. The endpoint is
/// then finally dropped when the [`crate::Endpoint`] itself is dropped.
///
/// All of this to say: keeping the quinn endpoint alive here does not impact the
/// lifetime of it since it's lifetime is shorter than that one that's stored in the
/// [`crate::Endpoint`].
endpoint: noq::Endpoint,
/// Shared state between an awful lot of iroh subsystems.
///
/// In particular both the [`EndpointInner`] as well as this actor itself have a
/// copy. But also other subsystems that consequently have access to way to much state.
sock: Arc<Socket>,
/// Tracks the networkmap endpoint entity for each endpoint discovery key.
remote_map: RemoteMap,
/// When set, is an AfterFunc timer that will call Socket::do_periodic_stun.
periodic_re_stun_timer: time::Interval,
/// An actor watching the local network interfaces.
///
/// The monitored changes are emitted via [`Self::local_interfaces_watcher`].
network_monitor: netmon::Monitor,
/// Watcher for changes to the local network interfaces, IP addresses and routes.
local_interfaces_watcher: n0_watcher::Direct<netmon::State>,
transports_network_change: transports::NetworkChangeSender,
/// Indicates the direct addr update state.
direct_addr_update_state: DirectAddrUpdateState,
direct_addr_done_rx: mpsc::Receiver<()>,
/// Polling state for [`Actor::notify_quic_network_change`].
///
/// When a network change is detected but no default route is available yet,
/// we poll with exponential backoff (100ms, 200ms, 400ms, 800ms, 1s, 1s, ...)
/// until the gateway appears. Once it does, we notify immediately.
/// After 5s total we notify anyway even without a gateway.
call_notify_quic_network_change: Option<PendingNetworkChangeNotify>,
}
impl Actor {
async fn run(
mut self,
mut msg_receiver: mpsc::Receiver<ActorMessage>,
shutdown_token: CancellationToken,
mut local_addrs_watcher: impl Watcher<Value = Vec<transports::Addr>> + Send + Sync,
) {
// Setup network monitoring
let mut current_netmon_state = self.local_interfaces_watcher.get();
let mut portmap_watcher = self
.direct_addr_update_state
.port_mapper
.watch_external_address();
let mut receiver_closed = false;
let mut portmap_watcher_closed = false;
let mut net_report_watcher = self.sock.net_report.watch();
// ensure we are doing an initial publish of our addresses
self.sock.publish_my_addr();
while !shutdown_token.is_cancelled() {
self.sock.metrics.socket.actor_tick_main.inc();
let portmap_watcher_changed = portmap_watcher.changed();
let notify_quic_network_change = match &self.call_notify_quic_network_change {
Some(pending) => {
MaybeFuture::Some(n0_future::time::sleep_until(pending.next_check))
}
None => MaybeFuture::None,
};
n0_future::pin!(notify_quic_network_change);
tokio::select! {
_ = shutdown_token.cancelled() => {
debug!("tick: shutting down");
return;
}
msg = msg_receiver.recv(), if !receiver_closed => {
let Some(msg) = msg else {
trace!("tick: socket receiver closed");
self.sock.metrics.socket.actor_tick_other.inc();
receiver_closed = true;
continue;
};
trace!(?msg, "tick: msg");
self.sock.metrics.socket.actor_tick_msg.inc();
self.handle_actor_message(msg).await;
}
tick = self.periodic_re_stun_timer.tick() => {
trace!("tick: re_stun {:?}", tick);
self.sock.metrics.socket.actor_tick_re_stun.inc();
self.re_stun(UpdateReason::Periodic);
}
new_addr = local_addrs_watcher.updated() => {
match new_addr {
Ok(addrs) => {
if !addrs.is_empty() {
trace!(?addrs, "local addrs");
self.sock.publish_my_addr();
}
}
Err(_) => {
warn!("local addr watcher stopped");
}
}
}
report = net_report_watcher.updated() => {
match report {
Ok((report, _)) => {
self.handle_net_report_report(report);
#[cfg(not(wasm_browser))]
{
self.periodic_re_stun_timer = new_re_stun_timer(true);
}
}
Err(_) => {
warn!("net report watcher stopped");
}
}
}
reason = self.direct_addr_done_rx.recv() => {
match reason {
Some(()) => {
// check if a new run needs to be scheduled
let state = self.local_interfaces_watcher.get();
self.direct_addr_update_state.try_run(state.into());
}
None => {
warn!("direct addr watcher died");
}
}
}
change = portmap_watcher_changed, if !portmap_watcher_closed => {
if change.is_err() {
trace!("tick: portmap watcher closed");
self.sock.metrics.socket.actor_tick_other.inc();
portmap_watcher_closed = true;
continue;
}
trace!("tick: portmap changed");
self.sock.metrics.socket.actor_tick_portmap_changed.inc();
let new_external_address = *portmap_watcher.borrow();
if new_external_address.is_some() {
self.sock.metrics.net_report.portmap_external_address_updated.inc();
}
debug!("external address updated: {new_external_address:?}");
self.re_stun(UpdateReason::PortmapUpdated);
},
state = self.local_interfaces_watcher.updated() => {
let Ok(state) = state else {
trace!("tick: link change receiver closed");
self.sock.metrics.socket.actor_tick_other.inc();
continue;
};
let is_major = state.is_major_change(&current_netmon_state);
event!(
target: "iroh::_events::link_change",
Level::DEBUG,
?state,
is_major
);
current_netmon_state = state;
self.sock.metrics.socket.actor_link_change.inc();
self.handle_network_change(is_major);
}
_remote_id = self.remote_map.cleanup() => {},
_ = &mut notify_quic_network_change => {
let has_network = self.has_usable_network();
let Some(pending) = self.call_notify_quic_network_change.as_mut() else {
continue;
};
if has_network || pending.expired() {
// Gateway appeared or we've waited long enough, notify now.
let is_major = pending.is_major;
self.call_notify_quic_network_change = None;
self.notify_quic_network_change(is_major);
} else {
// No gateway yet, back off and try again.
trace!(
interval = ?pending.interval,
elapsed = ?pending.started.elapsed(),
"no default route yet, retrying"
);
pending.advance();
}
}
else => {
trace!("tick: else");
}
}
}
}
/// Whether the local network has a default route and at least one IP address.
fn has_usable_network(&mut self) -> bool {
#[cfg(target_family = "wasm")]
{
true
}
#[cfg(not(target_family = "wasm"))]
{
let interfaces = self.local_interfaces_watcher.get();
interfaces.default_route_interface.is_some()
&& (interfaces.have_v4 || interfaces.have_v6)
}
}
/// Handles a change detected in the local network conditions.
///
/// This is triggered when the netmon actor detects a change in the local network
/// interfaces, assigned IP addresses and routes.
fn handle_network_change(&mut self, is_major: bool) {
debug!(is_major, "link change detected");
if is_major {
if let Err(err) = self.transports_network_change.rebind() {
warn!("failed to rebind transports: {err:?}");
}
self.transports_network_change.check_relay_connection();
#[cfg(not(wasm_browser))]
self.sock.dns_resolver.reset();
self.re_stun(UpdateReason::LinkChangeMajor);
} else {
self.re_stun(UpdateReason::LinkChangeMinor);
}
if self.has_usable_network() {
// This is considered a usable network change, propagate it to the QUIC stack
// right away.
self.call_notify_quic_network_change = None;
self.notify_quic_network_change(is_major);
} else {
// No default route yet (e.g., interface just came up but gateway not
// assigned). Poll with exponential backoff until the gateway appears.
match &mut self.call_notify_quic_network_change {
Some(pending) => {
// Update is_major if this change is more severe.
pending.is_major |= is_major;
}
None => {
self.call_notify_quic_network_change =
Some(PendingNetworkChangeNotify::new(is_major));
}
}
}
}
/// Notifies the QUIC stack of the network change we observed.
///
/// This is decoupled from receiving the network change, because we try to debounce
/// network changes as they often arrive in groups.
fn notify_quic_network_change(&mut self, is_major: bool) {
#[derive(Debug)]
struct Hint {
local_addrs: FxHashSet<IpAddr>,
}
impl NetworkChangeHint for Hint {
fn is_path_recoverable(
&self,
_path_id: noq::PathId,
network_path: noq_proto::FourTuple,
) -> bool {
match MultipathMappedAddr::from(network_path.remote()) {
MultipathMappedAddr::Mixed(_) => {
// This address is only ever used to send an Initial packet to, it
// should never appear as an established path.
error!("A mixed address can not be used for network changes");
false
}
MultipathMappedAddr::Relay(_) => {
// We pretend the relay path is never affected by link changes. The
// relay actor transparently reconnects and the addresses never
// change.
true
}
MultipathMappedAddr::Ip(_) => {
// If we no longer have a valid interface to send from for a local
// IP then it can not be recovered.
match network_path.local_ip() {
Some(local_ip) => self.local_addrs.contains(&local_ip),
None => true,
}
}
MultipathMappedAddr::Custom(_) => {
// Assume it is unrecoverable for now
false
}
}
}
}
let hint = Hint {
#[cfg(not(wasm_browser))]
local_addrs: {
let interfaces = self.local_interfaces_watcher.get();
interfaces
.local_addresses
.regular
.iter()
.chain(interfaces.local_addresses.loopback.iter())
.copied()
.collect()
},
#[cfg(wasm_browser)]
local_addrs: Default::default(),
};
self.endpoint.handle_network_change(Some(Arc::new(hint)));
self.remote_map.on_network_change(is_major);
}
fn handle_relay_map_change(&mut self) {
self.re_stun(UpdateReason::RelayMapChange);
}
fn re_stun(&mut self, why: UpdateReason) {
let state = self.local_interfaces_watcher.get();
self.direct_addr_update_state
.schedule_run(why, state.into());
}
/// Processes an incoming actor message.
///
/// Returns `true` if it was a shutdown.
async fn handle_actor_message(&mut self, msg: ActorMessage) {
match msg {
ActorMessage::NetworkChange => {
self.network_monitor.network_change().await.ok();
}
ActorMessage::RelayMapChange => {
self.handle_relay_map_change();
}
ActorMessage::ResolveRemote(addr, tx) => {
self.remote_map.resolve_remote(addr, tx).await;
}
ActorMessage::AddConnection(remote, conn, tx) => {
self.remote_map.add_connection(remote, conn, tx).await;
}
ActorMessage::DirectAddrRefresh => {
#[cfg(not(wasm_browser))]
{
let (report, _reason) = self.sock.net_report.get();
self.update_direct_addresses(report.as_ref());
}
}
#[cfg(all(test, with_crypto_provider))]
ActorMessage::ForceNetworkChange(is_major) => {
self.handle_network_change(is_major);
}
}
}
/// Updates the direct addresses of this socket.
///
/// Updates the [`DiscoveredDirectAddrs`] of this [`Socket`] with the current set of
/// direct addresses from:
///
/// - The portmapper.
/// - A net_report report.
/// - The local interfaces IP addresses.
/// - User configured addresses.
#[cfg(not(wasm_browser))]
fn update_direct_addresses(&mut self, net_report_report: Option<&net_report::Report>) {
// We only want to have one DirectAddr for each SocketAddr we have. So we store
// this as a map of SocketAddr -> DirectAddrType. At the end we will construct a
// DirectAddr from each entry.
let mut addrs: BTreeMap<SocketAddr, (DirectAddrType, Option<Ipv6AddrFlags>)> =
BTreeMap::new();
// First add PortMapper provided addresses.
let portmap_watcher = self
.direct_addr_update_state
.port_mapper
.watch_external_address();
let maybe_port_mapped = *portmap_watcher.borrow();
if let Some(portmap_ext) = maybe_port_mapped.map(SocketAddr::V4) {
addrs
.entry(portmap_ext)
.or_insert((DirectAddrType::Portmapped, None));
}
// Next add STUN addresses from the net_report report.
if let Some(net_report_report) = net_report_report {
if let Some(global_v4) = net_report_report.global_v4 {
addrs
.entry(global_v4.into())
.or_insert((DirectAddrType::Qad, None));
// If they're behind a hard NAT and are using a fixed
// port locally, assume they might've added a static
// port mapping on their router to the same explicit
// port that we are running with. Worst case it's an invalid candidate mapping.
let port = self.sock.ip_bind_addrs().iter().find_map(|addr| {
if addr.port() != 0 {
Some(addr.port())
} else {
None
}
});
if let Some(port) = port
&& net_report_report
.mapping_varies_by_dest()
.unwrap_or_default()
{
let mut addr = global_v4;
addr.set_port(port);
addrs
.entry(addr.into())
.or_insert((DirectAddrType::Qad4LocalPort, None));
}
}
if let Some(global_v6) = net_report_report.global_v6 {
addrs
.entry(global_v6.into())
.or_insert((DirectAddrType::Qad, None));
}
}
self.collect_local_addresses(&mut addrs);
// Add configured external addresses.
for addr in self.sock.configured_addrs.read().expect("poisoned").iter() {
addrs.entry(*addr).or_insert((DirectAddrType::Config, None));
}
// Finally create and store store all these direct addresses
let stored_addrs = addrs
.into_iter()
.filter_map(|(addr, (typ, flags))| {
// Filter out deprecated IPs
let is_deprecated = flags.map(|f| f.deprecated).unwrap_or(false);
if is_deprecated {
return None;
}
Some(DirectAddr { addr, typ })
})
.collect();
self.sock.store_direct_addresses(stored_addrs);
}
#[cfg(not(wasm_browser))]
fn collect_local_addresses(
&mut self,
addrs: &mut BTreeMap<SocketAddr, (DirectAddrType, Option<Ipv6AddrFlags>)>,
) {
let netmon_state = self.local_interfaces_watcher.get();
// Matches the addresses that have been bound vs the requested ones.
let local_addrs: Vec<(SocketAddr, SocketAddr)> = self
.sock
.ip_bind_addrs()
.iter()
.copied()
.zip(self.sock.ip_local_addrs())
.collect();
// Do we listen on any IPv4 unspecified address?
let has_ipv4_unspecified = local_addrs.iter().find_map(|(_, a)| {
if a.is_ipv4() && a.ip().is_unspecified() {
Some(a.port())
} else {
None
}
});
// Do we listen on any IPv6 unspecified address?
let has_ipv6_unspecified = local_addrs.iter().find_map(|(_, a)| {
if a.is_ipv6() && a.ip().is_unspecified() {
Some(a.port())
} else {
None
}
});
// If a socket is bound to the unspecified address, create SocketAddrs for
// each local IP address by pairing it with the port the socket is bound on.
if local_addrs
.iter()
.any(|(_, local)| local.ip().is_unspecified())
{
let LocalAddresses {
regular: mut ips,
loopback,
} = self.local_interfaces_watcher.get().local_addresses;
if ips.is_empty() && addrs.is_empty() {
// Include loopback addresses only if there are no other interfaces
// or public addresses, this allows testing offline.
ips = loopback;
}
for ip in ips {
let port_if_unspecified = match ip {
IpAddr::V4(_) => has_ipv4_unspecified,
IpAddr::V6(_) => has_ipv6_unspecified,
};
if let Some(port) = port_if_unspecified {
let addr = SocketAddr::new(ip, port);
let flags = find_flags(&netmon_state, ip);
addrs.entry(addr).or_insert((DirectAddrType::Local, flags));
}
}
}
// If a socket is bound to a specific address, add it.
for (bound, local) in local_addrs {
if !bound.ip().is_unspecified() {
let flags = find_flags(&netmon_state, local.ip());
addrs.entry(local).or_insert((DirectAddrType::Local, flags));
}
}
}
fn handle_net_report_report(&mut self, mut report: Option<net_report::Report>) {
if let Some(ref mut r) = report {
self.sock.ipv6_reported.store(r.udp_v6, Ordering::Relaxed);
if r.preferred_relay.is_none()
&& let Some(my_relay) = self.sock.my_relay()
{
r.preferred_relay.replace(my_relay);
}
// Notify all transports
self.transports_network_change.on_network_change(r);
}
#[cfg(not(wasm_browser))]
self.update_direct_addresses(report.as_ref());
}
}
#[cfg(not(wasm_browser))]
fn find_flags(state: &netmon::State, ip: IpAddr) -> Option<Ipv6AddrFlags> {
if ip.is_ipv6() {
state
.interfaces
.values()
.flat_map(|i| i.addrs())
.find_map(|addr| match addr {
IpNet::V4(_) => None,
IpNet::V6 { net, flags, .. } => {
if net.addr() == ip {
Some(flags)
} else {
None
}
}
})
} else {
None
}
}
fn new_re_stun_timer(initial_delay: bool) -> time::Interval {
// Pick a random duration between 20 and 26 seconds (just under 30s,
// a common UDP NAT timeout on Linux,etc)
let mut rng = rand::rng();
let d: Duration = rng.random_range(Duration::from_secs(20)..=Duration::from_secs(26));
if initial_delay {
debug!("scheduling periodic_stun to run in {}s", d.as_secs());
time::interval_at(time::Instant::now() + d, d)
} else {
debug!(
"scheduling periodic_stun to run immediately and in {}s",
d.as_secs()
);
time::interval(d)
}
}
/// The discovered direct addresses of this [`Socket`].
///
/// These are all the [`DirectAddr`]s that this [`Socket`] is aware of for itself.
/// They include all locally bound ones as well as those discovered by other mechanisms like
/// QAD.
#[derive(derive_more::Debug, Clone, Default)]
struct DiscoveredDirectAddrs {
/// The last set of discovered direct addresses.
addrs: Watchable<BTreeSet<DirectAddr>>,
/// The last time the direct addresses were updated, even if there was no change.
///
/// This is only ever None at startup.
updated_at: Arc<RwLock<Option<Instant>>>,
}
impl DiscoveredDirectAddrs {
/// Updates the direct addresses, returns `true` if they changed, `false` if not.
fn update(&self, addrs: BTreeSet<DirectAddr>) -> bool {
*self.updated_at.write().expect("poisoned") = Some(Instant::now());
let updated = self.addrs.set(addrs).is_ok();
if updated {
event!(
target: "iroh::_events::direct_addrs",
Level::DEBUG,
addrs = ?self.addrs.get(),
);
}
updated
}
fn sockaddrs(&self) -> impl Iterator<Item = SocketAddr> {
self.addrs.get().into_iter().map(|da| da.addr)
}
}
/// A *direct address* on which an iroh-endpoint might be contactable.
///
/// Direct addresses are UDP socket addresses on which an iroh endpoint could potentially be
/// contacted. These can come from various sources depending on the network topology of the
/// iroh endpoint, see [`DirectAddrType`] for the several kinds of sources.
///
/// This is essentially a combination of our local addresses combined with any reflexive
/// transport addresses we discovered using QAD.
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct DirectAddr {
/// The address.
pub addr: SocketAddr,
/// The origin of this direct address.
pub typ: DirectAddrType,
}
/// The type of direct address.
///
/// These are the various sources or origins from which an iroh endpoint might have found a
/// possible [`DirectAddr`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
#[non_exhaustive]
pub enum DirectAddrType {
/// Not yet determined..
Unknown,
/// A locally bound socket address.
Local,
/// Public internet address discovered via QAD.
///
/// When possible an iroh endpoint will perform QAD to discover which is the address
/// from which it sends data on the public internet. This can be different from locally
/// bound addresses when the endpoint is on a local network which performs NAT or similar.
Qad,
/// An address assigned by the router using port mapping.
///
/// When possible an iroh endpoint will request a port mapping from the local router to
/// get a publicly routable direct address.
Portmapped,
/// Hard NAT: QAD'ed IPv4 address + local fixed port.
///
/// It is possible to configure iroh to bound to a specific port and independently
/// configure the router to forward this port to the iroh endpoint. This indicates a
/// situation like this, which still uses QAD to discover the public address.
Qad4LocalPort,
/// An address explicitly provided by the user via configuration.
Config,
}
impl Display for DirectAddrType {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
DirectAddrType::Unknown => write!(f, "?"),
DirectAddrType::Local => write!(f, "local"),
DirectAddrType::Qad => write!(f, "qad"),
DirectAddrType::Portmapped => write!(f, "portmap"),
DirectAddrType::Qad4LocalPort => write!(f, "qad4localport"),
DirectAddrType::Config => write!(f, "config"),
}
}
}
#[cfg(all(test, with_crypto_provider))]
mod tests {
use std::{net::SocketAddrV4, sync::Arc, time::Duration};
use data_encoding::HEXLOWER;
use iroh_base::{EndpointAddr, EndpointId, TransportAddr};
use iroh_relay::tls::{CaTlsConfig, default_provider};
use n0_error::{Result, StackResultExt, StdResultExt};
use n0_future::{MergeBounded, StreamExt, time};
use n0_tracing_test::traced_test;
use n0_watcher::Watcher;
use rand::{CryptoRng, Rng, RngExt, SeedableRng};
use tokio_util::task::AbortOnDropHandle;
use tracing::{Instrument, error, info, info_span, instrument};
use super::Options;
use crate::{
Endpoint, SecretKey,
address_lookup::memory::MemoryLookup,
dns::DnsResolver,
endpoint::{QuicTransportConfig, presets},
socket::{
EndpointInner, StaticConfig, TransportConfig,
biased_rtt_path_selector::BiasedRttPathSelector,
mapped_addrs::{EndpointIdMappedAddr, MappedAddr},
},
tls::{self, DEFAULT_MAX_TLS_TICKETS, misc::RustlsTokenKey},
};
const ALPN: &[u8] = b"n0/test/1";
fn default_options(rng: &mut impl CryptoRng) -> Options {
let crypto_provider = default_provider();
let secret_key = SecretKey::from_bytes(&rng.random());
let tls_config = tls::TlsConfig::new(
secret_key.clone(),
DEFAULT_MAX_TLS_TICKETS,
crypto_provider.clone(),
);
let static_config = StaticConfig {
server_config: tls_config.make_server_config(false).unwrap(),
client_config: tls_config.make_client_config(false).unwrap(),
tls_config,
token_key: Arc::new(RustlsTokenKey::new(rng, &crypto_provider).unwrap()),
token_store: Arc::new(noq::TokenMemoryCache::default()),
transport_config: QuicTransportConfig::default(),
};
let server_config = static_config.create_server_config(vec![]);
Options {
transports: vec![
TransportConfig::default_ipv4(),
TransportConfig::default_ipv6(),
],
secret_key,
proxy_url: None,
dns_resolver: DnsResolver::new(),
server_config,
tls_config: CaTlsConfig::default()
.client_config(crypto_provider.clone())
.unwrap(),
#[cfg(any(test, feature = "test-utils"))]
address_lookup_user_data: None,
metrics: Default::default(),
hooks: Default::default(),
path_selector: Arc::new(BiasedRttPathSelector::default()),
portmapper_config: Default::default(),
net_report_config: Default::default(),
static_config,
configured_addrs: Default::default(),
}
}
#[instrument(skip_all, fields(me = %ep.id().fmt_short()))]
async fn echo_receiver(ep: Endpoint, loss: ExpectedLoss) -> Result {
info!("accepting conn");
let conn = ep.accept().await.expect("no conn");
info!("accepting");
let conn = conn.await.context("accepting")?;
info!("accepting bi");
let (mut send_bi, mut recv_bi) = conn.accept_bi().await.std_context("accept bi")?;
info!("reading");
let val = recv_bi
.read_to_end(usize::MAX)
.await
.std_context("read to end")?;
info!("replying");
for chunk in val.chunks(12) {
send_bi.write_all(chunk).await.std_context("write all")?;
}
info!("finishing");
send_bi.finish().std_context("finish")?;
let stats = conn.stats();
info!("stats: {:#?}", stats);
if matches!(loss, ExpectedLoss::AlmostNone) {
for info in conn.paths().iter() {
assert!(
info.stats().lost_packets < 10,
"[receiver] path {:?} should not loose many packets",
info.remote_addr()
);
}
}
conn.closed().await;
info!("closed");
ep.inner()?.noq_endpoint().wait_idle().await;
info!("idle");
Ok(())
}
#[instrument(skip_all, fields(me = %ep.id().fmt_short()))]
async fn echo_sender(
ep: Endpoint,
dest_id: EndpointId,
msg: &[u8],
loss: ExpectedLoss,
) -> Result {
info!("connecting to {}", dest_id.fmt_short());
let dest = EndpointAddr::new(dest_id);
let conn = ep.connect(dest, ALPN).await?;
info!("opening bi");
let (mut send_bi, mut recv_bi) = conn.open_bi().await.std_context("open bi")?;
info!("writing message");
send_bi.write_all(msg).await.std_context("write all")?;
info!("finishing");
send_bi.finish().std_context("finish")?;
info!("reading_to_end");
let val = recv_bi
.read_to_end(usize::MAX)
.await
.std_context("read to end")?;
assert_eq!(
val,
msg,
"[sender] expected {}, got {}",
HEXLOWER.encode(msg),
HEXLOWER.encode(&val)
);
let stats = conn.stats();
info!("stats: {:#?}", stats);
if matches!(loss, ExpectedLoss::AlmostNone) {
for info in conn.paths().iter() {
assert!(
info.stats().lost_packets < 10,
"[sender] path {:?} should not loose many packets",
info.remote_addr()
);
}
}
conn.close(0u32.into(), b"done");
info!("closed");
ep.inner()?.noq_endpoint().wait_idle().await;
info!("idle");
Ok(())
}
#[derive(Debug, Copy, Clone)]
enum ExpectedLoss {
AlmostNone,
YeahSure,
}
/// Runs a roundtrip between the [`echo_sender`] and [`echo_receiver`].
async fn run_roundtrip(
sender: Endpoint,
receiver: Endpoint,
payload: &[u8],
loss: ExpectedLoss,
) -> Result<()> {
tokio::time::timeout(Duration::from_secs(20), async move {
let send_endpoint_id = sender.id();
let recv_endpoint_id = receiver.id();
info!("\nroundtrip: {send_endpoint_id:#} -> {recv_endpoint_id:#}");
let receiver_task = AbortOnDropHandle::new(tokio::spawn(echo_receiver(receiver, loss)));
let sender_res = echo_sender(sender, recv_endpoint_id, payload, loss).await;
let sender_is_err = match sender_res {
Ok(()) => false,
Err(err) => {
error!("[sender] Error:\n{err:#?}");
true
}
};
let receiver_is_err = match receiver_task.await {
Ok(Ok(())) => false,
Ok(Err(err)) => {
error!("[receiver] Error:\n{err:#?}");
true
}
Err(joinerr) => {
if joinerr.is_panic() {
std::panic::resume_unwind(joinerr.into_panic());
} else {
error!("[receiver] Error:\n{joinerr:#?}");
}
true
}
};
if sender_is_err || receiver_is_err {
panic!("Sender or receiver errored");
}
})
.await
.std_context("timeout")?;
Ok(())
}
/// Returns a pair of endpoints with a shared [`MemoryLookup`].
///
/// The endpoints do not use a relay server but can connect to each other via local
/// addresses. Dialing by [`EndpointId`] is possible, and the addresses get updated even if
/// the endpoints rebind.
async fn endpoint_pair() -> (AbortOnDropHandle<()>, Endpoint, Endpoint) {
let address_lookup = MemoryLookup::new();
let ep1 = Endpoint::builder(presets::Minimal)
.alpns(vec![ALPN.to_vec()])
.address_lookup(address_lookup.clone())
.bind()
.await
.unwrap();
let ep2 = Endpoint::builder(presets::Minimal)
.alpns(vec![ALPN.to_vec()])
.address_lookup(address_lookup.clone())
.bind()
.await
.unwrap();
address_lookup.add_endpoint_info(ep1.addr());
address_lookup.add_endpoint_info(ep2.addr());
let ep1_addr_stream = ep1.watch_addr().stream();
let ep2_addr_stream = ep2.watch_addr().stream();
let mut addr_stream = MergeBounded::from_iter([ep1_addr_stream, ep2_addr_stream]);
let task = tokio::spawn(async move {
while let Some(addr) = addr_stream.next().await {
address_lookup.add_endpoint_info(addr);
}
});
(AbortOnDropHandle::new(task), ep1, ep2)
}
#[tokio::test(flavor = "multi_thread")]
#[traced_test]
async fn test_two_devices_roundtrip_noq_small() -> Result {
let (_guard, m1, m2) = endpoint_pair().await;
run_roundtrip(
m1.clone(),
m2.clone(),
b"hello m1",
ExpectedLoss::AlmostNone,
)
.await?;
run_roundtrip(
m2.clone(),
m1.clone(),
b"hello m2",
ExpectedLoss::AlmostNone,
)
.await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
#[traced_test]
async fn test_two_devices_roundtrip_noq_large() -> Result {
let (_guard, m1, m2) = endpoint_pair().await;
let mut data = vec![0u8; 10 * 1024];
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
rng.fill_bytes(&mut data);
run_roundtrip(m1.clone(), m2.clone(), &data, ExpectedLoss::AlmostNone).await?;
run_roundtrip(m2.clone(), m1.clone(), &data, ExpectedLoss::AlmostNone).await?;
Ok(())
}
// EADDRINUSE on the GitHub Android emulator persists past
// `force_network_change()`, so the rebind fails and connect() never
// wakes the connection driver. Passes locally.
#[cfg_attr(
target_os = "android",
ignore = "rebind flakes against the GitHub Android emulator"
)]
#[tokio::test]
#[traced_test]
async fn test_regression_network_change_rebind_wakes_connection_driver() -> Result {
let (_guard, m1, m2) = endpoint_pair().await;
println!("Net change");
m1.inner()?.force_network_change(true).await;
tokio::time::sleep(Duration::from_secs(1)).await; // wait for socket rebinding
let _handle = AbortOnDropHandle::new(tokio::spawn({
let endpoint = m2.clone();
async move {
while let Some(incoming) = endpoint.accept().await {
println!("Incoming first conn!");
let conn = incoming.await.anyerr()?;
conn.closed().await;
}
n0_error::Ok(())
}
}));
println!("first conn!");
let conn = m1.connect(m2.addr(), ALPN).await?;
println!("Closing first conn");
conn.close(0u32.into(), b"bye lolz");
conn.closed().await;
println!("Closed first conn");
Ok(())
}
fn offset(rng: &mut rand_chacha::ChaCha8Rng) -> Duration {
let delay = rng.random_range(1..=5);
Duration::from_millis(delay * 50)
}
/// Same structure as `test_two_devices_roundtrip_noq`, but interrupts regularly
/// with (simulated) network changes.
/// Regular network changes to m1 only.
#[tokio::test(flavor = "multi_thread")]
#[traced_test]
async fn test_two_devices_roundtrip_network_change_only_a() -> Result {
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let (_guard, m1, m2) = endpoint_pair().await;
let _network_change_guard = {
let m1 = m1.clone();
let mut rng = rng.clone();
let task = tokio::spawn(async move {
loop {
info!("[m1] network change");
m1.inner()
.expect("haven't closed the endpoint yet")
.force_network_change(true)
.await;
time::sleep(offset(&mut rng)).await;
}
});
AbortOnDropHandle::new(task)
};
let mut data = vec![0u8; 10 * 1024];
rng.fill_bytes(&mut data);
run_roundtrip(m1.clone(), m2.clone(), &data, ExpectedLoss::YeahSure).await?;
run_roundtrip(m2.clone(), m1.clone(), &data, ExpectedLoss::YeahSure).await?;
Ok(())
}
/// Regular network changes to both m1 and m2.
#[tokio::test(flavor = "multi_thread")]
#[traced_test]
async fn test_two_devices_roundtrip_network_change_a_and_b() -> Result {
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let (_guard, m1, m2) = endpoint_pair().await;
let _network_change_guard = {
let m1 = m1.clone();
let m2 = m2.clone();
let mut rng = rng.clone();
let task = tokio::spawn(async move {
info!("-- [m1] network change");
m1.inner()
.expect("haven't closed the endpoint yet")
.force_network_change(true)
.await;
info!("-- [m2] network change");
m2.inner()
.expect("haven't closed the endpoint yet")
.force_network_change(true)
.await;
time::sleep(offset(&mut rng)).await;
});
AbortOnDropHandle::new(task)
};
let mut data = vec![0u8; 10 * 1024];
rng.fill_bytes(&mut data);
run_roundtrip(m1.clone(), m2.clone(), &data, ExpectedLoss::YeahSure).await?;
run_roundtrip(m2.clone(), m1.clone(), &data, ExpectedLoss::YeahSure).await?;
Ok(())
}
#[tokio::test(flavor = "multi_thread")]
#[traced_test]
async fn test_two_devices_setup_teardown() -> Result {
for i in 0..10 {
info!("-- round {i}");
info!("setting up stack");
let (_guard, m1, m2) = endpoint_pair().await;
info!("closing endpoints");
let sock1 = m1.inner()?;
let sock2 = m2.inner()?;
m1.close().await;
m2.close().await;
assert!(sock1.is_closed());
assert!(sock2.is_closed());
}
Ok(())
}
#[tokio::test]
#[traced_test]
async fn test_direct_addresses() {
let mut rng = rand_chacha::ChaCha8Rng::seed_from_u64(0u64);
let sock = EndpointInner::bind(default_options(&mut rng))
.await
.unwrap();
// See if we can get endpoints.
let eps0 = sock.ip_addrs().get();
info!("{eps0:?}");
assert!(!eps0.is_empty());
// Getting the endpoints again immediately should give the same results.
let eps1 = sock.ip_addrs().get();
info!("{eps1:?}");
assert_eq!(eps0, eps1);
}
/// Creates a new [`noq::Endpoint`] hooked up to a [`Socket`].
///
/// This is without involving [`crate::endpoint::Endpoint`]. The socket will accept
/// connections using [`ALPN`].
///
/// Use [`socket_connect`] to establish connections.
#[instrument(name = "ep", skip_all, fields(me = %secret_key.public().fmt_short()))]
async fn socket_ep(secret_key: SecretKey) -> Result<EndpointInner> {
let crypto_provider = default_provider();
let tls_config = tls::TlsConfig::new(
secret_key.clone(),
DEFAULT_MAX_TLS_TICKETS,
crypto_provider.clone(),
);
let keylog = true;
let static_config = StaticConfig {
server_config: tls_config.make_server_config(keylog).unwrap(),
client_config: tls_config.make_client_config(keylog).unwrap(),
tls_config,
token_key: Arc::new(RustlsTokenKey::new(&mut rand::rng(), &crypto_provider).unwrap()),
token_store: Arc::new(noq::TokenMemoryCache::default()),
transport_config: QuicTransportConfig::default(),
};
let server_config = static_config.create_server_config(vec![ALPN.to_vec()]);
let dns_resolver = DnsResolver::new();
let opts = Options {
transports: vec![
TransportConfig::default_ipv4(),
TransportConfig::default_ipv6(),
],
secret_key: secret_key.clone(),
address_lookup_user_data: None,
dns_resolver,
proxy_url: None,
server_config,
tls_config: CaTlsConfig::default()
.client_config(crypto_provider.clone())
.unwrap(),
metrics: Default::default(),
hooks: Default::default(),
path_selector: Arc::new(BiasedRttPathSelector::default()),
portmapper_config: Default::default(),
net_report_config: Default::default(),
static_config,
configured_addrs: Default::default(),
};
let sock = EndpointInner::bind(opts).await?;
Ok(sock)
}
/// Connects from `ep` returned by [`socket_ep`] to the `endpoint_id`.
///
/// Uses [`ALPN`], `endpoint_id`, must match `addr`.
#[instrument(name = "connect", skip_all, fields(me = %ep_secret_key.public().fmt_short()))]
async fn socket_connect(
ep: noq::Endpoint,
ep_secret_key: SecretKey,
addr: EndpointIdMappedAddr,
endpoint_id: EndpointId,
) -> Result<noq::Connection> {
// Endpoint::connect sets this, do the same to have similar behaviour.
let mut transport_config = noq::TransportConfig::default();
transport_config.server_handshake_migration(true);
transport_config.keep_alive_interval(Some(Duration::from_secs(1)));
socket_connect_with_transport_config(
ep,
ep_secret_key,
addr,
endpoint_id,
Arc::new(transport_config),
)
.await
}
/// Connects from `ep` returned by [`socket_ep`] to the `endpoint_id`.
///
/// This version allows customising the transport config.
///
/// Uses [`ALPN`], `endpoint_id`, must match `addr`.
#[instrument(name = "connect", skip_all, fields(me = %ep_secret_key.public().fmt_short()))]
async fn socket_connect_with_transport_config(
ep: noq::Endpoint,
ep_secret_key: SecretKey,
mapped_addr: EndpointIdMappedAddr,
endpoint_id: EndpointId,
transport_config: Arc<noq::TransportConfig>,
) -> Result<noq::Connection> {
let mut quic_client_config = tls::TlsConfig::new(
ep_secret_key.clone(),
DEFAULT_MAX_TLS_TICKETS,
default_provider(),
)
.make_client_config(true)?;
quic_client_config.set_alpn_protocols(vec![ALPN.to_vec()]);
let mut client_config = noq::ClientConfig::new(Arc::new(quic_client_config));
client_config.transport_config(transport_config);
let connect = ep
.connect_with(
client_config,
mapped_addr.private_socket_addr(),
&tls::name::encode(endpoint_id),
)
.std_context("connect")?;
let connection = connect.await.anyerr()?;
Ok(connection)
}
#[tokio::test]
#[traced_test]
async fn test_try_send_no_send_addr() {
// Regression test: if there is no send_addr we should keep being able to use the
// Endpoint.
let secret_key_1 = SecretKey::from_bytes(&[1u8; 32]);
let secret_key_2 = SecretKey::from_bytes(&[2u8; 32]);
let endpoint_id_2 = secret_key_2.public();
let secret_key_missing_endpoint = SecretKey::from_bytes(&[255u8; 32]);
let endpoint_id_missing_endpoint = secret_key_missing_endpoint.public();
let sock_1 = socket_ep(secret_key_1.clone()).await.unwrap();
// Generate an address not present in the RemoteMap.
let bad_addr = EndpointIdMappedAddr::generate();
// 500ms is rather fast here. Running this locally it should always be the correct
// timeout. If this is too slow however the test will not become flaky as we are
// expecting the timeout, we might just get the timeout for the wrong reason. But
// this speeds up the test.
let res = tokio::time::timeout(
Duration::from_millis(500),
socket_connect(
sock_1.noq_endpoint().clone(),
secret_key_1.clone(),
bad_addr,
endpoint_id_missing_endpoint,
),
)
.await;
assert!(res.is_err(), "expecting timeout");
// Now check we can still create another connection with this endpoint.
let sock_2 = socket_ep(secret_key_2.clone()).await.unwrap();
// This needs an accept task
let accept_task = tokio::spawn({
async fn accept(ep: noq::Endpoint) -> Result<()> {
let incoming = ep.accept().await.std_context("no incoming")?;
let _conn = incoming
.accept()
.std_context("accept")?
.await
.std_context("accepting")?;
// Keep this connection alive for a while
tokio::time::sleep(Duration::from_secs(10)).await;
info!("accept finished");
Ok(())
}
let ep = sock_2.noq_endpoint().clone();
async move {
if let Err(err) = accept(ep).await {
error!("{err:#}");
}
}
.instrument(info_span!("ep2.accept, me = endpoint_id_2.fmt_short()"))
});
let _accept_task = AbortOnDropHandle::new(accept_task);
let addrs = sock_2
.ip_addrs()
.get()
.into_iter()
.map(|x| TransportAddr::Ip(x.addr));
let endpoint_addr_2 = EndpointAddr::from_parts(endpoint_id_2, addrs);
let addr = sock_1
.resolve_remote(endpoint_addr_2)
.await
.unwrap()
.unwrap();
let res = tokio::time::timeout(
Duration::from_secs(10),
socket_connect(
sock_1.noq_endpoint().clone(),
secret_key_1.clone(),
addr,
endpoint_id_2,
),
)
.await
.expect("timeout while connecting");
// aka assert!(res.is_ok()) but with nicer error reporting.
res.unwrap();
// TODO: Now check if we can connect to a repaired ep_3, but we can't modify that
// much internal state for now.
}
#[tokio::test]
#[traced_test]
async fn test_try_send_no_udp_addr_or_relay_url() {
// This specifically tests the `if udp_addr.is_none() && relay_url.is_none()`
// behaviour of Socket::try_send.
let secret_key_1 = SecretKey::from_bytes(&[1u8; 32]);
let secret_key_2 = SecretKey::from_bytes(&[2u8; 32]);
let endpoint_id_2 = secret_key_2.public();
let sock_1 = socket_ep(secret_key_1.clone()).await.unwrap();
let sock_2 = socket_ep(secret_key_2.clone()).await.unwrap();
let ep_2 = sock_2.noq_endpoint().clone();
// We need a task to accept the connection.
let accept_task = tokio::spawn({
async fn accept(ep: noq::Endpoint) -> Result<()> {
let incoming = ep.accept().await.std_context("no incoming")?;
let conn = incoming
.accept()
.std_context("accept")?
.await
.std_context("connecting")?;
let mut stream = conn.accept_uni().await.std_context("accept uni")?;
stream
.read_to_end(1 << 16)
.await
.std_context("read to end")?;
info!("accept finished");
Ok(())
}
async move {
if let Err(err) = accept(ep_2).await {
error!("{err:#}");
}
}
.instrument(info_span!("ep2.accept", me = %endpoint_id_2.fmt_short()))
});
let _accept_task = AbortOnDropHandle::new(accept_task);
// Add an entry in the RemoteMap of ep_1 with an invalid socket address
let empty_addr_2 = EndpointAddr::from_parts(
endpoint_id_2,
[TransportAddr::Ip(
// Reserved IP range for documentation (unreachable)
SocketAddrV4::new([192, 0, 2, 1].into(), 12345).into(),
)],
);
let addr_2 = sock_1.resolve_remote(empty_addr_2).await.unwrap().unwrap();
// Set a low max_idle_timeout so noq gives up on this quickly and our test does
// not take forever. You need to check the log output to verify this is really
// triggering the correct error.
// In test_try_send_no_send_addr() above you may have noticed we used
// tokio::time::timeout() on the connection attempt instead. Here however we want
// Noq itself to have fully given up on the connection attempt because we will
// later connect to **the same** endpoint. If Noq did not give up on the connection
// we'd close it on drop, and the retransmits of the close packets would interfere
// with the next handshake, closing it during the handshake. This makes the test a
// little slower though.
let mut transport_config = noq::TransportConfig::default();
transport_config.server_handshake_migration(true);
transport_config.max_idle_timeout(Some(Duration::from_millis(200).try_into().unwrap()));
let res = socket_connect_with_transport_config(
sock_1.noq_endpoint().clone(),
secret_key_1.clone(),
addr_2,
endpoint_id_2,
Arc::new(transport_config),
)
.await;
assert!(res.is_err(), "expected timeout");
info!("first connect timed out as expected");
// Provide correct addressing information
let correct_addr_2 = EndpointAddr::from_parts(
endpoint_id_2,
sock_2
.ip_addrs()
.get()
.into_iter()
.map(|x| TransportAddr::Ip(x.addr)),
);
let addr_2a = sock_1
.resolve_remote(correct_addr_2)
.await
.unwrap()
.unwrap();
assert_eq!(addr_2, addr_2a);
// We can now connect
tokio::time::timeout(Duration::from_secs(10), async move {
info!("establishing new connection");
let conn = socket_connect(
sock_1.noq_endpoint().clone(),
secret_key_1.clone(),
addr_2,
endpoint_id_2,
)
.await
.unwrap();
info!("have connection");
let mut stream = conn.open_uni().await.unwrap();
stream.write_all(b"hello").await.unwrap();
stream.finish().unwrap();
stream.stopped().await.unwrap();
info!("finished stream");
})
.await
.expect("connection timed out");
// TODO: could remove the addresses again, send, add it back and see it recover.
// But we don't have that much private access to the RemoteMap. This will do for now.
}
}
Reference in new issue View Git Blame Copy Permalink