feat(server): add multi-listener server set
`Server` answers on one address. A host that must serve several — an explicit IPv4 and IPv6 pair, or two interface addresses — needs several of them, and something has to own their construction, their shutdown and each other's failures. `ServerSet` is that owner, and deliberately not a protocol object: it holds no session state, no tokens and no clock, and nothing it does is visible on the wire. Each listener stays a complete `Server` with its own socket, core, session table and tokens, so a token issued by one is unknown at the others and every bound in the cloned `ServerConfig` — `max_sessions` included — applies per listener rather than per process. Construction binds every listener or none, preserving requested order and recording the addresses actually bound. `run` moves each listener into its own task, fans one shutdown future out over a watch channel, and joins them all: a listener that fails, or that returns before shutdown was asked for, shuts its siblings down and fails the set, because a service configured for two families must not continue silently as one. A genuine IPv6 listener sharing an explicit port with an IPv4 sibling is created `IPV6_V6ONLY` before it is bound, since a dual-stack socket would otherwise claim the IPv4 port too. That is the only pre-bind step, it uses safe `socket2` APIs, and an IPv4-mapped address is read as the IPv4 listener it is — never as the IPv6 half of a pair. Standalone `Server::bind` semantics are unchanged.
This commit is contained in:
4 files changed
+953
-5
No files matched your search
@@ -0,0 +1,467 @@
|
||||
//! One or more listeners run as a single service.
|
||||
//!
|
||||
//! Everything here is asserted from the outside, over real sockets, with the
|
||||
//! production encoders: what the listeners answer, what they refuse, and what
|
||||
//! survives one shared shutdown. The two properties worth the setup are that a
|
||||
//! set of listeners is genuinely several independent servers — separate tokens,
|
||||
//! separate session bounds — and that a set of *one* behaves exactly as a set of
|
||||
//! several does, because the server CLI runs every invocation through this path.
|
||||
|
||||
use std::{
|
||||
future::Future,
|
||||
io,
|
||||
net::{Ipv4Addr, Ipv6Addr, SocketAddr, UdpSocket as StdUdpSocket},
|
||||
time::Duration,
|
||||
};
|
||||
|
||||
use irtt_proto::{
|
||||
decode_echo_reply, decode_open_reply, encode_request, Clock, EchoReply, OpenReply, Params,
|
||||
ReceivedStats, RequestToEncode, StampAt,
|
||||
};
|
||||
use irtt_server::{ServerConfig, ServerSet, ServerSetError};
|
||||
use tokio::{net::UdpSocket, sync::oneshot, task::JoinHandle, time::timeout};
|
||||
|
||||
/// How long a reply is waited for.
|
||||
///
|
||||
/// Rejection is silence, so this doubles as the window a listener that must not
|
||||
/// answer is listened to: it only has to outlast a reply that would have come,
|
||||
/// and loopback replies take microseconds.
|
||||
const REPLY_TIMEOUT: Duration = Duration::from_secs(1);
|
||||
/// Every test is bounded, so a hang is a failure with a message rather than a
|
||||
/// suite that never finishes.
|
||||
const TEST_TIMEOUT: Duration = Duration::from_secs(10);
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn a_single_listener_set_binds_serves_and_stops() {
|
||||
// The CLI's ordinary case, and the reason it has no second code path: one
|
||||
// requested address, one resolved listener, one whole session over it.
|
||||
bounded(async {
|
||||
let set = ServerSet::bind([loopback(0)], unthrottled()).await.unwrap();
|
||||
assert_eq!(set.local_addrs().len(), 1);
|
||||
let addr = set.local_addrs()[0];
|
||||
assert_eq!(addr.ip(), Ipv4Addr::LOCALHOST);
|
||||
assert_ne!(addr.port(), 0, "a requested port of 0 is resolved");
|
||||
|
||||
let running = spawn(set);
|
||||
let mut peer = Peer::at(addr).await;
|
||||
let session = peer.open().await.expect("the listener answers an open");
|
||||
let reply = peer.echo(&session, 0).await.expect("and an echo");
|
||||
assert_eq!(reply.token, session.token);
|
||||
assert_eq!(reply.sequence, 0);
|
||||
peer.close(&session).await;
|
||||
assert!(
|
||||
peer.echo(&session, 1).await.is_none(),
|
||||
"a closed session is an unknown token"
|
||||
);
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn two_listeners_keep_their_order_and_serve_interleaved_traffic() {
|
||||
bounded(async {
|
||||
let set = ServerSet::bind([loopback(0), loopback(0)], unthrottled())
|
||||
.await
|
||||
.unwrap();
|
||||
let addrs = set.local_addrs().to_vec();
|
||||
assert_eq!(addrs.len(), 2);
|
||||
assert!(addrs.iter().all(|addr| addr.port() != 0));
|
||||
assert_ne!(
|
||||
addrs[0].port(),
|
||||
addrs[1].port(),
|
||||
"each ephemeral bind selects its own port"
|
||||
);
|
||||
|
||||
let running = spawn(set);
|
||||
let mut first = Peer::at(addrs[0]).await;
|
||||
let mut second = Peer::at(addrs[1]).await;
|
||||
|
||||
// Interleaved on purpose: both listener tasks have to be live at the
|
||||
// same time under one service, not served one after the other.
|
||||
let first_session = first.open().await.expect("the first listener opens");
|
||||
let second_session = second.open().await.expect("the second listener opens");
|
||||
for sequence in 0..2 {
|
||||
assert_eq!(
|
||||
first
|
||||
.echo(&first_session, sequence)
|
||||
.await
|
||||
.expect("the first listener echoes")
|
||||
.sequence,
|
||||
sequence
|
||||
);
|
||||
assert_eq!(
|
||||
second
|
||||
.echo(&second_session, sequence)
|
||||
.await
|
||||
.expect("the second listener echoes")
|
||||
.sequence,
|
||||
sequence
|
||||
);
|
||||
}
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn a_token_means_nothing_at_another_listener() {
|
||||
// Session namespaces are per listener because each listener owns its own
|
||||
// core. This is the externally visible consequence, and it is deliberate:
|
||||
// there is no process-wide token or session table to make a token portable.
|
||||
bounded(async {
|
||||
let set = ServerSet::bind([loopback(0), loopback(0)], unthrottled())
|
||||
.await
|
||||
.unwrap();
|
||||
let addrs = set.local_addrs().to_vec();
|
||||
let running = spawn(set);
|
||||
|
||||
let mut owner = Peer::at(addrs[0]).await;
|
||||
let session = owner.open().await.expect("the first listener opens");
|
||||
|
||||
let mut stranger = Peer::at(addrs[1]).await;
|
||||
assert!(
|
||||
stranger.echo(&session, 0).await.is_none(),
|
||||
"the second listener does not know a token it never issued"
|
||||
);
|
||||
assert!(
|
||||
owner.echo(&session, 0).await.is_some(),
|
||||
"and the listener that issued it still does"
|
||||
);
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn the_session_bound_applies_to_each_listener_separately() {
|
||||
// A cloned configuration is per-listener policy, not a process-wide budget:
|
||||
// a maximum of one session means one session *each*.
|
||||
bounded(async {
|
||||
let set = ServerSet::bind(
|
||||
[loopback(0), loopback(0)],
|
||||
unthrottled().with_max_sessions(1),
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
let addrs = set.local_addrs().to_vec();
|
||||
let running = spawn(set);
|
||||
|
||||
let mut first = Peer::at(addrs[0]).await;
|
||||
let mut second = Peer::at(addrs[1]).await;
|
||||
assert!(first.open().await.is_some());
|
||||
assert!(
|
||||
second.open().await.is_some(),
|
||||
"the second listener has its own session budget"
|
||||
);
|
||||
|
||||
let mut crowd = Peer::at(addrs[0]).await;
|
||||
assert!(
|
||||
crowd.open().await.is_none(),
|
||||
"a full listener refuses a second session silently"
|
||||
);
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn every_listener_serves_the_configured_policy() {
|
||||
// One representative policy is enough; the negotiation itself is covered
|
||||
// where it lives. What is under test is that the configuration reached every
|
||||
// listener, not what it does.
|
||||
bounded(async {
|
||||
let config = unthrottled().with_timestamp_allowance(irtt_server::TimestampAllowance::None);
|
||||
let set = ServerSet::bind([loopback(0), loopback(0)], config)
|
||||
.await
|
||||
.unwrap();
|
||||
let addrs = set.local_addrs().to_vec();
|
||||
let running = spawn(set);
|
||||
|
||||
for addr in addrs {
|
||||
let mut peer = Peer::at(addr).await;
|
||||
let session = peer.open().await.expect("the listener opens");
|
||||
assert_eq!(
|
||||
session.params.stamp_at,
|
||||
StampAt::None,
|
||||
"{addr} negotiated its own configuration"
|
||||
);
|
||||
}
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn one_shutdown_stops_every_listener_and_releases_its_socket() {
|
||||
bounded(async {
|
||||
let set = ServerSet::bind([loopback(0), loopback(0)], unthrottled())
|
||||
.await
|
||||
.unwrap();
|
||||
let addrs = set.local_addrs().to_vec();
|
||||
let running = spawn(set);
|
||||
|
||||
// Prove both are actually serving before asking them to stop.
|
||||
for &addr in &addrs {
|
||||
assert!(Peer::at(addr).await.open().await.is_some());
|
||||
}
|
||||
running.stop().await;
|
||||
|
||||
for addr in addrs {
|
||||
StdUdpSocket::bind(addr)
|
||||
.unwrap_or_else(|error| panic!("{addr} was not released: {error}"));
|
||||
}
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn a_listener_that_cannot_bind_leaves_the_whole_set_unbound() {
|
||||
// All or nothing: the set is built completely before anything is served, so
|
||||
// a failure part-way through releases what it had and starts no task.
|
||||
bounded(async {
|
||||
let occupied = UdpSocket::bind(loopback(0)).await.unwrap();
|
||||
let occupied_addr = occupied.local_addr().unwrap();
|
||||
let wanted = free_loopback_port();
|
||||
let wanted_addr = SocketAddr::from((Ipv4Addr::LOCALHOST, wanted));
|
||||
|
||||
let error = ServerSet::bind([wanted_addr, occupied_addr], unthrottled())
|
||||
.await
|
||||
.expect_err("the second listener cannot bind an occupied port");
|
||||
assert!(
|
||||
matches!(error, ServerSetError::ListenerSetup { addr, .. } if addr == occupied_addr),
|
||||
"the error names the listener that failed: {error}"
|
||||
);
|
||||
|
||||
StdUdpSocket::bind(wanted_addr)
|
||||
.expect("the listener bound before the failure was released again");
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn an_ipv4_and_a_genuine_ipv6_listener_share_one_port() {
|
||||
// The configuration this exists for. On a host whose IPv6 sockets are
|
||||
// dual-stack by default, the IPv6 bind would take the IPv4 port with it and
|
||||
// the pair could not coexist; the set binds that listener IPv6-only first.
|
||||
bounded(async {
|
||||
if !ipv6_loopback_available() {
|
||||
eprintln!("skipping the same-port test: IPv6 loopback is unavailable");
|
||||
return;
|
||||
}
|
||||
let port = free_loopback_port();
|
||||
let v4 = SocketAddr::from((Ipv4Addr::LOCALHOST, port));
|
||||
let v6 = SocketAddr::from((Ipv6Addr::LOCALHOST, port));
|
||||
|
||||
let set = ServerSet::bind([v4, v6], unthrottled()).await.unwrap();
|
||||
assert_eq!(set.local_addrs(), &[v4, v6][..]);
|
||||
let running = spawn(set);
|
||||
|
||||
for addr in [v4, v6] {
|
||||
let mut peer = Peer::at(addr).await;
|
||||
let session = peer.open().await.unwrap_or_else(|| panic!("{addr} opens"));
|
||||
assert!(peer.echo(&session, 0).await.is_some(), "{addr} echoes");
|
||||
}
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
/// The wildcard form of the same coexistence, where a wildcard listener is
|
||||
/// served at all. Reply source selection itself is covered by its own tests;
|
||||
/// this only asserts that both families bind on one port and answer.
|
||||
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
|
||||
#[tokio::test(flavor = "current_thread")]
|
||||
async fn wildcard_listeners_of_both_families_share_one_port() {
|
||||
bounded(async {
|
||||
if !ipv6_loopback_available() {
|
||||
eprintln!("skipping the wildcard same-port test: IPv6 loopback is unavailable");
|
||||
return;
|
||||
}
|
||||
let port = free_loopback_port();
|
||||
let v4: SocketAddr = SocketAddr::from((Ipv4Addr::UNSPECIFIED, port));
|
||||
let v6: SocketAddr = SocketAddr::from((Ipv6Addr::UNSPECIFIED, port));
|
||||
|
||||
let set = ServerSet::bind([v4, v6], unthrottled()).await.unwrap();
|
||||
let running = spawn(set);
|
||||
|
||||
for addr in [
|
||||
SocketAddr::from((Ipv4Addr::LOCALHOST, port)),
|
||||
SocketAddr::from((Ipv6Addr::LOCALHOST, port)),
|
||||
] {
|
||||
let mut peer = Peer::at(addr).await;
|
||||
let session = peer.open().await.unwrap_or_else(|| panic!("{addr} opens"));
|
||||
assert!(peer.echo(&session, 0).await.is_some(), "{addr} echoes");
|
||||
}
|
||||
|
||||
running.stop().await;
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
/// A minimal compliant client for one listener.
|
||||
///
|
||||
/// It uses the production encoders and decoders, and reads from an unconnected
|
||||
/// socket so that a reply from the wrong endpoint would be visible as a failed
|
||||
/// assertion rather than as silence.
|
||||
struct Peer {
|
||||
socket: UdpSocket,
|
||||
server: SocketAddr,
|
||||
buffer: Vec<u8>,
|
||||
}
|
||||
|
||||
impl Peer {
|
||||
async fn at(server: SocketAddr) -> Self {
|
||||
let bind = if server.is_ipv4() {
|
||||
SocketAddr::from((Ipv4Addr::LOCALHOST, 0))
|
||||
} else {
|
||||
SocketAddr::from((Ipv6Addr::LOCALHOST, 0))
|
||||
};
|
||||
Self {
|
||||
socket: UdpSocket::bind(bind).await.unwrap(),
|
||||
server,
|
||||
buffer: vec![0; 4096],
|
||||
}
|
||||
}
|
||||
|
||||
/// Opens a session, or reports the silence that means it was refused.
|
||||
async fn open(&mut self) -> Option<OpenReply> {
|
||||
let request = encode_request(
|
||||
RequestToEncode::Open {
|
||||
params: &requested_params(),
|
||||
no_test: false,
|
||||
},
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
let len = self.exchange(&request).await?;
|
||||
Some(decode_open_reply(&self.buffer[..len], None).unwrap())
|
||||
}
|
||||
|
||||
async fn echo(&mut self, session: &OpenReply, sequence: u32) -> Option<EchoReply> {
|
||||
let request = encode_request(
|
||||
RequestToEncode::Echo {
|
||||
token: session.token,
|
||||
sequence,
|
||||
params: &session.params,
|
||||
payload: &[],
|
||||
},
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
let len = self.exchange(&request).await?;
|
||||
Some(decode_echo_reply(&self.buffer[..len], &session.params, None).unwrap())
|
||||
}
|
||||
|
||||
async fn close(&mut self, session: &OpenReply) {
|
||||
let request = encode_request(
|
||||
RequestToEncode::Close {
|
||||
token: session.token,
|
||||
},
|
||||
None,
|
||||
)
|
||||
.unwrap();
|
||||
self.socket.send_to(&request, self.server).await.unwrap();
|
||||
}
|
||||
|
||||
/// Sends one request and returns the length of its reply, or `None` if the
|
||||
/// listener stayed silent.
|
||||
async fn exchange(&mut self, request: &[u8]) -> Option<usize> {
|
||||
self.socket.send_to(request, self.server).await.unwrap();
|
||||
match timeout(REPLY_TIMEOUT, self.socket.recv_from(&mut self.buffer)).await {
|
||||
Ok(received) => {
|
||||
let (len, from) = received.unwrap();
|
||||
assert_eq!(from, self.server, "a reply came from the wrong endpoint");
|
||||
Some(len)
|
||||
}
|
||||
Err(_) => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// The parameters every peer here asks for. Nothing in these tests depends on
|
||||
/// the values beyond their being a session a server will run.
|
||||
fn requested_params() -> Params {
|
||||
Params {
|
||||
protocol_version: 1,
|
||||
duration_ns: 10_000_000_000,
|
||||
interval_ns: 100_000_000,
|
||||
received_stats: ReceivedStats::Both,
|
||||
stamp_at: StampAt::Both,
|
||||
clock: Clock::Both,
|
||||
..Params::default()
|
||||
}
|
||||
}
|
||||
|
||||
fn unthrottled() -> ServerConfig {
|
||||
ServerConfig::default().with_min_send_interval(Duration::ZERO)
|
||||
}
|
||||
|
||||
fn loopback(port: u16) -> SocketAddr {
|
||||
SocketAddr::from((Ipv4Addr::LOCALHOST, port))
|
||||
}
|
||||
|
||||
/// A loopback port nothing currently holds.
|
||||
fn free_loopback_port() -> u16 {
|
||||
StdUdpSocket::bind(loopback(0))
|
||||
.unwrap()
|
||||
.local_addr()
|
||||
.unwrap()
|
||||
.port()
|
||||
}
|
||||
|
||||
fn ipv6_loopback_available() -> bool {
|
||||
match StdUdpSocket::bind(SocketAddr::from((Ipv6Addr::LOCALHOST, 0))) {
|
||||
Ok(_) => true,
|
||||
Err(error)
|
||||
if matches!(
|
||||
error.kind(),
|
||||
io::ErrorKind::AddrNotAvailable | io::ErrorKind::Unsupported
|
||||
) =>
|
||||
{
|
||||
false
|
||||
}
|
||||
Err(error) => panic!("unexpected IPv6 loopback bind failure: {error}"),
|
||||
}
|
||||
}
|
||||
|
||||
struct Running {
|
||||
shutdown: oneshot::Sender<()>,
|
||||
task: JoinHandle<Result<(), ServerSetError>>,
|
||||
}
|
||||
|
||||
impl Running {
|
||||
async fn stop(self) {
|
||||
self.shutdown.send(()).unwrap();
|
||||
self.task
|
||||
.await
|
||||
.expect("the supervising task completed")
|
||||
.expect("every listener stopped cleanly");
|
||||
}
|
||||
}
|
||||
|
||||
fn spawn(set: ServerSet) -> Running {
|
||||
let (shutdown, shutdown_rx) = oneshot::channel();
|
||||
let task = tokio::spawn(async move {
|
||||
set.run(async {
|
||||
let _ = shutdown_rx.await;
|
||||
})
|
||||
.await
|
||||
});
|
||||
Running { shutdown, task }
|
||||
}
|
||||
|
||||
async fn bounded<F>(test: F)
|
||||
where
|
||||
F: Future<Output = ()>,
|
||||
{
|
||||
timeout(TEST_TIMEOUT, test)
|
||||
.await
|
||||
.expect("the test exceeded its bounded runtime");
|
||||
}
|
||||
Reference in new issue
Block a user