fix(server): keep unmarkable listeners serving

This commit is contained in:
Nils Andreas Svee committed 2026-08-13 22:26:58 +07:00
1 parent 5420387846
commit 1830429ce5
3 files changed
+112 -22

No files matched your search

+16 -2
View File
@@ -194,12 +194,26 @@ session policy lives.
suspended. Do not introduce concurrent sends without replacing this with suspended. Do not introduce concurrent sends without replacing this with
per-packet control messages. per-packet control messages.
- Applying the class and sending are both per-packet transport steps: a failure - Applying the class and sending are both per-packet transport steps: a failure
of either drops that one reply and the listener keeps serving. A class that of either drops that one reply and the listener keeps serving. A marking that
could not be applied means the reply is not sent at all, because sending it could not be applied means the reply is not sent at all, because sending it
would put it on the wire under the previous reply's marking. would put it on the wire under the previous reply's marking.
- **One exception, and it is what keeps the server usable on hosts without the
option**: an *unmarked* reply may still be sent when this server has never
successfully applied a nonzero class to the socket, because then there is no
marking of ours for it to inherit. Some Windows builds do not support
`IP_TOS`, and a few targets have no safe setter at all; without this, such a
host would drop every open reply and no session could ever start. The flag
behind it is consulted only on failure — it never elides a call.
- The platform helper must report failure honestly, including for zero. "The
socket was cleared" and "nothing was written" are different statements, and a
socket passed to `Server::from_socket` may have been marked by its creator.
Whether an unappliable zero is nevertheless safe is the runtime's decision,
made from what this server itself applied.
- The option itself comes from `socket2`'s safe API, chosen by the listener's - The option itself comes from `socket2`'s safe API, chosen by the listener's
bound address family rather than the peer's. The crate forbids unsafe code; bound address family rather than the peer's. The crate forbids unsafe code;
there is no raw `setsockopt` and no `sendmsg`/`cmsg` machinery. there is no raw `setsockopt` and no `sendmsg`/`cmsg` machinery. Keep the
target lists identical to the pinned `socket2` version's own gates — check
its source rather than copying another crate's older matrix.
## Rate, lifetime and server-initiated close ## Rate, lifetime and server-initiated close
+81 -6
View File
@@ -44,6 +44,14 @@ pub struct Server {
/// queried at construction, in which case the send path asks again rather /// queried at construction, in which case the send path asks again rather
/// than guessing a family. /// than guessing a family.
listener_is_ipv4: Option<bool>, listener_is_ipv4: Option<bool>,
/// Whether this server has ever successfully applied a nonzero traffic
/// class to the socket.
///
/// This is not a cache and never elides a call — the class is applied
/// before every send regardless. It is consulted only when applying one
/// *fails*, to tell "the socket may be carrying a marking of ours" from
/// "it cannot be carrying one".
socket_is_marked: bool,
} }
impl Server { impl Server {
@@ -65,16 +73,36 @@ impl Server {
core: ServerCore::new(config), core: ServerCore::new(config),
recv_buffer: vec![0; RECEIVE_BUFFER_LEN], recv_buffer: vec![0; RECEIVE_BUFFER_LEN],
listener_is_ipv4, listener_is_ipv4,
socket_is_marked: false,
} }
} }
/// Applies one reply's raw traffic class to the listener socket. /// Applies one reply's raw traffic class to the listener socket, and
/// reports whether that reply may be sent.
/// ///
/// This runs before *every* send, and zero is applied as deliberately as /// The class is applied before *every* send, and zero as deliberately as
/// any other value: a listener serves many sessions from one socket, so /// any other value: a listener serves many sessions from one socket, so
/// skipping the call for an unmarked reply — an open reply, or a session /// skipping the call for an unmarked reply — an open reply, or a session
/// that negotiated nothing — would send it under whichever marking the /// that negotiated nothing — would send it under whichever marking the
/// previous reply left behind. /// previous reply left behind.
///
/// When the option cannot be applied, [`may_send_unappliable`] decides. A
/// marking this server asked for and did not get must never be replaced by
/// silently sending under the previous one, so those replies are dropped;
/// but a host that refuses the option outright — some Windows builds do not
/// support `IP_TOS`, and a few targets have no safe setter at all — must
/// still be able to run a server, and it can, because a socket this server
/// has never marked has no marking of ours to clear.
fn prepare_reply_traffic_class(&mut self, traffic_class: u8) -> bool {
match self.apply_reply_traffic_class(traffic_class) {
Ok(()) => {
self.socket_is_marked = traffic_class != 0;
true
}
Err(_) => may_send_unappliable(traffic_class, self.socket_is_marked),
}
}
fn apply_reply_traffic_class(&self, traffic_class: u8) -> io::Result<()> { fn apply_reply_traffic_class(&self, traffic_class: u8) -> io::Result<()> {
let is_ipv4 = match self.listener_is_ipv4 { let is_ipv4 = match self.listener_is_ipv4 {
Some(is_ipv4) => is_ipv4, Some(is_ipv4) => is_ipv4,
@@ -95,13 +123,16 @@ impl Server {
/// Graceful shutdown sends no session-close packets and leaves no hidden /// Graceful shutdown sends no session-close packets and leaves no hidden
/// receive task behind, even while a reply send is waiting for socket /// receive task behind, even while a reply send is waiting for socket
/// writability. Per-packet send failures and short sends drop only that /// writability. Per-packet send failures and short sends drop only that
/// reply, and so does a failure to apply that reply's traffic class — /// reply, and so does a failure to apply a marking that reply needed —
/// preparing one packet is not a reason to stop serving every other /// preparing one packet is not a reason to stop serving every other
/// session, and the next reply makes its own independent attempt. Receive /// session, and the next reply makes its own independent attempt. Receive
/// failures and internal core failures terminate the loop with an error. /// failures and internal core failures terminate the loop with an error.
/// ///
/// A reply whose class could not be applied is never sent anyway: sending it /// A reply whose marking could not be applied is not sent instead: sending
/// would put it on the wire under the previous reply's marking. /// it would put it on the wire under the previous reply's marking. An
/// *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 still runs
/// a working server. See [`Server::prepare_reply_traffic_class`].
pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError> pub async fn run<F>(&mut self, shutdown: F) -> Result<(), ServerRuntimeError>
where where
F: Future<Output = ()>, F: Future<Output = ()>,
@@ -126,7 +157,7 @@ impl Server {
} }
if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? { if let Some(reply) = self.core.handle_datagram(peer, &self.recv_buffer[..len])? {
if self.apply_reply_traffic_class(reply.traffic_class()).is_err() { if !self.prepare_reply_traffic_class(reply.traffic_class()) {
continue; continue;
} }
let send = self.socket.send_to(reply.bytes(), peer); let send = self.socket.send_to(reply.bytes(), peer);
@@ -148,6 +179,21 @@ impl Server {
} }
} }
/// Whether a reply may still be sent after its traffic class could not be
/// applied.
///
/// Only an unmarked reply on a socket this server has never successfully marked
/// qualifies. Both halves matter: a reply that wanted a marking must not go out
/// without one, and once this server has put a nonzero class on the socket, a
/// failure to clear it means the socket may still be carrying it.
///
/// A socket handed to [`Server::from_socket`] pre-marked by its creator is
/// outside what this can know; explicit marking is the server's own to manage
/// from a fresh listener.
fn may_send_unappliable(traffic_class: u8, socket_is_marked: bool) -> bool {
traffic_class == 0 && !socket_is_marked
}
/// Failure to create or run a Tokio UDP server. /// Failure to create or run a Tokio UDP server.
#[derive(Debug, Error)] #[derive(Debug, Error)]
pub enum ServerRuntimeError { pub enum ServerRuntimeError {
@@ -323,4 +369,33 @@ mod tests {
fn unthrottled() -> ServerConfig { fn unthrottled() -> ServerConfig {
ServerConfig::default().with_min_send_interval(Duration::ZERO) ServerConfig::default().with_min_send_interval(Duration::ZERO)
} }
/// The rule that decides what an unappliable traffic class means, which no
/// normal interface can reach: it needs a host that refuses the socket
/// option, and the hosts these tests run on do not.
#[test]
fn only_an_unmarked_reply_on_a_never_marked_socket_survives_an_apply_failure() {
for (traffic_class, socket_is_marked, sendable, why) in [
(
0,
false,
true,
"no marking wanted, and none of ours to clear",
),
(0, true, false, "our own marking may still be on the socket"),
(
0xb8,
false,
false,
"a marking that was wanted but not applied",
),
(0xb8, true, false, "and likewise over a marking already set"),
] {
assert_eq!(
may_send_unappliable(traffic_class, socket_is_marked),
sendable,
"{why}"
);
}
}
} }
+15 -14
View File
@@ -34,8 +34,8 @@ pub(crate) fn set_reply_traffic_class(
target_os = "fuchsia", target_os = "fuchsia",
target_os = "redox", target_os = "redox",
target_os = "solaris", target_os = "solaris",
target_os = "illumos",
target_os = "haiku", target_os = "haiku",
target_os = "wasi",
)))] )))]
fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> { fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
socket.set_tos_v4(traffic_class) socket.set_tos_v4(traffic_class)
@@ -45,8 +45,8 @@ fn set_ipv4_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result
target_os = "fuchsia", target_os = "fuchsia",
target_os = "redox", target_os = "redox",
target_os = "solaris", target_os = "solaris",
target_os = "illumos",
target_os = "haiku", target_os = "haiku",
target_os = "wasi",
))] ))]
fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> { fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
unsupported_traffic_class(traffic_class, "IPv4 TOS socket options") unsupported_traffic_class(traffic_class, "IPv4 TOS socket options")
@@ -62,6 +62,7 @@ fn set_ipv4_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Resul
target_os = "netbsd", target_os = "netbsd",
target_os = "openbsd", target_os = "openbsd",
target_os = "cygwin", target_os = "cygwin",
target_os = "illumos",
))] ))]
fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> { fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
socket.set_tclass_v6(traffic_class) socket.set_tclass_v6(traffic_class)
@@ -77,6 +78,7 @@ fn set_ipv6_traffic_class(socket: SockRef<'_>, traffic_class: u32) -> io::Result
target_os = "netbsd", target_os = "netbsd",
target_os = "openbsd", target_os = "openbsd",
target_os = "cygwin", target_os = "cygwin",
target_os = "illumos",
)))] )))]
fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> { fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Result<()> {
unsupported_traffic_class(traffic_class, "IPv6 Traffic Class socket options") unsupported_traffic_class(traffic_class, "IPv6 Traffic Class socket options")
@@ -84,18 +86,20 @@ fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Resul
/// A target where the safe API cannot express this option. /// A target where the safe API cannot express this option.
/// ///
/// Requesting *unmarked* still succeeds, because a platform that can never /// This reports [`io::ErrorKind::Unsupported`] for *every* value, zero
/// apply a nonzero class also never left one on the socket for the next reply to /// included. Returning success for zero would claim the socket had been cleared
/// inherit — the invariant the explicit zero exists to protect. Asking for a /// when nothing was written to it, which is not the same statement — a caller
/// real marking reports [`io::ErrorKind::Unsupported`] rather than pretending: /// handing [`Server::from_socket`](crate::Server::from_socket) an already
/// the runtime drops that one reply and stays alive, and no reply ever goes out /// prepared socket may have marked it by some other means. Whether an
/// carrying a class the server did not intend. /// unappliable zero is nevertheless safe to send is the runtime's decision, and
/// it is made from what this server itself has applied; this helper's job is
/// only to report honestly.
#[cfg(any( #[cfg(any(
target_os = "fuchsia", target_os = "fuchsia",
target_os = "redox", target_os = "redox",
target_os = "solaris", target_os = "solaris",
target_os = "illumos",
target_os = "haiku", target_os = "haiku",
target_os = "wasi",
not(any( not(any(
target_os = "android", target_os = "android",
target_os = "dragonfly", target_os = "dragonfly",
@@ -106,15 +110,12 @@ fn set_ipv6_traffic_class(_socket: SockRef<'_>, traffic_class: u32) -> io::Resul
target_os = "netbsd", target_os = "netbsd",
target_os = "openbsd", target_os = "openbsd",
target_os = "cygwin", target_os = "cygwin",
target_os = "illumos",
)), )),
))] ))]
fn unsupported_traffic_class(traffic_class: u32, option: &'static str) -> io::Result<()> { fn unsupported_traffic_class(_traffic_class: u32, option: &'static str) -> io::Result<()> {
if traffic_class == 0 {
Ok(())
} else {
Err(io::Error::new( Err(io::Error::new(
io::ErrorKind::Unsupported, io::ErrorKind::Unsupported,
format!("{option} are unsupported on this target"), format!("{option} are unsupported on this target"),
)) ))
}
} }