Cargo.lock

@@ -735,6 +735,7 @@ dependencies = [
  "getrandom 0.4.3",
  "irtt-client",
  "irtt-proto",
+ "nix 0.31.3",
  "socket2",
  "thiserror 2.0.20",
  "tokio",

crates/irtt-server/Cargo.toml

@@ -18,5 +18,11 @@ socket2 = { version = "0.6.5", features = ["all"] }
 thiserror = "2.0.20"
 tokio = { version = "1.53.1", default-features = false, features = ["rt", "net", "time", "sync", "macros"] }
 
+# Wildcard listeners recover each request's local destination address and send
+# the reply from it. Only these targets have a safe destination-metadata and
+# source-selection path here; elsewhere a wildcard bind is refused.
+[target.'cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))'.dependencies]
+nix = { version = "0.31.3", default-features = false, features = ["net", "uio"] }
+
 [dev-dependencies]
 irtt-client = { workspace = true, features = ["ancillary", "tokio"] }

crates/irtt-server/src/lib.rs

@@ -56,6 +56,7 @@ mod error;
 mod negotiate;
 mod runtime;
 mod session;
+mod socket_io;
 mod socket_options;
 mod token;
 

crates/irtt-server/src/runtime.rs

@@ -6,7 +6,9 @@ use socket2::SockRef;
 use thiserror::Error;
 use tokio::{net::UdpSocket, time::MissedTickBehavior};
 
-use crate::{socket_options::set_reply_traffic_class, ServerConfig, ServerCore, ServerError};
+use crate::{
+    socket_io, socket_options::set_reply_traffic_class, ServerConfig, ServerCore, ServerError,
+};
 
 /// Fixed transport receive capacity.
 ///
@@ -30,20 +32,34 @@ const MAINTENANCE_INTERVAL: Duration = Duration::from_secs(1);
 /// marking. Adding a concurrent sender would break that and require per-packet
 /// control messages instead.
 ///
-/// Bind an explicit local address when reply source-address identity matters.
-/// Wildcard binds rely on the kernel's source-address choice; destination
-/// packet metadata and per-packet source selection are not yet implemented for
-/// multi-homed hosts.
+/// # Reply source address
+///
+/// A reply must leave from the exact address the request was sent to. An
+/// explicit-address listener gets that from the bind: it can send from nothing
+/// else. A wildcard listener (`0.0.0.0` or `[::]`) cannot, because the routing
+/// table would choose the source, so it asks the kernel for each request's
+/// local destination and sends that request's reply from it.
+///
+/// That per-packet path exists on Linux, macOS and FreeBSD. Elsewhere a
+/// wildcard bind is **refused** at construction — see
+/// [`ServerRuntimeError::WildcardSourceSelectionUnsupported`] — rather than
+/// served by a listener whose replies a client on a second address would
+/// silently discard. Explicit-address listeners are unaffected everywhere.
 #[derive(Debug)]
 pub struct Server {
     socket: UdpSocket,
     core: ServerCore,
     recv_buffer: Vec<u8>,
     /// The listener's own address family, which selects the IPv4 TOS or IPv6
-    /// Traffic Class option. `None` only if the bound address could not be
-    /// queried at construction, in which case the send path asks again rather
-    /// than guessing a family.
-    listener_is_ipv4: Option<bool>,
+    /// Traffic Class option.
+    listener_is_ipv4: bool,
+    /// Whether this listener is wildcard-bound and must therefore recover each
+    /// request's local destination and reply from it.
+    ///
+    /// Decided once, from the bound address. It is never inferred later from a
+    /// peer: a listener that accepted a wildcard bind owes correct reply
+    /// sources to every request, not to the ones that look multi-homed.
+    select_reply_source: bool,
     /// Whether this server has ever successfully applied a nonzero traffic
     /// class to the socket.
     ///
@@ -57,24 +73,55 @@ pub struct Server {
 impl Server {
     /// Binds one UDP listener at `addr` and creates its independent session
     /// namespace.
+    ///
+    /// A wildcard `addr` additionally configures reply source selection, and
+    /// fails where that is unavailable; see [`Server::from_socket`].
     pub async fn bind(addr: SocketAddr, config: ServerConfig) -> Result<Self, ServerRuntimeError> {
         let socket = UdpSocket::bind(addr)
             .await
             .map_err(|source| ServerRuntimeError::Bind { addr, source })?;
-        Ok(Self::from_socket(socket, config))
+        Self::from_socket(socket, config)
     }
 
     /// Wraps an already prepared Tokio UDP socket in a server.
-    #[must_use]
-    pub fn from_socket(socket: UdpSocket, config: ServerConfig) -> Self {
-        let listener_is_ipv4 = socket.local_addr().ok().map(|addr| addr.is_ipv4());
-        Self {
+    ///
+    /// # Errors
+    ///
+    /// Construction is fallible because a wildcard listener has setup to do
+    /// before it may serve anything. It fails when the socket's bound address
+    /// cannot be queried — which is what says whether this is a wildcard
+    /// listener at all — and, for a wildcard socket, when this target has no
+    /// reply source selection or the kernel refuses to configure it.
+    ///
+    /// Failing here is the point. A wildcard listener that cannot recover a
+    /// request's local destination would start, run, and answer clients on a
+    /// second local address from an endpoint they never contacted, which they
+    /// discard as though the network had dropped it.
+    pub fn from_socket(
+        socket: UdpSocket,
+        config: ServerConfig,
+    ) -> Result<Self, ServerRuntimeError> {
+        let addr = socket
+            .local_addr()
+            .map_err(|source| ServerRuntimeError::LocalAddr { source })?;
+        let select_reply_source = addr.ip().is_unspecified();
+
+        if select_reply_source {
+            if !socket_io::SUPPORTED {
+                return Err(ServerRuntimeError::WildcardSourceSelectionUnsupported { addr });
+            }
+            socket_io::configure_destination_metadata(&socket, addr.is_ipv4())
+                .map_err(|source| ServerRuntimeError::SourceSelectionSetup { addr, source })?;
+        }
+
+        Ok(Self {
             socket,
             core: ServerCore::new(config),
             recv_buffer: vec![0; RECEIVE_BUFFER_LEN],
-            listener_is_ipv4,
+            listener_is_ipv4: addr.is_ipv4(),
+            select_reply_source,
             socket_is_marked: false,
-        }
+        })
     }
 
     /// Applies one reply's raw traffic class to the listener socket, and
@@ -104,11 +151,11 @@ impl Server {
     }
 
     fn apply_reply_traffic_class(&self, traffic_class: u8) -> io::Result<()> {
-        let is_ipv4 = match self.listener_is_ipv4 {
-            Some(is_ipv4) => is_ipv4,
-            None => self.socket.local_addr()?.is_ipv4(),
-        };
-        set_reply_traffic_class(SockRef::from(&self.socket), is_ipv4, traffic_class)
+        set_reply_traffic_class(
+            SockRef::from(&self.socket),
+            self.listener_is_ipv4,
+            traffic_class,
+        )
     }
 
     /// Returns the local endpoint selected for the listener.
@@ -133,6 +180,12 @@ impl Server {
     /// *unmarked* reply on a socket this server has never marked is the one
     /// exception, so a host that does not support the option at all — some
     /// Windows builds do not support `IP_TOS` — still runs a working server.
+    ///
+    /// A wildcard listener drops a request whose local destination did not
+    /// arrive with it, before the core sees it. There is no fall back to a
+    /// routing-table source: the listener promised a correct reply source when
+    /// it accepted the bind, and a request it cannot answer correctly must not
+    /// move a session's receive, rate or lifetime state either.
     pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
     where
         F: Future<Output = ()>,
@@ -146,12 +199,18 @@ impl Server {
             tokio::select! {
                 _ = &mut shutdown => return Ok(()),
                 _ = maintenance.tick() => self.core.maintain(),
-                received = self.socket.recv_from(&mut self.recv_buffer) => {
-                    let (len, peer) = match received {
-                        Ok(received) => received,
+                received = socket_io::receive(
+                    &self.socket,
+                    &mut self.recv_buffer,
+                    self.select_reply_source,
+                ) => {
+                    let received = match received {
+                        Ok(Some(received)) => received,
+                        Ok(None) => continue,
                         Err(source) if source.kind() == io::ErrorKind::Interrupted => continue,
                         Err(source) => return Err(ServerRuntimeError::Receive { source }),
                     };
+                    let (len, peer) = (received.len, received.peer);
                     if len == RECEIVE_BUFFER_LEN {
                         continue;
                     }
@@ -160,7 +219,12 @@ impl Server {
                         if !self.prepare_reply_traffic_class(reply.traffic_class()) {
                             continue;
                         }
-                        let send = self.socket.send_to(reply.bytes(), peer);
+                        let send = socket_io::send(
+                            &self.socket,
+                            reply.bytes(),
+                            peer,
+                            received.reply_source,
+                        );
                         tokio::pin!(send);
                         let sent = loop {
                             tokio::select! {
@@ -210,6 +274,26 @@ pub enum ServerRuntimeError {
         #[source]
         source: io::Error,
     },
+    /// A wildcard listener was requested on a target with no safe per-packet
+    /// reply source selection.
+    ///
+    /// The listener would answer from whichever local address the routing table
+    /// chose, which a client that contacted a different one discards.
+    #[error(
+        "wildcard listener {addr} cannot select its reply source address on this target: \
+         bind an explicit local address instead"
+    )]
+    WildcardSourceSelectionUnsupported { addr: SocketAddr },
+    /// Configuring destination-address metadata for a wildcard listener failed.
+    #[error(
+        "could not configure reply source-address selection for wildcard listener {addr}: \
+         {source}; bind an explicit local address instead"
+    )]
+    SourceSelectionSetup {
+        addr: SocketAddr,
+        #[source]
+        source: io::Error,
+    },
     /// Receiving from the listener failed irrecoverably.
     #[error("UDP receive failed: {source}")]
     Receive {
@@ -253,6 +337,31 @@ mod tests {
             .unwrap();
     }
 
+    /// Construction is where a wildcard listener's promise is made or refused.
+    ///
+    /// Both outcomes are asserted from the same code, because which one a target
+    /// gets is exactly the thing under test: a target with a source-selection
+    /// path configures it and serves; one without refuses the bind rather than
+    /// starting a listener whose replies may come from an address no client
+    /// contacted. An explicit bind is unaffected either way.
+    #[tokio::test(flavor = "current_thread")]
+    async fn from_socket_settles_wildcard_source_selection_before_serving_anything() {
+        let explicit = UdpSocket::bind("127.0.0.1:0").await.unwrap();
+        assert!(Server::from_socket(explicit, ServerConfig::default()).is_ok());
+
+        let wildcard = UdpSocket::bind("0.0.0.0:0").await.unwrap();
+        let wildcard = Server::from_socket(wildcard, ServerConfig::default());
+        if socket_io::SUPPORTED {
+            let server = wildcard.expect("a supported target configures source selection");
+            assert!(server.select_reply_source);
+        } else {
+            assert!(matches!(
+                wildcard,
+                Err(ServerRuntimeError::WildcardSourceSelectionUnsupported { .. })
+            ));
+        }
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn ipv4_open_echo_close_crosses_a_real_udp_socket() {
         exercise_udp_path(SocketAddr::from((Ipv4Addr::LOCALHOST, 0))).await;

crates/irtt-server/src/socket_io.rs

@@ -0,0 +1,155 @@
+//! Receiving a request's local destination and replying from it.
+//!
+//! A reply must leave from the exact address and port the request was addressed
+//! to, because clients read from connected UDP sockets and discard anything
+//! arriving from another endpoint. A listener bound to an explicit address
+//! satisfies that for free: every datagram it sends carries the address it is
+//! bound to, so ordinary `recv_from`/`send_to` is all it needs.
+//!
+//! A wildcard listener does not. On a multi-homed host the routing table
+//! chooses the source address, so a request to one of the host's addresses can
+//! be answered from another, and the client drops the reply. Such a listener
+//! therefore asks the kernel for each request's destination address as
+//! ancillary data and hands it back as the source of that request's reply.
+//!
+//! Which mechanism carries that is per-platform, and only [`SUPPORTED`] targets
+//! have one reachable through a safe API. Everything here goes through `nix`'s
+//! safe `recvmsg`/`sendmsg` wrappers; the crate forbids unsafe code, so there is
+//! no raw syscall, no `CMSG_*` walking and no hand-written ABI layer.
+
+use std::{io, net::SocketAddr};
+
+use tokio::net::UdpSocket;
+
+#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+mod ancillary;
+
+/// Whether this target can pin a wildcard listener's reply source address.
+///
+/// A wildcard bind is refused where this is false, rather than served by a
+/// listener whose replies may leave from an address the client never contacted.
+pub(crate) const SUPPORTED: bool = cfg!(any(
+    target_os = "linux",
+    target_os = "macos",
+    target_os = "freebsd"
+));
+
+/// Where one reply must leave from.
+///
+/// This is transport state belonging to a single datagram. It never reaches
+/// [`ServerCore`](crate::ServerCore), which knows peers and sessions and has no
+/// business knowing which of the host's addresses a request arrived on.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ReplySource {
+    /// Whatever the socket is bound to: an explicit-address listener already
+    /// sends from the one address it can.
+    Bound,
+    /// The IPv4 address this request was addressed to.
+    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+    V4 { address: std::net::Ipv4Addr },
+    /// The IPv6 address this request was addressed to, and the interface it
+    /// arrived on. The interface is part of the identity of a scoped address,
+    /// so it is carried rather than rediscovered.
+    #[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+    V6 {
+        address: std::net::Ipv6Addr,
+        interface_index: u32,
+    },
+}
+
+/// One received datagram and what answering it requires.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct ReceivedDatagram {
+    pub(crate) len: usize,
+    pub(crate) peer: SocketAddr,
+    pub(crate) reply_source: ReplySource,
+}
+
+/// Asks the kernel to report each datagram's local destination address.
+///
+/// Called once, at construction, and only for a wildcard listener: an
+/// explicit-address listener needs no ancillary metadata, and adding it would
+/// widen the platform surface of the common path for nothing.
+#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+pub(crate) fn configure_destination_metadata(socket: &UdpSocket, is_ipv4: bool) -> io::Result<()> {
+    ancillary::configure_destination_metadata(socket, is_ipv4)
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "freebsd")))]
+pub(crate) fn configure_destination_metadata(
+    _socket: &UdpSocket,
+    _is_ipv4: bool,
+) -> io::Result<()> {
+    Err(unsupported())
+}
+
+/// Receives one datagram, reporting where its reply must come from.
+///
+/// `select_reply_source` is the listener's own wildcard-ness, decided once at
+/// construction. It is never inferred from a peer address.
+///
+/// `Ok(None)` is a datagram the runtime must drop without letting it reach the
+/// core: it was truncated, its ancillary metadata was truncated, or a wildcard
+/// listener could not recover its local destination. Answering it correctly is
+/// impossible, so it must not advance any session's state either.
+///
+/// Cancel-safe: nothing is consumed until a datagram is fully received.
+pub(crate) async fn receive(
+    socket: &UdpSocket,
+    buffer: &mut [u8],
+    select_reply_source: bool,
+) -> io::Result<Option<ReceivedDatagram>> {
+    if select_reply_source {
+        return receive_with_destination(socket, buffer).await;
+    }
+
+    let (len, peer) = socket.recv_from(buffer).await?;
+    Ok(Some(ReceivedDatagram {
+        len,
+        peer,
+        reply_source: ReplySource::Bound,
+    }))
+}
+
+#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+async fn receive_with_destination(
+    socket: &UdpSocket,
+    buffer: &mut [u8],
+) -> io::Result<Option<ReceivedDatagram>> {
+    ancillary::receive_with_destination(socket, buffer).await
+}
+
+/// Unreachable in practice: a wildcard bind is refused at construction here, so
+/// no listener on this target ever asks for source selection.
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "freebsd")))]
+async fn receive_with_destination(
+    _socket: &UdpSocket,
+    _buffer: &mut [u8],
+) -> io::Result<Option<ReceivedDatagram>> {
+    Err(unsupported())
+}
+
+/// Sends one reply to `peer` from `reply_source`.
+///
+/// A short send is reported as such and is the caller's to treat as the loss of
+/// that one reply, exactly as an ordinary `send_to` short send is.
+pub(crate) async fn send(
+    socket: &UdpSocket,
+    bytes: &[u8],
+    peer: SocketAddr,
+    reply_source: ReplySource,
+) -> io::Result<usize> {
+    match reply_source {
+        ReplySource::Bound => socket.send_to(bytes, peer).await,
+        #[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
+        selected => ancillary::send_from_source(socket, bytes, peer, selected).await,
+    }
+}
+
+#[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "freebsd")))]
+fn unsupported() -> io::Error {
+    io::Error::new(
+        io::ErrorKind::Unsupported,
+        "reply source-address selection is unsupported on this target",
+    )
+}

crates/irtt-server/src/socket_io/ancillary.rs

@@ -0,0 +1,299 @@
+//! Destination-address ancillary data on Linux, macOS and FreeBSD.
+//!
+//! Three platforms, two mechanisms per address family, all reached through
+//! `nix`'s safe wrappers:
+//!
+//! | | receive | send |
+//! |---|---|---|
+//! | Linux IPv4 | `IP_PKTINFO` | `IP_PKTINFO` |
+//! | Linux IPv6 | `IPV6_RECVPKTINFO` | `IPV6_PKTINFO` |
+//! | macOS IPv4 | `IP_RECVDSTADDR` | `IP_PKTINFO` |
+//! | macOS IPv6 | `IPV6_RECVPKTINFO` | `IPV6_PKTINFO` |
+//! | FreeBSD IPv4 | `IP_RECVDSTADDR` | `IP_SENDSRCADDR` |
+//! | FreeBSD IPv6 | `IPV6_RECVPKTINFO` | `IPV6_PKTINFO` |
+//!
+//! The BSD IPv4 options are not Linux's under another name, and the receive
+//! structure is not the send structure: `in_pktinfo` carries three fields on
+//! receive of which only two mean anything on send. Each direction is built
+//! explicitly for that reason.
+//!
+//! Both directions run nonblocking through Tokio readiness, following the same
+//! `readable()`/`try_io` shape the client's ancillary receive path uses. One
+//! socket, one task, no blocking call and no spawned thread.
+
+use std::{
+    io::{self, IoSlice, IoSliceMut},
+    net::{Ipv4Addr, Ipv6Addr, SocketAddr},
+    os::fd::AsRawFd,
+};
+
+use nix::{
+    libc,
+    sys::socket::{
+        recvmsg, sendmsg, setsockopt, sockopt, ControlMessage, ControlMessageOwned, MsgFlags,
+        RecvMsg, SockaddrStorage,
+    },
+};
+use tokio::{io::Interest, net::UdpSocket};
+
+use super::{ReceivedDatagram, ReplySource};
+
+/// Room for the destination control message. The largest of them is an
+/// `in6_pktinfo`, and this leaves headroom for a second message the kernel may
+/// add, so `MSG_CTRUNC` reflects a real problem rather than a tight buffer.
+const CONTROL_LEN: usize = 128;
+
+pub(super) fn configure_destination_metadata(socket: &UdpSocket, is_ipv4: bool) -> io::Result<()> {
+    if is_ipv4 {
+        configure_ipv4_destination(socket)
+    } else {
+        // A dual-stack wildcard socket reports IPv4-mapped destinations through
+        // this option too, so the IPv6 listener needs only this one.
+        setsockopt(socket, sockopt::Ipv6RecvPacketInfo, &true).map_err(io::Error::from)
+    }
+}
+
+#[cfg(target_os = "linux")]
+fn configure_ipv4_destination(socket: &UdpSocket) -> io::Result<()> {
+    setsockopt(socket, sockopt::Ipv4PacketInfo, &true).map_err(io::Error::from)
+}
+
+#[cfg(any(target_os = "macos", target_os = "freebsd"))]
+fn configure_ipv4_destination(socket: &UdpSocket) -> io::Result<()> {
+    setsockopt(socket, sockopt::Ipv4RecvDstAddr, &true).map_err(io::Error::from)
+}
+
+pub(super) async fn receive_with_destination(
+    socket: &UdpSocket,
+    buffer: &mut [u8],
+) -> io::Result<Option<ReceivedDatagram>> {
+    loop {
+        socket.readable().await?;
+        match socket.try_io(Interest::READABLE, || receive_once(socket, buffer)) {
+            Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
+            result => return result,
+        }
+    }
+}
+
+fn receive_once(socket: &UdpSocket, buffer: &mut [u8]) -> io::Result<Option<ReceivedDatagram>> {
+    let mut control = ControlBuffer::new();
+    let mut iov = [IoSliceMut::new(buffer)];
+    let message = recvmsg::<SockaddrStorage>(
+        socket.as_raw_fd(),
+        &mut iov,
+        Some(control.as_mut_slice()),
+        MsgFlags::MSG_DONTWAIT,
+    )?;
+
+    Ok(received_datagram(&message))
+}
+
+/// Turns one received message into a datagram the runtime may serve, or `None`
+/// for one it must drop.
+///
+/// A wildcard listener promised to answer from the address the request was sent
+/// to. If the payload was truncated, if the control buffer was truncated, or if
+/// no destination arrived with the datagram, that promise cannot be kept for
+/// this request — so it is dropped here, before the core sees it and before any
+/// session's receive, rate or lifetime state could move.
+fn received_datagram(message: &RecvMsg<'_, '_, SockaddrStorage>) -> Option<ReceivedDatagram> {
+    if message
+        .flags
+        .intersects(MsgFlags::MSG_TRUNC | MsgFlags::MSG_CTRUNC)
+    {
+        return None;
+    }
+
+    Some(ReceivedDatagram {
+        len: message.bytes,
+        peer: peer_endpoint(message.address.as_ref()?)?,
+        reply_source: reply_source(message)?,
+    })
+}
+
+/// The request's full source endpoint, keeping the IPv6 zone the core compares
+/// endpoints by.
+fn peer_endpoint(address: &SockaddrStorage) -> Option<SocketAddr> {
+    if let Some(peer) = address.as_sockaddr_in() {
+        return Some(SocketAddr::from(*peer));
+    }
+    address
+        .as_sockaddr_in6()
+        .map(|peer| SocketAddr::from(*peer))
+}
+
+fn reply_source(message: &RecvMsg<'_, '_, SockaddrStorage>) -> Option<ReplySource> {
+    for control_message in message.cmsgs().ok()? {
+        match control_message {
+            // Linux reports the header destination in `ipi_addr`; `ipi_spec_dst`
+            // is the route's local address, which is the same for the unicast
+            // traffic a server sees and is not what the client addressed.
+            #[cfg(target_os = "linux")]
+            ControlMessageOwned::Ipv4PacketInfo(info) => {
+                return ipv4_reply_source(ipv4_addr(info.ipi_addr));
+            }
+            #[cfg(any(target_os = "macos", target_os = "freebsd"))]
+            ControlMessageOwned::Ipv4RecvDstAddr(address) => {
+                return ipv4_reply_source(ipv4_addr(address));
+            }
+            ControlMessageOwned::Ipv6PacketInfo(info) => {
+                let address = Ipv6Addr::from(info.ipi6_addr.s6_addr);
+                if address.is_unspecified() {
+                    return None;
+                }
+                return Some(ReplySource::V6 {
+                    address,
+                    interface_index: info.ipi6_ifindex,
+                });
+            }
+            _ => {}
+        }
+    }
+    None
+}
+
+/// An unspecified destination is no destination: nothing can be sent from
+/// `0.0.0.0`, so the datagram is dropped rather than answered from a source the
+/// kernel would pick instead.
+fn ipv4_reply_source(address: Ipv4Addr) -> Option<ReplySource> {
+    (!address.is_unspecified()).then_some(ReplySource::V4 { address })
+}
+
+fn ipv4_addr(address: libc::in_addr) -> Ipv4Addr {
+    Ipv4Addr::from(address.s_addr.to_ne_bytes())
+}
+
+fn in_addr(address: Ipv4Addr) -> libc::in_addr {
+    libc::in_addr {
+        s_addr: u32::from_ne_bytes(address.octets()),
+    }
+}
+
+pub(super) async fn send_from_source(
+    socket: &UdpSocket,
+    bytes: &[u8],
+    peer: SocketAddr,
+    reply_source: ReplySource,
+) -> io::Result<usize> {
+    loop {
+        socket.writable().await?;
+        match socket.try_io(Interest::WRITABLE, || {
+            send_once(socket, bytes, peer, reply_source)
+        }) {
+            Err(error) if error.kind() == io::ErrorKind::WouldBlock => continue,
+            result => return result,
+        }
+    }
+}
+
+fn send_once(
+    socket: &UdpSocket,
+    bytes: &[u8],
+    peer: SocketAddr,
+    reply_source: ReplySource,
+) -> io::Result<usize> {
+    let destination = SockaddrStorage::from(peer);
+    let iov = [IoSlice::new(bytes)];
+
+    let sent = match reply_source {
+        ReplySource::Bound => sendmsg(
+            socket.as_raw_fd(),
+            &iov,
+            &[],
+            MsgFlags::MSG_DONTWAIT,
+            Some(&destination),
+        )?,
+        ReplySource::V4 { address } => send_from_ipv4(socket, &iov, &destination, address)?,
+        ReplySource::V6 {
+            address,
+            interface_index,
+        } => {
+            // For a send, `in6_pktinfo` names the source address and the
+            // interface to leave by — the interface being what makes a scoped
+            // address unambiguous.
+            let info = libc::in6_pktinfo {
+                ipi6_addr: libc::in6_addr {
+                    s6_addr: address.octets(),
+                },
+                ipi6_ifindex: interface_index,
+            };
+            sendmsg(
+                socket.as_raw_fd(),
+                &iov,
+                &[ControlMessage::Ipv6PacketInfo(&info)],
+                MsgFlags::MSG_DONTWAIT,
+                Some(&destination),
+            )?
+        }
+    };
+
+    Ok(sent)
+}
+
+/// `IP_PKTINFO` on a send reads only `ipi_spec_dst`, the source address, and
+/// `ipi_ifindex`. `ipi_addr` is a receive-only field and is left zero.
+///
+/// The interface is left unset deliberately. What the protocol requires is the
+/// source *address*; pinning the arrival interface as well would add a way for
+/// a reply to fail on a host whose return path is not the one the request came
+/// in on, and buy nothing.
+#[cfg(any(target_os = "linux", target_os = "macos"))]
+fn send_from_ipv4(
+    socket: &UdpSocket,
+    iov: &[IoSlice<'_>],
+    destination: &SockaddrStorage,
+    address: Ipv4Addr,
+) -> io::Result<usize> {
+    let info = libc::in_pktinfo {
+        ipi_ifindex: 0,
+        ipi_spec_dst: in_addr(address),
+        ipi_addr: in_addr(Ipv4Addr::UNSPECIFIED),
+    };
+
+    sendmsg(
+        socket.as_raw_fd(),
+        iov,
+        &[ControlMessage::Ipv4PacketInfo(&info)],
+        MsgFlags::MSG_DONTWAIT,
+        Some(destination),
+    )
+    .map_err(io::Error::from)
+}
+
+/// FreeBSD has no send-side `IP_PKTINFO`; `IP_SENDSRCADDR` carries the source
+/// address on its own.
+#[cfg(target_os = "freebsd")]
+fn send_from_ipv4(
+    socket: &UdpSocket,
+    iov: &[IoSlice<'_>],
+    destination: &SockaddrStorage,
+    address: Ipv4Addr,
+) -> io::Result<usize> {
+    let source = in_addr(address);
+
+    sendmsg(
+        socket.as_raw_fd(),
+        iov,
+        &[ControlMessage::Ipv4SendSrcAddr(&source)],
+        MsgFlags::MSG_DONTWAIT,
+        Some(destination),
+    )
+    .map_err(io::Error::from)
+}
+
+/// Per-receive control buffer. Control data must satisfy `cmsghdr` alignment,
+/// which 8-byte alignment covers on these targets. It lives on the stack for
+/// the duration of one receive; nothing is allocated per datagram.
+#[repr(align(8))]
+struct ControlBuffer([u8; CONTROL_LEN]);
+
+impl ControlBuffer {
+    fn new() -> Self {
+        Self([0; CONTROL_LEN])
+    }
+
+    fn as_mut_slice(&mut self) -> &mut [u8] {
+        &mut self.0
+    }
+}