From c859f4fd76807cfc76ec78c6a1fcdfb078b016bc Mon Sep 17 00:00:00 2001 From: Randell Jesup Date: Fri, 4 Sep 2026 05:31:05 +0000 Subject: [PATCH] WebTransport: count a session's expired outgoing datagrams Implement Protocol::record_expired_outgoing_datagrams on the WebTransport session, so SessionStats::datagrams_expired_outgoing counts datagrams shed for exceeding outgoingMaxAge; the field existed but stayed zero. The HTTP/3 layer picks the counts up in its per-session sweep, and close_session collects any still pending before returning the final stats. A session the peer closes drops its queue as soon as it leaves the active state; a close capsule without a FIN would otherwise leave the queue in place until the FIN arrives. Datagrams still queued when a session closes are dropped, not counted as expired. --- neqo-http3/src/connection.rs | 36 +++++- .../src/features/extended_connect/session.rs | 45 +++++++- .../src/features/extended_connect/stats.rs | 4 +- .../tests/webtransport/datagrams.rs | 107 +++++++++++++++++- .../extended_connect/webtransport_session.rs | 4 + neqo-http3/src/webtransport.rs | 79 ++++++++++++- 6 files changed, 258 insertions(+), 17 deletions(-) diff --git a/neqo-http3/src/connection.rs b/neqo-http3/src/connection.rs index 25dd561586..d38136732b 100644 --- a/neqo-http3/src/connection.rs +++ b/neqo-http3/src/connection.rs @@ -4,8 +4,6 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -#[cfg(test)] -use std::num::NonZeroUsize; use std::{ cell::RefCell, fmt::{self, Debug, Display, Formatter}, @@ -13,11 +11,15 @@ use std::{ rc::Rc, time::Instant, }; +#[cfg(test)] +use std::{num::NonZeroUsize, time::Duration}; use neqo_common::{ Bytes, Decoder, Header, MessageType, Role, qdebug, qerror, qinfo, qtrace, qwarn, }; use neqo_qpack as qpack; +#[cfg(test)] +use neqo_transport::DatagramQueueCapacity; use neqo_transport::{ AppError, CloseReason, Connection, DatagramQueueOutcome, DatagramTracking, State, StreamId, StreamType, ZeroRttState, @@ -1659,6 +1661,34 @@ impl Http3Connection { Ok(()) } + /// Test-only; see [`Self::extended_connect_set_max_buffered_datagrams`]. + #[cfg(test)] + pub(crate) fn webtransport_session_set_datagram_max_age( + &self, + session_id: StreamId, + conn: &mut Connection, + max_age: Option, + now: Instant, + ) -> Res<()> { + self.webtransport_session(session_id)? + .borrow() + .set_datagram_max_age(conn, max_age, now); + Ok(()) + } + + /// Test-only; see [`Self::extended_connect_set_max_buffered_datagrams`]. + #[cfg(test)] + pub(crate) fn extended_connect_datagram_queue_capacity( + &self, + session_id: StreamId, + conn: &Connection, + ) -> Res { + Ok(self + .validate_extended_connect_session(session_id)? + .borrow() + .datagram_queue_capacity(conn)) + } + /// Expire stale outgoing datagrams on every active extended-CONNECT /// session's queue and count them per session. Called once per /// `process_http3` tick. The transport counts expiries on the queue @@ -1860,6 +1890,8 @@ impl Http3Connection { wt: &Rc>, conn: &mut Connection, ) { + wt.borrow().drop_queued_datagrams(conn); + let (recv, send) = wt.borrow_mut().take_sub_streams(); #[expect( diff --git a/neqo-http3/src/features/extended_connect/session.rs b/neqo-http3/src/features/extended_connect/session.rs index 1edab6c54d..c090e1ad00 100644 --- a/neqo-http3/src/features/extended_connect/session.rs +++ b/neqo-http3/src/features/extended_connect/session.rs @@ -4,8 +4,6 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -#[cfg(test)] -use std::num::NonZeroUsize; use std::{ cell::RefCell, fmt::{self, Debug, Display, Formatter}, @@ -13,8 +11,12 @@ use std::{ str::from_utf8, time::Instant, }; +#[cfg(test)] +use std::{num::NonZeroUsize, time::Duration}; use neqo_common::{Bytes, Encoder, Header, MessageType, Role, qdebug, qtrace}; +#[cfg(test)] +use neqo_transport::DatagramQueueCapacity; use neqo_transport::{ AppError, Connection, DatagramQueueOutcome, DatagramTracking, StreamId, streams::{SendGroupId, SendOrder}, @@ -377,11 +379,21 @@ impl Session { &mut self.control_stream_recv, now, )? { - self.state = new_state; + self.enter_state(new_state, conn, now); } Ok(()) } + /// Change state; on the first move into a closing state, expire and drop + /// this session's queued datagrams, as none of them can be sent anymore. + fn enter_state(&mut self, state: State, conn: &mut Connection, now: Instant) { + if state.closing_state() && !self.state.closing_state() { + self.expire_datagrams(conn, now); + self.drop_queued_datagrams(conn); + } + self.state = state; + } + /// # Errors /// /// Return an error if the stream was closed on the transport layer, but that information is not @@ -394,8 +406,7 @@ impl Session { now: Instant, ) -> Res<()> { qdebug!("[{self}]: close_session"); - self.state = State::Done; - conn.drop_session_datagrams(self.id); + self.enter_state(State::Done, conn, now); if let Some(close_frame) = self.protocol.close_frame(error, message) { self.control_stream_send @@ -509,6 +520,24 @@ impl Session { conn.set_max_buffered_datagrams(self.id, mark); } + /// Test-only; see `Http3Connection::extended_connect_set_max_buffered_datagrams`. + #[cfg(test)] + pub(crate) fn set_datagram_max_age( + &self, + conn: &mut Connection, + max_age: Option, + now: Instant, + ) { + conn.set_datagram_max_age(self.id, max_age, now); + } + + /// Test-only; see `Http3Connection::extended_connect_set_max_buffered_datagrams`. + #[cfg(test)] + #[must_use] + pub(crate) fn datagram_queue_capacity(&self, conn: &Connection) -> DatagramQueueCapacity { + conn.datagram_queue_capacity(self.id) + } + /// Expire stale queued datagrams, counting them against this session's /// stats, and return how many expired. See /// `Connection::expire_session_datagrams`. @@ -523,6 +552,12 @@ impl Session { expired } + /// Drop this session's queue, which lives on `conn` and would otherwise + /// outlive the session. + pub(crate) fn drop_queued_datagrams(&self, conn: &mut Connection) { + conn.drop_session_datagrams(self.id); + } + pub(crate) fn datagram(&self, datagram: Bytes) { if self.state != State::Active { qdebug!("[{self}]: received datagram on {:?} session.", self.state); diff --git a/neqo-http3/src/features/extended_connect/stats.rs b/neqo-http3/src/features/extended_connect/stats.rs index b840196210..053da52544 100644 --- a/neqo-http3/src/features/extended_connect/stats.rs +++ b/neqo-http3/src/features/extended_connect/stats.rs @@ -24,8 +24,6 @@ )] #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct SessionStats { - /// Outgoing datagrams that expired before being sent. - /// - /// Currently always zero; populated once datagram expiry is wired up. + /// Outgoing datagrams that expired (per `outgoingMaxAge`) before being sent. pub datagrams_expired_outgoing: u64, } diff --git a/neqo-http3/src/features/extended_connect/tests/webtransport/datagrams.rs b/neqo-http3/src/features/extended_connect/tests/webtransport/datagrams.rs index 1eeb3132d3..7006cf29ac 100644 --- a/neqo-http3/src/features/extended_connect/tests/webtransport/datagrams.rs +++ b/neqo-http3/src/features/extended_connect/tests/webtransport/datagrams.rs @@ -4,7 +4,7 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -use std::num::NonZeroUsize; +use std::{num::NonZeroUsize, time::Duration}; use neqo_common::{Encoder, event::Provider as _, to_u64}; use neqo_transport::{ConnectionParameters, DatagramQueueOutcome, streams::SendGroupId}; @@ -15,6 +15,7 @@ use crate::{ features::extended_connect::tests::webtransport::{ DATAGRAM_SIZE, WtTest, wt_default_parameters, }, + frames::WebTransportFrame, webtransport::{ClientSession as _, ServerEvent, ServerSession}, }; @@ -87,6 +88,33 @@ fn max_datagram_size_smaller_than_session_prefix() { assert_eq!(wt.max_datagram_size(wt_session.stream_id()), Ok(0)); } +#[test] +fn datagram_expires_before_being_sent() { + let mut wt = WtTest::new(); + let wt_session = wt.create_wt_session(); + let t0 = now(); + + wt_session.set_datagram_max_age(Some(Duration::from_millis(5)), t0); + assert_eq!( + wt_session.send_datagram(DGRAM, Some(1), t0, SendGroupId::new(0), 0), + Ok(DatagramQueueOutcome::Ok) + ); + assert_eq!(wt_session.datagram_queue_capacity().queued_datagrams, 1); + + // No packets ever need to be built in between: expiry must not wait on + // that. Driving the server's own HTTP/3 tick (not exchange_packets, + // which uses its own clock) is enough on its own. + let later = t0 + Duration::from_millis(10); + drop(wt.server.process_output(later)); + + assert_eq!( + wt_session.datagram_queue_capacity().queued_datagrams, + 0, + "the stale datagram must be gone before it is ever handed to the QUIC layer" + ); + assert_eq!(wt_session.stats().datagrams_expired_outgoing, 1); +} + #[test] fn datagram_larger_than_peers_limit_is_rejected_synchronously() { let mut wt = WtTest::new(); @@ -102,6 +130,7 @@ fn datagram_larger_than_peers_limit_is_rejected_synchronously() { Err(crate::Error::Transport(neqo_transport::Error::TooMuchData)), "an oversized datagram must fail before ever reaching the queue" ); + assert_eq!(wt_session.datagram_queue_capacity().queued_datagrams, 0); } /// The other side of the boundary: a datagram of exactly `max_datagram_size` @@ -120,6 +149,7 @@ fn datagram_of_exactly_the_peers_limit_is_accepted() { wt_session.send_datagram(&largest, Some(1), now(), SendGroupId::new(0), 0), Ok(DatagramQueueOutcome::Ok) ); + assert_eq!(wt_session.datagram_queue_capacity().queued_datagrams, 1); } #[test] @@ -199,6 +229,81 @@ fn outgoing_datagram_space_available_forwarded() { ); } +#[test] +fn close_session_stats_include_datagrams_expired_since_the_last_sweep() { + let mut wt = WtTest::new(); + let wt_session = wt.create_wt_session(); + let t0 = now(); + + wt_session.set_datagram_max_age(Some(Duration::from_millis(5)), t0); + assert_eq!( + wt_session.send_datagram(DGRAM, Some(1), t0, SendGroupId::new(0), 0), + Ok(DatagramQueueOutcome::Ok) + ); + + let stats = wt_session + .close_session(0, "", t0 + Duration::from_millis(10)) + .unwrap(); + assert_eq!(stats.datagrams_expired_outgoing, 1); +} + +/// A client that resets the session takes the server's queued datagrams +/// with it. +#[test] +fn session_reset_by_client_drops_the_servers_queued_datagrams() { + let mut wt = WtTest::new(); + let wt_session = wt.create_wt_session(); + let session_id = wt_session.stream_id(); + + // One datagram per packet, so the server's reply cannot flush them all. + let largest = vec![0; usize::try_from(wt_session.max_datagram_size().unwrap()).unwrap()]; + for id in 0..3 { + assert_eq!( + wt_session.send_datagram(&largest, Some(id), now(), SendGroupId::new(0), 0), + Ok(DatagramQueueOutcome::Ok) + ); + } + assert_eq!(wt_session.transport_queued_datagrams(), 3); + wt.client + .cancel_fetch(session_id, crate::Error::HttpNone.code()) + .unwrap(); + let reset = wt.client.process_output(now()).dgram().unwrap(); + drop(wt.server.process(Some(reset), now())); + + assert_eq!(wt_session.transport_queued_datagrams(), 0); +} + +/// A close capsule without a FIN leaves the server's session in `FinPending` +/// until the FIN arrives, which may be never; its queue must not wait for it. +#[test] +fn close_capsule_without_fin_drops_the_servers_queued_datagrams() { + let mut wt = WtTest::new(); + let wt_session = wt.create_wt_session(); + let session_id = wt_session.stream_id(); + let t0 = now(); + + // One datagram per packet, so the server's reply cannot flush them all. + let largest = vec![0; usize::try_from(wt_session.max_datagram_size().unwrap()).unwrap()]; + for id in 0..3 { + assert_eq!( + wt_session.send_datagram(&largest, Some(id), t0, SendGroupId::new(0), 0), + Ok(DatagramQueueOutcome::Ok) + ); + } + + let mut enc = Encoder::default(); + WebTransportFrame::CloseSession { + error: 0, + message: String::new(), + } + .encode(&mut enc); + wt.client.send_data(session_id, enc.as_ref(), t0).unwrap(); + let close = wt.client.process_output(t0).dgram().unwrap(); + drop(wt.server.process(Some(close), t0)); + + assert_eq!(wt_session.transport_queued_datagrams(), 0); +} + #[test] fn datagram_send_order_controls_priority() { let mut wt = WtTest::new(); diff --git a/neqo-http3/src/features/extended_connect/webtransport_session.rs b/neqo-http3/src/features/extended_connect/webtransport_session.rs index d7ab363b9d..e648790caa 100644 --- a/neqo-http3/src/features/extended_connect/webtransport_session.rs +++ b/neqo-http3/src/features/extended_connect/webtransport_session.rs @@ -247,6 +247,10 @@ impl Protocol for Session { Some(&self.stats) } + fn record_expired_outgoing_datagrams(&mut self, count: u64) { + self.stats.datagrams_expired_outgoing += count; + } + fn register_send_group(&mut self, id: SendGroupId) -> Res<()> { Self::register_send_group(self, id) } diff --git a/neqo-http3/src/webtransport.rs b/neqo-http3/src/webtransport.rs index 321b1b727f..7febc6a321 100644 --- a/neqo-http3/src/webtransport.rs +++ b/neqo-http3/src/webtransport.rs @@ -4,8 +4,6 @@ // option. This file may not be copied, modified, or distributed // except according to those terms. -#[cfg(test)] -use std::num::NonZeroUsize; use std::{ cell::RefCell, fmt::{self, Display, Formatter}, @@ -13,8 +11,12 @@ use std::{ rc::Rc, time::Instant, }; +#[cfg(test)] +use std::{num::NonZeroUsize, time::Duration}; use neqo_common::{Bytes, Encoder, Header, qdebug, qinfo, qtrace, to_u64}; +#[cfg(test)] +use neqo_transport::DatagramQueueCapacity; use neqo_transport::{ Connection, DatagramQueueOutcome, DatagramTracking, StreamId, StreamType, recv_stream, send_stream, server::ConnectionRef, streams::SendOrder, @@ -501,13 +503,16 @@ impl Handler for Http3Connection { now: Instant, ) -> Res { qtrace!("Close WebTransport session {session_id:?}"); - // Snapshot the stats before tearing the session down, so the caller sees - // the final values. This also rejects non-WebTransport sessions. + // Collect expiries and snapshot the stats before tearing the session down, + // so the caller sees the final values. This also rejects non-WebTransport + // sessions. // // `extended_connect_close_session` then checks the type again. That is // deliberate: it is shared with connect-udp, which needs the check for its own - // close path, so it cannot rely on this one having happened. Two lookups once - // per session close is not worth a validation-skipping variant. + // close path, so it cannot rely on this one having happened. + self.webtransport_session(session_id)? + .borrow_mut() + .expire_datagrams(conn, now); let stats = self.webtransport_session_stats(session_id)?; self.extended_connect_close_session( conn, @@ -812,6 +817,68 @@ impl ServerSession { .expect("test session must exist"); } + /// Set the outgoing-datagram queue's `outgoingMaxAge`, or clear it back + /// to the implementation-defined default with `None`. + /// + /// Test-only; see [`Self::set_max_buffered_datagrams`]. + #[cfg(test)] + pub(crate) fn set_datagram_max_age(&self, max_age: Option, now: Instant) { + let session_id = self.stream_handler.stream_id(); + self.stream_handler + .handler + .borrow_mut() + .base_handler_mut() + .webtransport_session_set_datagram_max_age( + session_id, + &mut self.stream_handler.conn.borrow_mut(), + max_age, + now, + ) + .expect("test session must exist"); + } + + /// This session's statistics, e.g. `datagrams_expired_outgoing`. + /// + /// Test-only; see [`Self::set_max_buffered_datagrams`]. + #[cfg(test)] + pub(crate) fn stats(&self) -> extended_connect::stats::SessionStats { + let session_id = self.stream_handler.stream_id(); + self.stream_handler + .handler + .borrow_mut() + .base_handler_mut() + .webtransport_session_stats(session_id) + .expect("test session must exist") + } + + /// Snapshot of the outgoing-datagram queue's current byte/count state. + /// + /// Test-only; see [`Self::set_max_buffered_datagrams`]. + #[cfg(test)] + pub(crate) fn datagram_queue_capacity(&self) -> DatagramQueueCapacity { + let session_id = self.stream_handler.stream_id(); + self.stream_handler + .handler + .borrow_mut() + .base_handler_mut() + .extended_connect_datagram_queue_capacity( + session_id, + &self.stream_handler.conn.borrow(), + ) + .expect("test session must exist") + } + + /// Test-only: what the transport still queues for this session, which + /// can outlive the session itself. + #[cfg(test)] + pub(crate) fn transport_queued_datagrams(&self) -> usize { + self.stream_handler + .conn + .borrow() + .datagram_queue_capacity(self.stream_handler.stream_id()) + .queued_datagrams + } + // TODO: Currently not called in neqo or gecko. It should likely be called at least from gecko. #[must_use] pub fn remote_datagram_size(&self) -> u64 {