dosh/src/udp.rs

//! UDP error classification shared by Dosh clients, servers, and embedders.

use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
use std::time::{Duration, Instant};
use tokio::net::UdpSocket;

pub fn is_transient_udp_error(err: &std::io::Error) -> bool {
    matches!(
        err.kind(),
        std::io::ErrorKind::Interrupted
            | std::io::ErrorKind::TimedOut
            | std::io::ErrorKind::WouldBlock
            | std::io::ErrorKind::ConnectionRefused
            | std::io::ErrorKind::ConnectionReset
            | std::io::ErrorKind::AddrNotAvailable
            | std::io::ErrorKind::NetworkUnreachable
            | std::io::ErrorKind::HostUnreachable
    ) || err.raw_os_error().is_some_and(is_transient_udp_os_error)
}

pub fn is_transient_udp_send_error(err: &std::io::Error) -> bool {
    is_transient_udp_error(err)
}

pub fn unspecified_bind_addr_for_peer(peer: SocketAddr) -> SocketAddr {
    match peer {
        SocketAddr::V4(_) => SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0)),
        SocketAddr::V6(_) => SocketAddr::from((Ipv6Addr::UNSPECIFIED, 0)),
    }
}

pub async fn bind_udp_for_peer(peer: SocketAddr) -> std::io::Result<UdpSocket> {
    UdpSocket::bind(unspecified_bind_addr_for_peer(peer)).await
}

pub async fn send_udp_retrying_transient(
    socket: &UdpSocket,
    packet: &[u8],
    peer: SocketAddr,
    wait: Duration,
) -> std::io::Result<()> {
    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
    loop {
        match socket.send_to(packet, peer).await {
            Ok(_) => return Ok(()),
            Err(err) if is_transient_udp_send_error(&err) && Instant::now() < deadline => {
                sleep_until_retry(deadline).await;
            }
            Err(err) => return Err(err),
        }
    }
}

pub async fn recv_udp_retrying_transient(
    socket: &UdpSocket,
    buf: &mut [u8],
    wait: Duration,
) -> std::io::Result<(usize, SocketAddr)> {
    let deadline = Instant::now() + wait.max(Duration::from_millis(1));
    loop {
        let now = Instant::now();
        if now >= deadline {
            return Err(std::io::Error::from(std::io::ErrorKind::TimedOut));
        }
        match tokio::time::timeout(deadline - now, socket.recv_from(buf)).await {
            Ok(Ok(value)) => return Ok(value),
            Ok(Err(err)) if is_transient_udp_error(&err) && Instant::now() < deadline => {
                sleep_until_retry(deadline).await;
            }
            Ok(Err(err)) => return Err(err),
            Err(_) => return Err(std::io::Error::from(std::io::ErrorKind::TimedOut)),
        }
    }
}

async fn sleep_until_retry(deadline: Instant) {
    let now = Instant::now();
    if now >= deadline {
        return;
    }
    tokio::time::sleep((deadline - now).min(Duration::from_millis(20))).await;
}

#[cfg(unix)]
fn is_transient_udp_os_error(code: i32) -> bool {
    matches!(
        code,
        libc::EADDRNOTAVAIL
            | libc::ECONNABORTED
            | libc::ECONNREFUSED
            | libc::ECONNRESET
            | libc::EHOSTDOWN
            | libc::EHOSTUNREACH
            | libc::ENOBUFS
            | libc::ENETDOWN
            | libc::ENETRESET
            | libc::ENETUNREACH
            | libc::ETIMEDOUT
    )
}

#[cfg(windows)]
fn is_transient_udp_os_error(code: i32) -> bool {
    matches!(
        code,
        10035  // WSAEWOULDBLOCK
            | 10049 // WSAEADDRNOTAVAIL
            | 10050 // WSAENETDOWN
            | 10051 // WSAENETUNREACH
            | 10052 // WSAENETRESET
            | 10053 // WSAECONNABORTED
            | 10054 // WSAECONNRESET
            | 10055 // WSAENOBUFS
            | 10060 // WSAETIMEDOUT
            | 10061 // WSAECONNREFUSED
            | 10064 // WSAEHOSTDOWN
            | 10065 // WSAEHOSTUNREACH
    )
}

#[cfg(not(any(unix, windows)))]
fn is_transient_udp_os_error(_code: i32) -> bool {
    false
}

#[cfg(test)]
mod tests {
    use super::{
        is_transient_udp_error, is_transient_udp_send_error, recv_udp_retrying_transient,
        send_udp_retrying_transient, unspecified_bind_addr_for_peer,
    };
    use std::net::{Ipv4Addr, Ipv6Addr, SocketAddr};
    use std::time::Duration;
    use tokio::net::UdpSocket;

    #[test]
    fn classifies_portable_transient_udp_errors() {
        for kind in [
            std::io::ErrorKind::Interrupted,
            std::io::ErrorKind::TimedOut,
            std::io::ErrorKind::WouldBlock,
            std::io::ErrorKind::ConnectionRefused,
            std::io::ErrorKind::ConnectionReset,
            std::io::ErrorKind::AddrNotAvailable,
            std::io::ErrorKind::NetworkUnreachable,
            std::io::ErrorKind::HostUnreachable,
        ] {
            let err = std::io::Error::from(kind);
            assert!(is_transient_udp_error(&err));
            assert!(is_transient_udp_send_error(&err));
        }
        let fatal = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
        assert!(!is_transient_udp_error(&fatal));
        assert!(!is_transient_udp_send_error(&fatal));
    }

    #[test]
    fn peer_bind_addr_matches_peer_address_family() {
        assert_eq!(
            unspecified_bind_addr_for_peer(SocketAddr::from((Ipv4Addr::LOCALHOST, 50000))),
            SocketAddr::from((Ipv4Addr::UNSPECIFIED, 0))
        );
        assert_eq!(
            unspecified_bind_addr_for_peer(SocketAddr::from((Ipv6Addr::LOCALHOST, 50000))),
            SocketAddr::from((Ipv6Addr::UNSPECIFIED, 0))
        );
    }

    #[tokio::test]
    async fn retrying_udp_send_and_receive_round_trip() {
        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let sender = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let addr = receiver.local_addr().unwrap();

        send_udp_retrying_transient(&sender, b"hello", addr, Duration::from_millis(50))
            .await
            .unwrap();
        let mut buf = [0u8; 16];
        let (n, peer) = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(50))
            .await
            .unwrap();

        assert_eq!(&buf[..n], b"hello");
        assert_eq!(peer, sender.local_addr().unwrap());
    }

    #[tokio::test]
    async fn retrying_udp_receive_times_out_cleanly() {
        let receiver = UdpSocket::bind("127.0.0.1:0").await.unwrap();
        let mut buf = [0u8; 16];
        let err = recv_udp_retrying_transient(&receiver, &mut buf, Duration::from_millis(1))
            .await
            .unwrap_err();

        assert_eq!(err.kind(), std::io::ErrorKind::TimedOut);
    }

    #[cfg(unix)]
    #[test]
    fn classifies_unix_network_churn_as_transient() {
        for code in [
            libc::EADDRNOTAVAIL,
            libc::ECONNABORTED,
            libc::ECONNREFUSED,
            libc::ECONNRESET,
            libc::EHOSTDOWN,
            libc::EHOSTUNREACH,
            libc::ENOBUFS,
            libc::ENETDOWN,
            libc::ENETRESET,
            libc::ENETUNREACH,
            libc::ETIMEDOUT,
        ] {
            let err = std::io::Error::from_raw_os_error(code);
            assert!(is_transient_udp_error(&err));
            assert!(is_transient_udp_send_error(&err));
        }
    }

    #[cfg(windows)]
    #[test]
    fn classifies_windows_network_churn_as_transient() {
        for code in [
            10035, 10049, 10050, 10051, 10052, 10053, 10054, 10055, 10060, 10061, 10064, 10065,
        ] {
            let err = std::io::Error::from_raw_os_error(code);
            assert!(is_transient_udp_error(&err));
            assert!(is_transient_udp_send_error(&err));
        }
    }
}