WebTransport: count a session's expired outgoing datagrams

neqo-http3/src/connection.rs

@@ -4,20 +4,22 @@
 // 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},
     mem,
     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,
@@ -1016,7 +1018,7 @@
         stream_id: StreamId,
         error: AppError,
     ) -> Res<()> {
         qdebug!("[{self}] Reset sending side of stream {stream_id} error={error}");
 
         if self.send_stream_is_critical(stream_id) {
             return Err(Error::InvalidStreamId);
@@ -1055,7 +1057,7 @@
         stream_id: StreamId,
         error: AppError,
     ) -> Res<()> {
         qdebug!("[{self}] Send stop sending for stream {stream_id} error={error}");
         if self.recv_stream_is_critical(stream_id) {
             return Err(Error::InvalidStreamId);
         }
@@ -1132,7 +1134,7 @@
         error: AppError,
         conn: &mut Connection,
     ) -> Res<()> {
         qinfo!("[{self}] cancel_fetch {stream_id} error={error}");
         let send_stream = self.send_streams.get(&stream_id);
         let recv_stream = self.recv_streams.get(&stream_id);
         match (send_stream, recv_stream) {
@@ -1659,6 +1661,34 @@
         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<Duration>,
+        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<DatagramQueueCapacity> {
+        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 @@
         wt: &Rc<RefCell<extended_connect::session::Session>>,
         conn: &mut Connection,
     ) {
+        wt.borrow().drop_queued_datagrams(conn);
+
         let (recv, send) = wt.borrow_mut().take_sub_streams();
 
         #[expect(

neqo-http3/src/features/extended_connect/session.rs

@@ -4,17 +4,19 @@
 // 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},
     rc::Rc,
     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 @@
             &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 @@
         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 @@
         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<Duration>,
+        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 @@
         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);

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,
 }

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();

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)
     }