From 885579b266820141117e52fad580325ca2d93e91 Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 15 Jul 2026 16:58:32 +0200 Subject: [PATCH 01/10] feat(netwatch): add a ConfigureSocket hook to BindOptions SO_MARK covers Linux, but it has no equivalent on other platforms, and the option that replaces it there (IP_BOUND_IF / IPV6_BOUND_IF on Apple) pins the socket to an interface, so it needs the current default-route interface rather than a constant, and has to be re-resolved whenever the socket rebinds. Rather than grow a per-platform option for each of these, hand the caller the socket. The hook runs before bind and again on every rebind, so a hook that reads current network state re-reads it on each network change. An error from the hook fails the bind: a socket that was meant to be kept out of a tunnel and silently wasn't is worse than one that failed to bind. The socket's Domain is passed in because it cannot be read back off the socket portably (SO_DOMAIN is Linux-only) and the option to set is family-specific. --- netwatch/src/lib.rs | 2 +- netwatch/src/udp.rs | 162 +++++++++++++++++++++++++++++++++++++++++--- 2 files changed, 154 insertions(+), 10 deletions(-) diff --git a/netwatch/src/lib.rs b/netwatch/src/lib.rs index 99a0596d..233068fb 100644 --- a/netwatch/src/lib.rs +++ b/netwatch/src/lib.rs @@ -12,4 +12,4 @@ mod udp; pub use self::ip_family::IpFamily; #[cfg(not(wasm_browser))] -pub use self::udp::{BindOptions, UdpSender, UdpSocket}; +pub use self::udp::{BindOptions, ConfigureSocket, UdpSender, UdpSocket}; diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index 184c855b..bbae1636 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -10,11 +10,38 @@ use std::{ use atomic_waker::AtomicWaker; use noq_udp::Transmit; +use socket2::SockRef; use tokio::io::Interest; use tracing::{debug, trace, warn}; use super::IpFamily; +/// A hook run on the socket after it is created and before it is bound. +/// +/// This is the escape hatch for socket options this crate does not model itself. The +/// motivating ones decide how the socket's egress is routed: a VPN that points the +/// default route at its own tunnel device has to keep its own underlay traffic off +/// that route, or the transport is routed into the tunnel it is carrying. [`SO_MARK`] +/// covers that on Linux, but it has no equivalent elsewhere: Apple platforms need +/// `IP_BOUND_IF` / `IPV6_BOUND_IF` instead, which pins the socket to an interface and +/// so needs the current default-route interface, not a constant. +/// +/// The hook is applied again on every rebind, so a hook that resolves something about +/// the current network re-resolves it on each network change instead of pinning a +/// value that goes stale. +/// +/// An error from the hook fails the bind. A socket that was meant to be kept out of a +/// tunnel and silently wasn't is worse than one that failed to bind. +/// +/// The socket's [`Domain`] is passed in because it cannot be read back off the socket +/// portably (`SO_DOMAIN` is Linux-only) and the option to set is usually +/// family-specific. +/// +/// [`SO_MARK`]: BindOptions::set_mark +/// [`Domain`]: socket2::Domain +pub type ConfigureSocket = + Arc, socket2::Domain) -> io::Result<()> + Send + Sync + 'static>; + /// Wrapper around a tokio UDP socket. #[derive(Debug)] pub struct UdpSocket { @@ -33,9 +60,10 @@ const SOCKET_BUFFER_SIZE: usize = 7 << 20; /// /// Used by [`UdpSocket::bind_with`]. The default options match what the other /// `bind_*` constructors use. -#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +#[derive(Default, Clone)] pub struct BindOptions { mark: Option, + configure: Option, } impl BindOptions { @@ -56,6 +84,27 @@ impl BindOptions { self.mark = mark; self } + + /// Sets a [`ConfigureSocket`] hook, run on the socket before it is bound and again + /// on every rebind. + /// + /// Use this for socket options that are platform-specific or that depend on the + /// current network state, which [`set_mark`] cannot express. + /// + /// [`set_mark`]: Self::set_mark + pub fn configure_socket(mut self, configure: ConfigureSocket) -> Self { + self.configure = Some(configure); + self + } +} + +impl std::fmt::Debug for BindOptions { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("BindOptions") + .field("mark", &self.mark) + .field("configure", &self.configure.as_ref().map(|_| "")) + .finish() + } } impl UdpSocket { @@ -98,7 +147,7 @@ impl UdpSocket { /// Bind to any provided [`SocketAddr`], using the given [`BindOptions`]. pub fn bind_with(addr: impl Into, opts: BindOptions) -> io::Result { - let socket = SocketState::bind(addr.into(), opts.mark)?; + let socket = SocketState::bind(addr.into(), opts.mark, opts.configure)?; Ok(UdpSocket { socket: RwLock::new(socket), @@ -760,7 +809,6 @@ impl Future for SendToFut<'_, '_> { } } -#[derive(Debug)] enum SocketState { Connected { socket: tokio::net::UdpSocket, @@ -769,18 +817,42 @@ enum SocketState { addr: SocketAddr, /// The fwmark to (re)apply to the socket, if any. mark: Option, + /// The hook to (re)run on the socket, if any. + configure: Option, }, Closed { /// The addr to rebind to when recovering. addr: SocketAddr, /// The fwmark to reapply when rebinding, if any. mark: Option, + /// The hook to rerun when rebinding, if any. + configure: Option, last_max_gso_segments: NonZeroUsize, last_gro_segments: NonZeroUsize, last_may_fragment: bool, }, } +impl std::fmt::Debug for SocketState { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + Self::Connected { + socket, addr, mark, .. + } => f + .debug_struct("SocketState::Connected") + .field("socket", socket) + .field("addr", addr) + .field("mark", mark) + .finish_non_exhaustive(), + Self::Closed { addr, mark, .. } => f + .debug_struct("SocketState::Closed") + .field("addr", addr) + .field("mark", mark) + .finish_non_exhaustive(), + } + } +} + impl SocketState { fn try_get_connected(&self) -> io::Result<(&tokio::net::UdpSocket, &noq_udp::UdpSocketState)> { match self { @@ -789,6 +861,7 @@ impl SocketState { state, addr: _, mark: _, + configure: _, } => Ok((socket, state)), Self::Closed { .. } => { warn!("socket closed"); @@ -797,7 +870,11 @@ impl SocketState { } } - fn bind(addr: SocketAddr, mark: Option) -> io::Result { + fn bind( + addr: SocketAddr, + mark: Option, + configure: Option, + ) -> io::Result { let network = IpFamily::from(addr.ip()); let socket = socket2::Socket::new( network.into(), @@ -832,6 +909,11 @@ impl SocketState { #[cfg(not(any(target_os = "linux", target_os = "android")))] let _ = mark; + // Run the caller's hook. A failure here fails the bind: see [`ConfigureSocket`]. + if let Some(configure) = &configure { + configure(SockRef::from(&socket), network.into())?; + } + // Binding must happen before calling noq, otherwise `local_addr` // is not yet available on all OSes. socket.bind(&addr.into())?; @@ -861,13 +943,24 @@ impl SocketState { state: socket_state, addr: local_addr, mark, + configure, }) } fn rebind(&mut self) -> io::Result<()> { - let (addr, mark) = match self { - Self::Connected { addr, mark, .. } => (*addr, *mark), - Self::Closed { addr, mark, .. } => (*addr, *mark), + let (addr, mark, configure) = match self { + Self::Connected { + addr, + mark, + configure, + .. + } => (*addr, *mark, configure.clone()), + Self::Closed { + addr, + mark, + configure, + .. + } => (*addr, *mark, configure.clone()), }; debug!("rebinding {}", addr); @@ -877,13 +970,14 @@ impl SocketState { *self = SocketState::Closed { addr, mark, + configure: configure.clone(), last_max_gso_segments: state.max_gso_segments(), last_gro_segments: state.gro_segments(), last_may_fragment: state.may_fragment(), }; } - match Self::bind(addr, mark) { + match Self::bind(addr, mark, configure) { Ok(new_state) => { *self = new_state; Ok(()) @@ -903,11 +997,16 @@ impl SocketState { fn close(&mut self) -> Option<(tokio::net::UdpSocket, noq_udp::UdpSocketState)> { match self { Self::Connected { - state, addr, mark, .. + state, + addr, + mark, + configure, + .. } => { let s = SocketState::Closed { addr: *addr, mark: *mark, + configure: configure.clone(), last_max_gso_segments: state.max_gso_segments(), last_gro_segments: state.gro_segments(), last_may_fragment: state.may_fragment(), @@ -1119,6 +1218,11 @@ impl Future for SendFutNoq<'_, '_> { #[cfg(test)] mod tests { + use std::{ + net::Ipv4Addr, + sync::atomic::{AtomicUsize, Ordering}, + }; + use testresult::TestResult; use super::*; @@ -1180,6 +1284,46 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_configure_socket_runs_on_bind_and_rebind() -> TestResult { + let calls = Arc::new(AtomicUsize::new(0)); + let seen = calls.clone(); + let opts = BindOptions::new().configure_socket(Arc::new( + move |sock: SockRef<'_>, domain: socket2::Domain| { + // The hook gets a real socket, and is told which family it is. + assert_eq!(domain, socket2::Domain::IPV4); + sock.set_reuse_address(true)?; + seen.fetch_add(1, Ordering::SeqCst); + Ok(()) + }, + )); + + let socket = UdpSocket::bind_with((Ipv4Addr::LOCALHOST, 0), opts)?; + assert_eq!(calls.load(Ordering::SeqCst), 1, "hook did not run on bind"); + + socket.rebind()?; + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "hook did not run again on rebind" + ); + + Ok(()) + } + + #[tokio::test] + async fn test_configure_socket_error_fails_the_bind() -> TestResult { + let opts = BindOptions::new() + .configure_socket(Arc::new(|_: SockRef<'_>, _| Err(io::Error::other("nope")))); + + assert!( + UdpSocket::bind_with((Ipv4Addr::LOCALHOST, 0), opts).is_err(), + "a failing hook must fail the bind" + ); + + Ok(()) + } + #[tokio::test] async fn test_udp_mark_broken() -> TestResult { let socket_a = UdpSocket::bind_local(IpFamily::V4, 0)?; From 00923f3ed52b58ab37be40ddd4c29d3cdc51ad02 Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 15 Jul 2026 17:25:31 +0200 Subject: [PATCH 02/10] refactor(netwatch): make the bind hook a SocketConfigurator trait A named trait documents the contract (reruns per rebind, an error fails the bind, the Domain parameter) better than a bare Arc alias, and lets a caller implement it on a stateful type. Closures still work via a blanket impl. The stored form is a private newtype with an opaque Debug, so BindOptions and SocketState go back to plain derives. --- netwatch/src/lib.rs | 2 +- netwatch/src/udp.rs | 128 +++++++++++++++++++++++--------------------- 2 files changed, 68 insertions(+), 62 deletions(-) diff --git a/netwatch/src/lib.rs b/netwatch/src/lib.rs index 7f60daf8..5bbfd43a 100644 --- a/netwatch/src/lib.rs +++ b/netwatch/src/lib.rs @@ -9,4 +9,4 @@ mod udp; pub use self::ip_family::IpFamily; #[cfg(not(wasm_browser))] -pub use self::udp::{BindOptions, ConfigureSocket, UdpSender, UdpSocket}; +pub use self::udp::{BindOptions, SocketConfigurator, UdpSender, UdpSocket}; diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index bbae1636..93265a96 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -16,7 +16,7 @@ use tracing::{debug, trace, warn}; use super::IpFamily; -/// A hook run on the socket after it is created and before it is bound. +/// Configures each socket a [`UdpSocket`] creates, right before it is bound. /// /// This is the escape hatch for socket options this crate does not model itself. The /// motivating ones decide how the socket's egress is routed: a VPN that points the @@ -26,21 +26,53 @@ use super::IpFamily; /// `IP_BOUND_IF` / `IPV6_BOUND_IF` instead, which pins the socket to an interface and /// so needs the current default-route interface, not a constant. /// -/// The hook is applied again on every rebind, so a hook that resolves something about -/// the current network re-resolves it on each network change instead of pinning a -/// value that goes stale. -/// -/// An error from the hook fails the bind. A socket that was meant to be kept out of a -/// tunnel and silently wasn't is worse than one that failed to bind. -/// -/// The socket's [`Domain`] is passed in because it cannot be read back off the socket -/// portably (`SO_DOMAIN` is Linux-only) and the option to set is usually -/// family-specific. +/// Any `Fn(SockRef<'_>, socket2::Domain) -> io::Result<()> + Send + Sync + 'static` +/// closure implements this trait, so simple configurators need no named type. /// /// [`SO_MARK`]: BindOptions::set_mark -/// [`Domain`]: socket2::Domain -pub type ConfigureSocket = - Arc, socket2::Domain) -> io::Result<()> + Send + Sync + 'static>; +pub trait SocketConfigurator: Send + Sync + 'static { + /// Called on the socket after it is created and before it is bound, and called + /// again on every rebind, so a configurator that resolves something about the + /// current network re-resolves it on each network change instead of pinning a + /// value that goes stale. + /// + /// An error fails the bind. A socket that was meant to be kept out of a tunnel + /// and silently wasn't is worse than one that failed to bind. + /// + /// The socket's [`Domain`] is passed in because it cannot be read back off the + /// socket portably (`SO_DOMAIN` is Linux-only) and the option to set is usually + /// family-specific. + /// + /// [`Domain`]: socket2::Domain + fn configure(&self, socket: SockRef<'_>, domain: socket2::Domain) -> io::Result<()>; +} + +impl SocketConfigurator for F +where + F: Fn(SockRef<'_>, socket2::Domain) -> io::Result<()> + Send + Sync + 'static, +{ + fn configure(&self, socket: SockRef<'_>, domain: socket2::Domain) -> io::Result<()> { + self(socket, domain) + } +} + +impl SocketConfigurator for Arc { + fn configure(&self, socket: SockRef<'_>, domain: socket2::Domain) -> io::Result<()> { + (**self).configure(socket, domain) + } +} + +/// A [`SocketConfigurator`] as stored: type-erased, cloneable, and opaque to `Debug` +/// (a bare trait object would otherwise force manual `Debug` impls on everything +/// that holds it). +#[derive(Clone)] +struct Configurator(Arc); + +impl std::fmt::Debug for Configurator { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("Configurator(..)") + } +} /// Wrapper around a tokio UDP socket. #[derive(Debug)] @@ -60,10 +92,10 @@ const SOCKET_BUFFER_SIZE: usize = 7 << 20; /// /// Used by [`UdpSocket::bind_with`]. The default options match what the other /// `bind_*` constructors use. -#[derive(Default, Clone)] +#[derive(Debug, Default, Clone)] pub struct BindOptions { mark: Option, - configure: Option, + configure: Option, } impl BindOptions { @@ -85,28 +117,19 @@ impl BindOptions { self } - /// Sets a [`ConfigureSocket`] hook, run on the socket before it is bound and again + /// Sets a [`SocketConfigurator`], run on the socket before it is bound and again /// on every rebind. /// /// Use this for socket options that are platform-specific or that depend on the /// current network state, which [`set_mark`] cannot express. /// /// [`set_mark`]: Self::set_mark - pub fn configure_socket(mut self, configure: ConfigureSocket) -> Self { - self.configure = Some(configure); + pub fn configure_socket(mut self, configurator: impl SocketConfigurator) -> Self { + self.configure = Some(Configurator(Arc::new(configurator))); self } } -impl std::fmt::Debug for BindOptions { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.debug_struct("BindOptions") - .field("mark", &self.mark) - .field("configure", &self.configure.as_ref().map(|_| "")) - .finish() - } -} - impl UdpSocket { /// Bind only Ipv4 on any interface. pub fn bind_v4(port: u16) -> io::Result { @@ -809,6 +832,7 @@ impl Future for SendToFut<'_, '_> { } } +#[derive(Debug)] enum SocketState { Connected { socket: tokio::net::UdpSocket, @@ -817,42 +841,22 @@ enum SocketState { addr: SocketAddr, /// The fwmark to (re)apply to the socket, if any. mark: Option, - /// The hook to (re)run on the socket, if any. - configure: Option, + /// The configurator to (re)run on the socket, if any. + configure: Option, }, Closed { /// The addr to rebind to when recovering. addr: SocketAddr, /// The fwmark to reapply when rebinding, if any. mark: Option, - /// The hook to rerun when rebinding, if any. - configure: Option, + /// The configurator to rerun when rebinding, if any. + configure: Option, last_max_gso_segments: NonZeroUsize, last_gro_segments: NonZeroUsize, last_may_fragment: bool, }, } -impl std::fmt::Debug for SocketState { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - match self { - Self::Connected { - socket, addr, mark, .. - } => f - .debug_struct("SocketState::Connected") - .field("socket", socket) - .field("addr", addr) - .field("mark", mark) - .finish_non_exhaustive(), - Self::Closed { addr, mark, .. } => f - .debug_struct("SocketState::Closed") - .field("addr", addr) - .field("mark", mark) - .finish_non_exhaustive(), - } - } -} - impl SocketState { fn try_get_connected(&self) -> io::Result<(&tokio::net::UdpSocket, &noq_udp::UdpSocketState)> { match self { @@ -873,7 +877,7 @@ impl SocketState { fn bind( addr: SocketAddr, mark: Option, - configure: Option, + configure: Option, ) -> io::Result { let network = IpFamily::from(addr.ip()); let socket = socket2::Socket::new( @@ -909,9 +913,10 @@ impl SocketState { #[cfg(not(any(target_os = "linux", target_os = "android")))] let _ = mark; - // Run the caller's hook. A failure here fails the bind: see [`ConfigureSocket`]. + // Run the caller's configurator. A failure here fails the bind: see + // [`SocketConfigurator::configure`]. if let Some(configure) = &configure { - configure(SockRef::from(&socket), network.into())?; + configure.0.configure(SockRef::from(&socket), network.into())?; } // Binding must happen before calling noq, otherwise `local_addr` @@ -1288,15 +1293,15 @@ mod tests { async fn test_configure_socket_runs_on_bind_and_rebind() -> TestResult { let calls = Arc::new(AtomicUsize::new(0)); let seen = calls.clone(); - let opts = BindOptions::new().configure_socket(Arc::new( + let opts = BindOptions::new().configure_socket( move |sock: SockRef<'_>, domain: socket2::Domain| { - // The hook gets a real socket, and is told which family it is. + // The configurator gets a real socket, and is told which family it is. assert_eq!(domain, socket2::Domain::IPV4); sock.set_reuse_address(true)?; seen.fetch_add(1, Ordering::SeqCst); Ok(()) }, - )); + ); let socket = UdpSocket::bind_with((Ipv4Addr::LOCALHOST, 0), opts)?; assert_eq!(calls.load(Ordering::SeqCst), 1, "hook did not run on bind"); @@ -1313,12 +1318,13 @@ mod tests { #[tokio::test] async fn test_configure_socket_error_fails_the_bind() -> TestResult { - let opts = BindOptions::new() - .configure_socket(Arc::new(|_: SockRef<'_>, _| Err(io::Error::other("nope")))); + let opts = BindOptions::new().configure_socket(|_: SockRef<'_>, _: socket2::Domain| { + Err(io::Error::other("nope")) + }); assert!( UdpSocket::bind_with((Ipv4Addr::LOCALHOST, 0), opts).is_err(), - "a failing hook must fail the bind" + "a failing configurator must fail the bind" ); Ok(()) From 7d77e11b36d26e5a61cefa87117fdae81912fbcb Mon Sep 17 00:00:00 2001 From: Dario Date: Wed, 15 Jul 2026 19:09:09 +0200 Subject: [PATCH 03/10] refactor(netwatch): skip the configurator in Debug via derive_more Drops the newtype and its hand-written Debug impl: the field is stored as a plain Arc and BindOptions/SocketState derive Debug through derive_more with #[debug(skip)], which the crate already uses elsewhere. --- netwatch/src/udp.rs | 31 +++++++++++-------------------- 1 file changed, 11 insertions(+), 20 deletions(-) diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index 93265a96..b875f5c3 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -62,18 +62,6 @@ impl SocketConfigurator for Arc { } } -/// A [`SocketConfigurator`] as stored: type-erased, cloneable, and opaque to `Debug` -/// (a bare trait object would otherwise force manual `Debug` impls on everything -/// that holds it). -#[derive(Clone)] -struct Configurator(Arc); - -impl std::fmt::Debug for Configurator { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("Configurator(..)") - } -} - /// Wrapper around a tokio UDP socket. #[derive(Debug)] pub struct UdpSocket { @@ -92,10 +80,11 @@ const SOCKET_BUFFER_SIZE: usize = 7 << 20; /// /// Used by [`UdpSocket::bind_with`]. The default options match what the other /// `bind_*` constructors use. -#[derive(Debug, Default, Clone)] +#[derive(derive_more::Debug, Default, Clone)] pub struct BindOptions { mark: Option, - configure: Option, + #[debug(skip)] + configure: Option>, } impl BindOptions { @@ -125,7 +114,7 @@ impl BindOptions { /// /// [`set_mark`]: Self::set_mark pub fn configure_socket(mut self, configurator: impl SocketConfigurator) -> Self { - self.configure = Some(Configurator(Arc::new(configurator))); + self.configure = Some(Arc::new(configurator)); self } } @@ -832,7 +821,7 @@ impl Future for SendToFut<'_, '_> { } } -#[derive(Debug)] +#[derive(derive_more::Debug)] enum SocketState { Connected { socket: tokio::net::UdpSocket, @@ -842,7 +831,8 @@ enum SocketState { /// The fwmark to (re)apply to the socket, if any. mark: Option, /// The configurator to (re)run on the socket, if any. - configure: Option, + #[debug(skip)] + configure: Option>, }, Closed { /// The addr to rebind to when recovering. @@ -850,7 +840,8 @@ enum SocketState { /// The fwmark to reapply when rebinding, if any. mark: Option, /// The configurator to rerun when rebinding, if any. - configure: Option, + #[debug(skip)] + configure: Option>, last_max_gso_segments: NonZeroUsize, last_gro_segments: NonZeroUsize, last_may_fragment: bool, @@ -877,7 +868,7 @@ impl SocketState { fn bind( addr: SocketAddr, mark: Option, - configure: Option, + configure: Option>, ) -> io::Result { let network = IpFamily::from(addr.ip()); let socket = socket2::Socket::new( @@ -916,7 +907,7 @@ impl SocketState { // Run the caller's configurator. A failure here fails the bind: see // [`SocketConfigurator::configure`]. if let Some(configure) = &configure { - configure.0.configure(SockRef::from(&socket), network.into())?; + configure.configure(SockRef::from(&socket), network.into())?; } // Binding must happen before calling noq, otherwise `local_addr` From 1dec793f1d575b8132781b14459b354d84290aaf Mon Sep 17 00:00:00 2001 From: Dario Date: Thu, 16 Jul 2026 15:38:18 +0200 Subject: [PATCH 04/10] refactor(netwatch): shorten the SocketConfigurator docs and drop BindOptions::set_mark A mark is a one-line configurator, so the dedicated option is redundant. Reverts the SO_MARK plumbing from #179 (unreleased) in favor of the general hook. --- netwatch/src/udp.rs | 94 +++++++-------------------------------------- 1 file changed, 14 insertions(+), 80 deletions(-) diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index b875f5c3..bc61fba5 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -18,32 +18,14 @@ use super::IpFamily; /// Configures each socket a [`UdpSocket`] creates, right before it is bound. /// -/// This is the escape hatch for socket options this crate does not model itself. The -/// motivating ones decide how the socket's egress is routed: a VPN that points the -/// default route at its own tunnel device has to keep its own underlay traffic off -/// that route, or the transport is routed into the tunnel it is carrying. [`SO_MARK`] -/// covers that on Linux, but it has no equivalent elsewhere: Apple platforms need -/// `IP_BOUND_IF` / `IPV6_BOUND_IF` instead, which pins the socket to an interface and -/// so needs the current default-route interface, not a constant. -/// -/// Any `Fn(SockRef<'_>, socket2::Domain) -> io::Result<()> + Send + Sync + 'static` -/// closure implements this trait, so simple configurators need no named type. -/// -/// [`SO_MARK`]: BindOptions::set_mark +/// An escape hatch for socket options this crate does not model itself, e.g. +/// `SO_MARK` on Linux or `IP_BOUND_IF` on Apple platforms. Any matching `Fn` +/// closure implements it. pub trait SocketConfigurator: Send + Sync + 'static { - /// Called on the socket after it is created and before it is bound, and called - /// again on every rebind, so a configurator that resolves something about the - /// current network re-resolves it on each network change instead of pinning a - /// value that goes stale. - /// - /// An error fails the bind. A socket that was meant to be kept out of a tunnel - /// and silently wasn't is worse than one that failed to bind. + /// Called before the socket is bound, and again on every rebind. /// - /// The socket's [`Domain`] is passed in because it cannot be read back off the - /// socket portably (`SO_DOMAIN` is Linux-only) and the option to set is usually - /// family-specific. - /// - /// [`Domain`]: socket2::Domain + /// An error fails the bind. `domain` is passed in because it cannot be read + /// back off the socket portably (`SO_DOMAIN` is Linux-only). fn configure(&self, socket: SockRef<'_>, domain: socket2::Domain) -> io::Result<()>; } @@ -82,7 +64,6 @@ const SOCKET_BUFFER_SIZE: usize = 7 << 20; /// `bind_*` constructors use. #[derive(derive_more::Debug, Default, Clone)] pub struct BindOptions { - mark: Option, #[debug(skip)] configure: Option>, } @@ -93,26 +74,8 @@ impl BindOptions { Self::default() } - /// Sets the fwmark to apply to the socket, or `None` to leave it unmarked. - /// - /// The mark is applied with `SO_MARK` and is reapplied whenever the socket is - /// rebound, so the caller can policy-route the socket's egress, for example - /// around a full-tunnel default route. - /// - /// This only has an effect on Linux and Android. On every other platform the - /// mark is ignored. - pub fn set_mark(mut self, mark: Option) -> Self { - self.mark = mark; - self - } - /// Sets a [`SocketConfigurator`], run on the socket before it is bound and again /// on every rebind. - /// - /// Use this for socket options that are platform-specific or that depend on the - /// current network state, which [`set_mark`] cannot express. - /// - /// [`set_mark`]: Self::set_mark pub fn configure_socket(mut self, configurator: impl SocketConfigurator) -> Self { self.configure = Some(Arc::new(configurator)); self @@ -159,7 +122,7 @@ impl UdpSocket { /// Bind to any provided [`SocketAddr`], using the given [`BindOptions`]. pub fn bind_with(addr: impl Into, opts: BindOptions) -> io::Result { - let socket = SocketState::bind(addr.into(), opts.mark, opts.configure)?; + let socket = SocketState::bind(addr.into(), opts.configure)?; Ok(UdpSocket { socket: RwLock::new(socket), @@ -828,8 +791,6 @@ enum SocketState { state: noq_udp::UdpSocketState, /// The addr we are binding to. addr: SocketAddr, - /// The fwmark to (re)apply to the socket, if any. - mark: Option, /// The configurator to (re)run on the socket, if any. #[debug(skip)] configure: Option>, @@ -837,8 +798,6 @@ enum SocketState { Closed { /// The addr to rebind to when recovering. addr: SocketAddr, - /// The fwmark to reapply when rebinding, if any. - mark: Option, /// The configurator to rerun when rebinding, if any. #[debug(skip)] configure: Option>, @@ -855,7 +814,6 @@ impl SocketState { socket, state, addr: _, - mark: _, configure: _, } => Ok((socket, state)), Self::Closed { .. } => { @@ -865,11 +823,7 @@ impl SocketState { } } - fn bind( - addr: SocketAddr, - mark: Option, - configure: Option>, - ) -> io::Result { + fn bind(addr: SocketAddr, configure: Option>) -> io::Result { let network = IpFamily::from(addr.ip()); let socket = socket2::Socket::new( network.into(), @@ -894,16 +848,6 @@ impl SocketState { socket.set_only_v6(true)?; } - // Apply the fwmark, if set. Only supported on linux and android. - #[cfg(any(target_os = "linux", target_os = "android"))] - if let Some(mark) = mark - && let Err(err) = socket.set_mark(mark) - { - warn!("failed to set SO_MARK {} on udp socket: {:?}", mark, err); - } - #[cfg(not(any(target_os = "linux", target_os = "android")))] - let _ = mark; - // Run the caller's configurator. A failure here fails the bind: see // [`SocketConfigurator::configure`]. if let Some(configure) = &configure { @@ -938,25 +882,18 @@ impl SocketState { socket, state: socket_state, addr: local_addr, - mark, configure, }) } fn rebind(&mut self) -> io::Result<()> { - let (addr, mark, configure) = match self { + let (addr, configure) = match self { Self::Connected { - addr, - mark, - configure, - .. - } => (*addr, *mark, configure.clone()), + addr, configure, .. + } => (*addr, configure.clone()), Self::Closed { - addr, - mark, - configure, - .. - } => (*addr, *mark, configure.clone()), + addr, configure, .. + } => (*addr, configure.clone()), }; debug!("rebinding {}", addr); @@ -965,7 +902,6 @@ impl SocketState { if let Self::Connected { state, .. } = self { *self = SocketState::Closed { addr, - mark, configure: configure.clone(), last_max_gso_segments: state.max_gso_segments(), last_gro_segments: state.gro_segments(), @@ -973,7 +909,7 @@ impl SocketState { }; } - match Self::bind(addr, mark, configure) { + match Self::bind(addr, configure) { Ok(new_state) => { *self = new_state; Ok(()) @@ -995,13 +931,11 @@ impl SocketState { Self::Connected { state, addr, - mark, configure, .. } => { let s = SocketState::Closed { addr: *addr, - mark: *mark, configure: configure.clone(), last_max_gso_segments: state.max_gso_segments(), last_gro_segments: state.gro_segments(), From 4e8a09b495d1adafe0cff03836f9d7a3e8b115e7 Mon Sep 17 00:00:00 2001 From: Dario Date: Thu, 16 Jul 2026 15:41:10 +0200 Subject: [PATCH 05/10] docs(netwatch): show a SO_MARK closure example on SocketConfigurator --- netwatch/src/udp.rs | 15 ++++++++++++++- 1 file changed, 14 insertions(+), 1 deletion(-) diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index bc61fba5..ddabbc82 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -20,7 +20,20 @@ use super::IpFamily; /// /// An escape hatch for socket options this crate does not model itself, e.g. /// `SO_MARK` on Linux or `IP_BOUND_IF` on Apple platforms. Any matching `Fn` -/// closure implements it. +/// closure implements it: +/// +/// ```no_run +/// use netwatch::{BindOptions, UdpSocket}; +/// +/// let opts = BindOptions::new().configure_socket( +/// |socket: socket2::SockRef<'_>, _domain: socket2::Domain| { +/// #[cfg(any(target_os = "linux", target_os = "android"))] +/// socket.set_mark(0x80)?; +/// Ok(()) +/// }, +/// ); +/// let socket = UdpSocket::bind_with("0.0.0.0:0".parse::().unwrap(), opts); +/// ``` pub trait SocketConfigurator: Send + Sync + 'static { /// Called before the socket is bound, and again on every rebind. /// From 721a02647e9fd1c5f458ff6dd7172c33803d2f71 Mon Sep 17 00:00:00 2001 From: Dario Date: Thu, 16 Jul 2026 17:31:34 +0200 Subject: [PATCH 06/10] fix(netwatch): allow semicolon_in_expressions_from_macros in build.rs New nightly cargo lints fire inside the cfg_aliases macro; allow it until https://github.com/katharostech/cfg_aliases/pull/15 lands, same as n0-computer/noq#748. --- netwatch/build.rs | 4 ++++ netwatch/src/udp.rs | 5 ++--- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/netwatch/build.rs b/netwatch/build.rs index 404d4748..7ae3180e 100644 --- a/netwatch/build.rs +++ b/netwatch/build.rs @@ -1,5 +1,9 @@ use cfg_aliases::cfg_aliases; +#[allow( + semicolon_in_expressions_from_macros, + reason = "cfg_aliases needs an update: https://github.com/katharostech/cfg_aliases/pull/15" +)] fn main() { // Setup cfg aliases cfg_aliases! { diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index ddabbc82..8c83c4d5 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -1256,9 +1256,8 @@ mod tests { #[tokio::test] async fn test_configure_socket_error_fails_the_bind() -> TestResult { - let opts = BindOptions::new().configure_socket(|_: SockRef<'_>, _: socket2::Domain| { - Err(io::Error::other("nope")) - }); + let opts = BindOptions::new() + .configure_socket(|_: SockRef<'_>, _: socket2::Domain| Err(io::Error::other("nope"))); assert!( UdpSocket::bind_with((Ipv4Addr::LOCALHOST, 0), opts).is_err(), From 4d7a15dfbb96b3c3c82d9446b11970c5cb36275e Mon Sep 17 00:00:00 2001 From: Dario Date: Fri, 17 Jul 2026 08:20:04 +0200 Subject: [PATCH 07/10] cargo update -p cfg_aliases --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index efc6f2fe..278648ca 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -124,9 +124,9 @@ checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" [[package]] name = "cfg_aliases" -version = "0.2.1" +version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" [[package]] name = "chacha20" From 036b9c28aef92fbaf608759bc4189f590e554ed0 Mon Sep 17 00:00:00 2001 From: Dario Date: Sat, 15 Aug 2026 19:17:50 +0200 Subject: [PATCH 08/10] revert(netwatch): drop the semicolon_in_expressions_from_macros allow cfg_aliases 0.2.2 no longer trips the lint, so the allow in build.rs is dead weight. --- netwatch/build.rs | 4 ---- 1 file changed, 4 deletions(-) diff --git a/netwatch/build.rs b/netwatch/build.rs index 7ae3180e..404d4748 100644 --- a/netwatch/build.rs +++ b/netwatch/build.rs @@ -1,9 +1,5 @@ use cfg_aliases::cfg_aliases; -#[allow( - semicolon_in_expressions_from_macros, - reason = "cfg_aliases needs an update: https://github.com/katharostech/cfg_aliases/pull/15" -)] fn main() { // Setup cfg aliases cfg_aliases! { From 96669b461c226a54dead8a967b72d3508506af48 Mon Sep 17 00:00:00 2001 From: Dario Date: Sat, 15 Aug 2026 19:17:50 +0200 Subject: [PATCH 09/10] feat(netwatch): add a configure-socket hook to BindOptions main reverted #179, so this re-adds BindOptions carrying only the general hook: a closure run on every socket right after creation and before bind(), and again on every internal rebind. bind_with replaces the private bind_raw, which all the other bind_* constructors now go through. The hook takes a SocketRef, a borrowed handle that implements AsFd on unix and AsSocket on Windows, and netwatch's own IpFamily. Neither socket2 nor any other external type reaches the public API, so callers can reach for whatever socket crate they like (socket2::SockRef::from(&socket), plain setsockopt) without having to match netwatch's version of it. --- netwatch/src/lib.rs | 2 +- netwatch/src/udp.rs | 192 +++++++++++++++++++++++++++++++++++++++----- 2 files changed, 172 insertions(+), 22 deletions(-) diff --git a/netwatch/src/lib.rs b/netwatch/src/lib.rs index ff172974..29866433 100644 --- a/netwatch/src/lib.rs +++ b/netwatch/src/lib.rs @@ -9,4 +9,4 @@ mod udp; pub use self::ip_family::IpFamily; #[cfg(not(wasm_browser))] -pub use self::udp::{UdpSender, UdpSocket}; +pub use self::udp::{BindOptions, SocketRef, UdpSender, UdpSocket}; diff --git a/netwatch/src/udp.rs b/netwatch/src/udp.rs index be5ce69f..725bf910 100644 --- a/netwatch/src/udp.rs +++ b/netwatch/src/udp.rs @@ -1,3 +1,7 @@ +#[cfg(unix)] +use std::os::fd::{AsFd, BorrowedFd}; +#[cfg(windows)] +use std::os::windows::io::{AsSocket, BorrowedSocket}; use std::{ future::Future, io, @@ -28,6 +32,82 @@ pub struct UdpSocket { /// UDP socket read/write buffer size (7MB). The value of 7MB is chosen as it /// is the max supported by a default configuration of macOS. Some platforms will silently clamp the value. const SOCKET_BUFFER_SIZE: usize = 7 << 20; + +/// A socket that is about to be bound, handed to the hook set with +/// [`BindOptions::configure_socket`]. +/// +/// It implements [`AsFd`] on unix and [`AsSocket`] on Windows, which is what +/// socket wrappers take, so the hook can set options with the socket crate of +/// its choice: `socket2::SockRef::from(&socket)`, or plain `libc::setsockopt` +/// on the raw fd. +#[derive(Debug)] +pub struct SocketRef<'a>(&'a socket2::Socket); + +#[cfg(unix)] +impl AsFd for SocketRef<'_> { + fn as_fd(&self) -> BorrowedFd<'_> { + self.0.as_fd() + } +} + +#[cfg(windows)] +impl AsSocket for SocketRef<'_> { + fn as_socket(&self) -> BorrowedSocket<'_> { + self.0.as_socket() + } +} + +/// The hook set with [`BindOptions::configure_socket`]. +type Configurator = Arc, IpFamily) -> io::Result<()> + Send + Sync>; + +/// Options to bind a [`UdpSocket`] with. +/// +/// Used by [`UdpSocket::bind_with`]. The default options match what the other +/// `bind_*` constructors use. +#[derive(derive_more::Debug, Default, Clone)] +pub struct BindOptions { + #[debug(skip)] + configure: Option, +} + +impl BindOptions { + /// Creates the default options. + pub fn new() -> Self { + Self::default() + } + + /// Sets a hook to run on the socket just before it is bound. + /// + /// This is the escape hatch for socket options this crate does not model + /// itself, like `SO_MARK` on Linux or `IP_BOUND_IF` on Apple platforms. The + /// hook runs on every bind, including the rebinds [`UdpSocket`] does to + /// recover from network changes, since each of those creates a new socket. + /// That also lets the hook pick up state that changed in between, like the + /// interface the default route now points at. + /// + /// An error from the hook fails the bind, rather than leaving a socket that + /// silently missed its configuration. + /// + /// ```no_run + /// use std::net::{Ipv4Addr, SocketAddr}; + /// + /// use netwatch::{BindOptions, UdpSocket}; + /// + /// let opts = BindOptions::new().configure_socket(|socket, _family| { + /// socket2::SockRef::from(&socket).set_recv_buffer_size(1 << 20) + /// }); + /// let socket = UdpSocket::bind_with(SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)), opts)?; + /// # Ok::<(), std::io::Error>(()) + /// ``` + pub fn configure_socket( + mut self, + configure: impl Fn(SocketRef<'_>, IpFamily) -> io::Result<()> + Send + Sync + 'static, + ) -> Self { + self.configure = Some(Arc::new(configure)); + self + } +} + impl UdpSocket { /// Bind only Ipv4 on any interface. pub fn bind_v4(port: u16) -> io::Result { @@ -52,18 +132,30 @@ impl UdpSocket { /// Bind to the given port only on localhost. pub fn bind_local(network: IpFamily, port: u16) -> io::Result { let addr = SocketAddr::new(network.local_addr(), port); - Self::bind_raw(addr) + Self::bind_with(addr, BindOptions::default()) } /// Bind to the given port and listen on all interfaces. pub fn bind(network: IpFamily, port: u16) -> io::Result { let addr = SocketAddr::new(network.unspecified_addr(), port); - Self::bind_raw(addr) + Self::bind_with(addr, BindOptions::default()) } /// Bind to any provided [`SocketAddr`]. pub fn bind_full(addr: impl Into) -> io::Result { - Self::bind_raw(addr) + Self::bind_with(addr, BindOptions::default()) + } + + /// Bind to any provided [`SocketAddr`], using the given [`BindOptions`]. + pub fn bind_with(addr: impl Into, opts: BindOptions) -> io::Result { + let socket = SocketState::bind(addr.into(), opts.configure)?; + + Ok(UdpSocket { + socket: RwLock::new(socket), + recv_waker: AtomicWaker::default(), + send_waker: AtomicWaker::default(), + is_broken: AtomicBool::new(false), + }) } /// Is the socket broken and needs a rebind? @@ -96,17 +188,6 @@ impl UdpSocket { Ok(()) } - fn bind_raw(addr: impl Into) -> io::Result { - let socket = SocketState::bind(addr.into())?; - - Ok(UdpSocket { - socket: RwLock::new(socket), - recv_waker: AtomicWaker::default(), - send_waker: AtomicWaker::default(), - is_broken: AtomicBool::new(false), - }) - } - /// Receives a single datagram message on the socket from the remote address /// to which it is connected. On success, returns the number of bytes read. /// @@ -729,17 +810,23 @@ impl Future for SendToFut<'_, '_> { } } -#[derive(Debug)] +#[derive(derive_more::Debug)] enum SocketState { Connected { socket: tokio::net::UdpSocket, state: noq_udp::UdpSocketState, /// The addr we are binding to. addr: SocketAddr, + /// The hook to rerun when rebinding, if any. + #[debug(skip)] + configure: Option, }, Closed { /// The addr to rebind to when recovering. addr: SocketAddr, + /// The hook to rerun when rebinding, if any. + #[debug(skip)] + configure: Option, last_max_gso_segments: NonZeroUsize, last_gro_segments: NonZeroUsize, last_may_fragment: bool, @@ -753,6 +840,7 @@ impl SocketState { socket, state, addr: _, + configure: _, } => Ok((socket, state)), Self::Closed { .. } => { warn!("socket closed"); @@ -761,7 +849,7 @@ impl SocketState { } } - fn bind(addr: SocketAddr) -> io::Result { + fn bind(addr: SocketAddr, configure: Option) -> io::Result { let network = IpFamily::from(addr.ip()); let socket = socket2::Socket::new( network.into(), @@ -786,6 +874,13 @@ impl SocketState { socket.set_only_v6(true)?; } + // Let the caller configure the socket before it is bound. An error here + // fails the bind: a socket that silently missed its configuration would + // send traffic where the caller did not want it. + if let Some(configure) = &configure { + configure(SocketRef(&socket), network)?; + } + // Binding must happen before calling noq, otherwise `local_addr` // is not yet available on all OSes. socket.bind(&addr.into())?; @@ -814,13 +909,18 @@ impl SocketState { socket, state: socket_state, addr: local_addr, + configure, }) } fn rebind(&mut self) -> io::Result<()> { - let addr = match self { - Self::Connected { addr, .. } => *addr, - Self::Closed { addr, .. } => *addr, + let (addr, configure) = match self { + Self::Connected { + addr, configure, .. + } + | Self::Closed { + addr, configure, .. + } => (*addr, configure.clone()), }; debug!("rebinding {}", addr); @@ -829,13 +929,14 @@ impl SocketState { if let Self::Connected { state, .. } = self { *self = SocketState::Closed { addr, + configure: configure.clone(), last_max_gso_segments: state.max_gso_segments(), last_gro_segments: state.gro_segments(), last_may_fragment: state.may_fragment(), }; } - match Self::bind(addr) { + match Self::bind(addr, configure) { Ok(new_state) => { *self = new_state; Ok(()) @@ -854,9 +955,15 @@ impl SocketState { fn close(&mut self) -> Option<(tokio::net::UdpSocket, noq_udp::UdpSocketState)> { match self { - Self::Connected { state, addr, .. } => { + Self::Connected { + state, + addr, + configure, + .. + } => { let s = SocketState::Closed { addr: *addr, + configure: configure.clone(), last_max_gso_segments: state.max_gso_segments(), last_gro_segments: state.gro_segments(), last_may_fragment: state.may_fragment(), @@ -1068,6 +1175,11 @@ impl Future for SendFutNoq<'_, '_> { #[cfg(test)] mod tests { + use std::{ + net::Ipv4Addr, + sync::atomic::{AtomicUsize, Ordering}, + }; + use testresult::TestResult; use super::*; @@ -1129,6 +1241,44 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_configure_socket_runs_on_every_bind() -> TestResult { + let calls = Arc::new(AtomicUsize::new(0)); + let seen = calls.clone(); + let opts = BindOptions::new().configure_socket(move |socket, family| { + assert_eq!(family, IpFamily::V4); + // The hook gets a real socket, before it is bound. + socket2::SockRef::from(&socket).set_reuse_address(true)?; + seen.fetch_add(1, Ordering::SeqCst); + Ok(()) + }); + + let socket = UdpSocket::bind_with(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), opts)?; + assert_eq!(calls.load(Ordering::SeqCst), 1, "hook did not run on bind"); + + socket.rebind()?; + assert_eq!( + calls.load(Ordering::SeqCst), + 2, + "hook did not run again on rebind" + ); + + Ok(()) + } + + #[tokio::test] + async fn test_configure_socket_error_fails_the_bind() -> TestResult { + let opts = + BindOptions::new().configure_socket(|_socket, _family| Err(io::Error::other("nope"))); + + assert!( + UdpSocket::bind_with(SocketAddr::from((Ipv4Addr::LOCALHOST, 0)), opts).is_err(), + "a failing hook must fail the bind" + ); + + Ok(()) + } + #[tokio::test] async fn test_udp_mark_broken() -> TestResult { let socket_a = UdpSocket::bind_local(IpFamily::V4, 0)?; From 7f5513cd602c29e0b564e6c07d81614244d95730 Mon Sep 17 00:00:00 2001 From: Dario Date: Tue, 1 Sep 2026 14:58:29 +0200 Subject: [PATCH 10/10] chore(deps): update chacha20 to a non-yanked version --- Cargo.lock | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 4eafc528..1850a21f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -130,9 +130,9 @@ checksum = "f079e83a288787bcd14a6aea84cee5c87a67c5a3e660c30f557a3d24761b3527" [[package]] name = "chacha20" -version = "0.10.1" +version = "0.10.2" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" dependencies = [ "cfg-if", "cpufeatures",