refactor(irtt-client): modularize client components

crates/irtt-client/src/config.rs

@@ -0,0 +1,74 @@
+use std::{net::SocketAddr, time::Duration};
+
+use irtt_proto::{Clock, ReceivedStats, StampAt};
+
+pub(crate) const DEFAULT_PORT: u16 = 2112;
+pub(crate) const DEFAULT_DURATION: Duration = Duration::from_secs(3);
+pub(crate) const DEFAULT_INTERVAL: Duration = Duration::from_secs(1);
+pub(crate) const DEFAULT_OPEN_TIMEOUTS: [Duration; 4] = [
+    Duration::from_secs(1),
+    Duration::from_secs(2),
+    Duration::from_secs(4),
+    Duration::from_secs(8),
+];
+pub(crate) const MIN_OPEN_TIMEOUT: Duration = Duration::from_millis(200);
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ClientConfig {
+    pub server_addr: String,
+    pub duration: Option<Duration>,
+    pub interval: Duration,
+    pub length: u32,
+    pub received_stats: ReceivedStats,
+    pub stamp_at: StampAt,
+    pub clock: Clock,
+    pub dscp: u8,
+    pub hmac_key: Option<Vec<u8>>,
+    pub server_fill: Option<String>,
+    pub open_timeouts: Vec<Duration>,
+    pub run_mode: RunMode,
+    pub negotiation_policy: NegotiationPolicy,
+    pub socket_config: SocketConfig,
+}
+
+impl Default for ClientConfig {
+    fn default() -> Self {
+        Self {
+            server_addr: format!("127.0.0.1:{DEFAULT_PORT}"),
+            duration: Some(DEFAULT_DURATION),
+            interval: DEFAULT_INTERVAL,
+            length: 0,
+            received_stats: ReceivedStats::Both,
+            stamp_at: StampAt::Both,
+            clock: Clock::Both,
+            dscp: 0,
+            hmac_key: None,
+            server_fill: None,
+            open_timeouts: DEFAULT_OPEN_TIMEOUTS.to_vec(),
+            run_mode: RunMode::Normal,
+            negotiation_policy: NegotiationPolicy::Strict,
+            socket_config: SocketConfig::default(),
+        }
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Default)]
+pub struct SocketConfig {
+    pub bind_addr: Option<SocketAddr>,
+    pub ttl: Option<u32>,
+    pub ipv4_only: bool,
+    pub ipv6_only: bool,
+    pub recv_timeout: Option<Duration>,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum NegotiationPolicy {
+    Strict,
+    Loose,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RunMode {
+    Normal,
+    NoTest,
+}

crates/irtt-client/src/error.rs

@@ -0,0 +1,40 @@
+use std::{io, time::Duration};
+
+use thiserror::Error;
+
+#[derive(Debug, Error)]
+pub enum ClientError {
+    #[error("failed to resolve server address {addr:?}")]
+    Resolve { addr: String },
+    #[error("socket error: {0}")]
+    Socket(#[from] io::Error),
+    #[error("protocol error: {0}")]
+    Protocol(#[from] irtt_proto::ProtoError),
+    #[error("all open requests timed out")]
+    OpenTimeout,
+    #[error("open timeout {timeout:?} is below the minimum {minimum:?}")]
+    OpenTimeoutTooSmall {
+        timeout: Duration,
+        minimum: Duration,
+    },
+    #[error("server rejected the open request")]
+    ServerRejected,
+    #[error("unexpected no-test open reply")]
+    UnexpectedNoTestReply,
+    #[error("no-test reply included a non-zero connection token: {token}")]
+    NonZeroNoTestToken { token: u64 },
+    #[error("protocol version mismatch: requested {requested}, received {received}")]
+    ProtocolVersionMismatch { requested: i64, received: i64 },
+    #[error("server returned a zero connection token")]
+    ZeroToken,
+    #[error("strict negotiation rejected changed params: {reason}")]
+    NegotiationRejected { reason: String },
+    #[error("client is not open")]
+    NotOpen,
+    #[error("client session is already closed")]
+    AlreadyClosed,
+    #[error("operation is not implemented for this milestone")]
+    NotImplementedForMilestone,
+    #[error("duration is too large to encode as nanoseconds")]
+    DurationOverflow,
+}

crates/irtt-client/src/event.rs

@@ -0,0 +1,41 @@
+use std::net::SocketAddr;
+
+use crate::{session::NegotiatedParams, timing::ClientTimestamp};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum OpenOutcome {
+    Started {
+        remote: SocketAddr,
+        token: u64,
+        negotiated: NegotiatedParams,
+        event: ClientEvent,
+    },
+    NoTestCompleted {
+        remote: SocketAddr,
+        negotiated: NegotiatedParams,
+        event: ClientEvent,
+    },
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum ClientEvent {
+    SessionStarted {
+        remote: SocketAddr,
+        token: u64,
+        negotiated: NegotiatedParams,
+        at: ClientTimestamp,
+    },
+    NoTestCompleted {
+        remote: SocketAddr,
+        negotiated: NegotiatedParams,
+        at: ClientTimestamp,
+    },
+    SessionClosed {
+        remote: SocketAddr,
+        token: u64,
+        at: ClientTimestamp,
+    },
+    Warning {
+        message: String,
+    },
+}

crates/irtt-client/src/session.rs

@@ -0,0 +1,54 @@
+use irtt_proto::{Params, PROTOCOL_VERSION};
+
+use crate::{config::NegotiationPolicy, error::ClientError};
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct NegotiatedParams {
+    pub params: Params,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum ClientState {
+    Connected,
+    Open { token: u64 },
+    NoTestCompleted,
+    Closed,
+}
+
+pub(crate) fn validate_negotiated_params(
+    requested: &Params,
+    returned: &Params,
+    policy: NegotiationPolicy,
+) -> Result<(), ClientError> {
+    if returned.protocol_version != PROTOCOL_VERSION {
+        return Err(ClientError::ProtocolVersionMismatch {
+            requested: PROTOCOL_VERSION,
+            received: returned.protocol_version,
+        });
+    }
+    if returned.duration_ns > requested.duration_ns {
+        return Err(ClientError::NegotiationRejected {
+            reason: "duration increased".to_owned(),
+        });
+    }
+    if returned.length > requested.length {
+        return Err(ClientError::NegotiationRejected {
+            reason: "length increased".to_owned(),
+        });
+    }
+    // TODO/OPEN: apply a documented negotiated interval safety floor once the
+    // clean spec or black-box report defines one. For now, only reject
+    // impossible non-positive intervals.
+    if returned.interval_ns <= 0 {
+        return Err(ClientError::NegotiationRejected {
+            reason: "interval must be positive".to_owned(),
+        });
+    }
+
+    if policy == NegotiationPolicy::Strict && returned != requested {
+        return Err(ClientError::NegotiationRejected {
+            reason: "returned params differ from requested params".to_owned(),
+        });
+    }
+    Ok(())
+}

crates/irtt-client/src/socket.rs

@@ -0,0 +1,103 @@
+use std::{
+    net::{SocketAddr, ToSocketAddrs, UdpSocket},
+    time::Duration,
+};
+
+use socket2::{Domain, Protocol, Socket, Type};
+
+use crate::{
+    config::{ClientConfig, SocketConfig, DEFAULT_PORT, MIN_OPEN_TIMEOUT},
+    error::ClientError,
+};
+
+pub(crate) fn validate_open_timeouts(timeouts: &[Duration]) -> Result<(), ClientError> {
+    for timeout in timeouts {
+        if *timeout < MIN_OPEN_TIMEOUT {
+            return Err(ClientError::OpenTimeoutTooSmall {
+                timeout: *timeout,
+                minimum: MIN_OPEN_TIMEOUT,
+            });
+        }
+    }
+    Ok(())
+}
+
+pub(crate) fn resolve_remote(config: &ClientConfig) -> Result<SocketAddr, ClientError> {
+    let addr = normalize_server_addr(&config.server_addr);
+    let mut addrs = addr
+        .to_socket_addrs()
+        .map_err(|_| ClientError::Resolve { addr: addr.clone() })?;
+    addrs
+        .find(|addr| {
+            (!config.socket_config.ipv4_only || addr.is_ipv4())
+                && (!config.socket_config.ipv6_only || addr.is_ipv6())
+        })
+        .ok_or(ClientError::Resolve { addr })
+}
+
+pub(crate) fn normalize_server_addr(addr: &str) -> String {
+    if addr.parse::<SocketAddr>().is_ok() {
+        return addr.to_owned();
+    }
+    if addr.starts_with('[') {
+        return addr.to_owned();
+    }
+    if addr.parse::<std::net::Ipv6Addr>().is_ok() {
+        return format!("[{addr}]:{DEFAULT_PORT}");
+    }
+    if addr
+        .rsplit_once(':')
+        .is_some_and(|(_, port)| port.parse::<u16>().is_ok())
+    {
+        return addr.to_owned();
+    }
+    format!("{addr}:{DEFAULT_PORT}")
+}
+
+pub(crate) fn connect_udp_socket(
+    config: &SocketConfig,
+    remote: SocketAddr,
+) -> Result<UdpSocket, ClientError> {
+    let domain = if remote.is_ipv4() {
+        Domain::IPV4
+    } else {
+        Domain::IPV6
+    };
+    let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
+
+    if config.ipv6_only && remote.is_ipv6() {
+        socket.set_only_v6(true)?;
+    }
+    if let Some(ttl) = config.ttl {
+        socket.set_ttl(ttl)?;
+    }
+
+    let bind_addr = config.bind_addr.unwrap_or_else(|| {
+        if remote.is_ipv4() {
+            SocketAddr::from(([0, 0, 0, 0], 0))
+        } else {
+            SocketAddr::from(([0_u16; 8], 0))
+        }
+    });
+    socket.bind(&bind_addr.into())?;
+    socket.connect(&remote.into())?;
+
+    let socket: UdpSocket = socket.into();
+    socket.set_read_timeout(config.recv_timeout)?;
+    Ok(socket)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn normalize_server_addr_adds_default_ports() {
+        assert_eq!(normalize_server_addr("127.0.0.1"), "127.0.0.1:2112");
+        assert_eq!(normalize_server_addr("127.0.0.1:1234"), "127.0.0.1:1234");
+        assert_eq!(normalize_server_addr("localhost"), "localhost:2112");
+        assert_eq!(normalize_server_addr("localhost:1234"), "localhost:1234");
+        assert_eq!(normalize_server_addr("::1"), "[::1]:2112");
+        assert_eq!(normalize_server_addr("[::1]:1234"), "[::1]:1234");
+    }
+}

crates/irtt-client/src/timing.rs

@@ -0,0 +1,16 @@
+use std::time::{Instant, SystemTime};
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct ClientTimestamp {
+    pub wall: SystemTime,
+    pub mono: Instant,
+}
+
+impl ClientTimestamp {
+    pub fn now() -> Self {
+        Self {
+            wall: SystemTime::now(),
+            mono: Instant::now(),
+        }
+    }
+}