Merge pull request #99 from Lochnair/refactor/server-receive-drop-policy

refactor(server): centralize receive drop policy
This commit was merged in pull request #99.
This commit is contained in:
Nils Andreas Svee authored and GitHub committed 2026-08-15 11:33:43 +07:00
commit 277fbfcb57
2 files changed
+95 -20

No files matched your search

+4 -7
View File
@@ -19,9 +19,10 @@ use crate::{
/// Fixed transport receive capacity. /// Fixed transport receive capacity.
/// ///
/// Standard UDP payloads fit in this buffer. A datagram reported as filling it /// Standard UDP payloads fit in this buffer; IPv6 jumbograms are not supported.
/// exactly is dropped conservatively instead of passing potentially truncated /// This is allocation policy only: whether a received datagram may reach the
/// bytes to the protocol core. IPv6 jumbograms are not supported. /// protocol core — the conservative rejection of one filling the buffer
/// included — belongs to [`socket_io::receive`].
const RECEIVE_BUFFER_LEN: usize = 65_536; const RECEIVE_BUFFER_LEN: usize = 65_536;
const MAINTENANCE_INTERVAL: Duration = Duration::from_secs(1); const MAINTENANCE_INTERVAL: Duration = Duration::from_secs(1);
@@ -218,10 +219,6 @@ impl Server {
Err(source) => return Err(ServerRuntimeError::Receive { source }), Err(source) => return Err(ServerRuntimeError::Receive { source }),
}; };
let (len, peer) = (received.len, received.peer); let (len, peer) = (received.len, received.peer);
if len == RECEIVE_BUFFER_LEN {
continue;
}
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.prepare_reply_traffic_class(reply.traffic_class()) { if !self.prepare_reply_traffic_class(reply.traffic_class()) {
continue; continue;
+91 -13
View File
@@ -90,10 +90,23 @@ pub(crate) fn configure_destination_metadata(
/// `select_reply_source` is the listener's own wildcard-ness, decided once at /// `select_reply_source` is the listener's own wildcard-ness, decided once at
/// construction. It is never inferred from a peer address. /// construction. It is never inferred from a peer address.
/// ///
/// `Ok(None)` is a datagram the runtime must drop without letting it reach the /// This is the one place deciding whether a datagram may reach
/// core: it was truncated, its ancillary metadata was truncated, or a wildcard /// [`ServerCore`](crate::ServerCore). `Ok(Some(..))` is a datagram the transport
/// listener could not recover its local destination. Answering it correctly is /// layer accepts; `Ok(None)` is one the caller must consume and drop without
/// impossible, so it must not advance any session's state either. /// advancing any protocol or session state, because answering it correctly is
/// impossible. `Ok(None)` covers:
///
/// - payload truncation reported by an ancillary receive (`MSG_TRUNC`);
/// - ancillary metadata truncation (`MSG_CTRUNC`);
/// - a wildcard listener's missing or unusable local destination;
/// - a datagram filling `buffer` exactly, which is conservatively treated as
/// potentially truncated.
///
/// That last rule exists because `recv_from` reports no truncation flag, so an
/// exactly-full result cannot be told apart from a larger datagram cut down to
/// the supplied capacity. It is expressed against `buffer.len()`, so the policy
/// follows whatever receive capacity the caller supplies, and it applies to both
/// receive paths alike.
/// ///
/// Cancel-safe: nothing is consumed until a datagram is fully received. /// Cancel-safe: nothing is consumed until a datagram is fully received.
pub(crate) async fn receive( pub(crate) async fn receive(
@@ -101,16 +114,19 @@ pub(crate) async fn receive(
buffer: &mut [u8], buffer: &mut [u8],
select_reply_source: bool, select_reply_source: bool,
) -> io::Result<Option<ReceivedDatagram>> { ) -> io::Result<Option<ReceivedDatagram>> {
if select_reply_source { let capacity = buffer.len();
return receive_with_destination(socket, buffer).await; let received = if select_reply_source {
} receive_with_destination(socket, buffer).await?
} else {
let (len, peer) = socket.recv_from(buffer).await?;
Some(ReceivedDatagram {
len,
peer,
reply_source: ReplySource::Bound,
})
};
let (len, peer) = socket.recv_from(buffer).await?; Ok(received.filter(|datagram| datagram.len < capacity))
Ok(Some(ReceivedDatagram {
len,
peer,
reply_source: ReplySource::Bound,
}))
} }
#[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))] #[cfg(any(target_os = "linux", target_os = "macos", target_os = "freebsd"))]
@@ -155,3 +171,65 @@ fn unsupported() -> io::Error {
"reply source-address selection is unsupported on this target", "reply source-address selection is unsupported on this target",
) )
} }
#[cfg(test)]
mod tests {
use std::net::Ipv4Addr;
use super::*;
/// Sends `payload` to a fresh explicit-address listener and returns what the
/// transport layer decided about it, received into a `capacity`-byte buffer.
///
/// The buffer is deliberately tiny: the full-buffer rule is written against
/// the caller's capacity, so it is provable without a 65,536-byte datagram.
async fn receive_one(
capacity: usize,
payload: &[u8],
) -> (io::Result<Option<ReceivedDatagram>>, Vec<u8>) {
let listener = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
let sender = UdpSocket::bind((Ipv4Addr::LOCALHOST, 0)).await.unwrap();
sender
.send_to(payload, listener.local_addr().unwrap())
.await
.unwrap();
let mut buffer = vec![0; capacity];
let received = receive(&listener, &mut buffer, false).await;
(received, buffer)
}
#[tokio::test(flavor = "current_thread")]
async fn a_datagram_shorter_than_the_buffer_reaches_the_core_intact() {
let payload = b"irtt";
let (received, buffer) = receive_one(16, payload).await;
let received = received
.unwrap()
.expect("a datagram that cannot have been truncated is accepted");
assert_eq!(received.len, payload.len());
assert_eq!(received.peer.ip(), Ipv4Addr::LOCALHOST);
assert_eq!(received.reply_source, ReplySource::Bound);
assert_eq!(&buffer[..received.len], payload);
}
/// An exactly-full receive is ambiguous, not obviously fine: `recv_from`
/// reports no truncation flag, so this datagram is indistinguishable from a
/// larger one cut down to the same capacity. Both are dropped.
#[tokio::test(flavor = "current_thread")]
async fn a_datagram_filling_the_buffer_is_dropped_as_possibly_truncated() {
let (received, _) = receive_one(4, b"irtt").await;
assert!(received.unwrap().is_none());
}
/// What matters is that an oversized datagram never reaches the core, not
/// how the platform says so: Unix truncates it to the buffer, which the
/// full-buffer rule then drops, while Windows fails the receive outright
/// with `WSAEMSGSIZE`. Asserting the shared guarantee keeps this test true
/// on both.
#[tokio::test(flavor = "current_thread")]
async fn a_datagram_larger_than_the_buffer_never_reaches_the_core() {
let (received, _) = receive_one(4, b"irtt-rs").await;
assert!(!matches!(received, Ok(Some(_))));
}
}