Add RFC 9221 unreliable datagram support

src/connection/connection.rs

@@ -87,6 +87,12 @@ pub struct Connection {
     /// Whether this is a server connection.
     is_server: bool,
 
+    /// RFC 9221 DATAGRAM frames received, awaiting the application (FIFO).
+    dgram_recv: VecDeque<Vec<u8>>,
+    /// DATAGRAM frames queued by the application to send (FIFO). Unreliable:
+    /// dropped, never retransmitted, if lost or if the queue overflows.
+    dgram_send: VecDeque<Vec<u8>>,
+
     /// Connection Identifiers.
     cids: cid::ConnectionIdMgr,
 
@@ -251,6 +257,8 @@ impl Connection {
         let mut conn = Connection {
             version: crate::QUIC_VERSION_V1,
             is_server,
+            dgram_recv: VecDeque::new(),
+            dgram_send: VecDeque::new(),
             cids,
             spaces: space::PacketNumSpaceMap::new(),
             paths,
@@ -719,6 +727,17 @@ impl Connection {
 
             Frame::Ping { .. } => (), // just ignore
 
+            Frame::Datagram { data, .. } => {
+                // RFC 9221: deliver to the application recv queue. Bounded so
+                // a flood can't grow memory unboundedly (unreliable: drop
+                // oldest, mirroring a full NIC rx ring).
+                const MAX_DGRAM_RECV: usize = 1024;
+                if self.dgram_recv.len() >= MAX_DGRAM_RECV {
+                    self.dgram_recv.pop_front();
+                }
+                self.dgram_recv.push_back(data.to_vec());
+            }
+
             Frame::Ack {
                 ack_delay,
                 ack_ranges,
@@ -2009,6 +2028,10 @@ impl Connection {
         // Write STREAM frames
         self.try_write_stream_frames(out, st, pkt_type, path_id)?;
 
+        // Write DATAGRAM frames (RFC 9221) — after streams so bulk stream
+        // data doesn't starve, but they still ride each packet's spare room.
+        self.try_write_datagram_frames(out, st, pkt_type)?;
+
         // Write a NEW_TOKEN frame
         self.try_write_new_token_frame(out, st, pkt_type, path_id)?;
 
@@ -2068,6 +2091,54 @@ impl Connection {
         Ok(())
     }
 
+    /// Write DATAGRAM frames (RFC 9221) from the send queue into the packet.
+    /// Unreliable: each is emitted at most once (no retransmit on loss), and
+    /// one too large to fit any packet is dropped rather than blocking the
+    /// queue. Only in 1-RTT packets (application data).
+    fn try_write_datagram_frames(
+        &mut self,
+        out: &mut [u8],
+        st: &mut FrameWriteStatus,
+        pkt_type: PacketType,
+    ) -> Result<()> {
+        if pkt_type != PacketType::OneRTT {
+            return Ok(());
+        }
+        // Only emit if the peer advertised DATAGRAM support (RFC 9221);
+        // otherwise sending one is a protocol violation.
+        if self.peer_transport_params.max_datagram_frame_size == 0 {
+            self.dgram_send.clear();
+            return Ok(());
+        }
+        while let Some(front) = self.dgram_send.front() {
+            // length-prefixed form (0x31): 1-byte type + varint len + data,
+            // so other frames may coexist in the same packet.
+            let wire = 1 + codec::encode_varint_len(front.len() as u64) + front.len();
+            let remaining = out.len().saturating_sub(st.written);
+            if wire > remaining {
+                // "Too big for any packet" is judged against the negotiated
+                // max, not this packet's leftover room — otherwise a datagram
+                // that doesn't fit the current (ACK-filled) packet but WOULD
+                // fit a fresh one gets stuck in the queue forever.
+                let peer_max = self.peer_transport_params.max_datagram_frame_size as usize;
+                if wire > peer_max {
+                    self.dgram_send.pop_front(); // exceeds peer's max — drop
+                    continue;
+                }
+                break; // try the next (emptier) packet
+            }
+            let data = self.dgram_send.pop_front().unwrap();
+            let frame = Frame::Datagram {
+                length: Some(data.len()),
+                data: Bytes::from(data),
+            };
+            Connection::write_frame_to_packet(frame, out, st)?;
+            st.ack_eliciting = true;
+            st.in_flight = true;
+        }
+        Ok(())
+    }
+
     /// Write PATH_RESPONSE/PATH_CHALLENGE frames if needed.
     fn try_write_path_validation_frames(
         &mut self,
@@ -3111,6 +3182,7 @@ impl Connection {
                 || path.need_send_ping
                 || self.cids.need_send_cid_control_frames()
                 || self.streams.need_send_stream_frames()
+                || !self.dgram_send.is_empty()
                 || self.spaces.need_send_buffered_frames())
         {
             if !self.is_server && self.tls_session.is_in_early_data() {
@@ -3129,6 +3201,10 @@ impl Connection {
             || self.local_error.as_ref().is_some_and(|e| e.is_app)
             || self.cids.need_send_cid_control_frames()
             || self.streams.need_send_stream_frames()
+            // DATAGRAM frames may be sent on any path; without this a
+            // datagram-only workload never consults the multipath scheduler
+            // and everything defaults to the primary path.
+            || !self.dgram_send.is_empty()
     }
 
     /// Find space id for the specified packet type and path id.
@@ -3396,6 +3472,48 @@ impl Connection {
         self.flags.contains(HandshakeCompleted)
     }
 
+    /// Queue an unreliable DATAGRAM (RFC 9221) to send. It is emitted in the
+    /// next outgoing packet(s) and never retransmitted if lost. Bounded send
+    /// queue: returns Error::Done when full (caller drops, like a NIC tx ring).
+    pub fn datagram_send(&mut self, data: &[u8]) -> Result<()> {
+        const MAX_DGRAM_SEND: usize = 1024;
+        if self.dgram_send.len() >= MAX_DGRAM_SEND {
+            return Err(Error::Done);
+        }
+        self.dgram_send.push_back(data.to_vec());
+        // Put the connection on the endpoint's send/tick queues so the queued
+        // datagram is flushed on the next poll (mirrors stream_write).
+        self.mark_tickable(true);
+        self.mark_sendable(true);
+        Ok(())
+    }
+
+    /// Read the next received DATAGRAM into `out`. Returns the byte length, or
+    /// Error::Done if none are queued.
+    pub fn datagram_recv(&mut self, out: &mut [u8]) -> Result<usize> {
+        let d = self.dgram_recv.pop_front().ok_or(Error::Done)?;
+        let n = d.len().min(out.len());
+        out[..n].copy_from_slice(&d[..n]);
+        Ok(n)
+    }
+
+    /// Number of DATAGRAMs waiting to be read.
+    pub fn datagram_recv_queue_len(&self) -> usize {
+        self.dgram_recv.len()
+    }
+
+    /// Largest DATAGRAM payload the peer will accept, or None if the peer did
+    /// not advertise DATAGRAM support. The caller sizes its frames below this.
+    pub fn datagram_max_send_size(&self) -> Option<usize> {
+        let m = self.peer_transport_params.max_datagram_frame_size;
+        if m == 0 {
+            return None;
+        }
+        // m bounds the whole DATAGRAM frame; subtract worst-case header
+        // (1-byte type + up to 8-byte length varint).
+        Some((m as usize).saturating_sub(9))
+    }
+
     /// Check whether the connection handshake is confirmed.
     pub fn is_confirmed(&self) -> bool {
         self.flags.contains(HandshakeConfirmed)
@@ -4963,6 +5081,54 @@ pub(crate) mod tests {
         }
     }
 
+    #[test]
+    fn datagram_end_to_end() -> Result<()> {
+        // Enable RFC 9221 DATAGRAM support on both ends.
+        let mut client_config = TestPair::new_test_config(false)?;
+        client_config.set_max_datagram_frame_size(1500);
+        let mut server_config = TestPair::new_test_config(true)?;
+        server_config.set_max_datagram_frame_size(1500);
+        let mut pair = TestPair::new(&mut client_config, &mut server_config)?;
+        pair.handshake()?;
+
+        // Peer support must have been negotiated via transport params.
+        assert!(pair.client.datagram_max_send_size().is_some());
+        assert!(pair.server.datagram_max_send_size().is_some());
+
+        // client -> server unreliable datagram (the tunnel's uplink)
+        let up = b"tunneled IP packet over a TQUIC DATAGRAM";
+        pair.client.datagram_send(up)?;
+        pair.move_forward()?;
+        let mut out = [0u8; 2048];
+        let n = pair.server.datagram_recv(&mut out)?;
+        assert_eq!(&out[..n], up);
+        assert_eq!(pair.server.datagram_recv(&mut out), Err(Error::Done)); // no more
+
+        // server -> client (the downlink)
+        let down = b"reply datagram";
+        pair.server.datagram_send(down)?;
+        pair.move_forward()?;
+        let n = pair.client.datagram_recv(&mut out)?;
+        assert_eq!(&out[..n], down);
+        Ok(())
+    }
+
+    #[test]
+    fn datagram_peer_no_support_is_dropped() -> Result<()> {
+        // Client enables datagrams; server does NOT advertise support.
+        let mut client_config = TestPair::new_test_config(false)?;
+        client_config.set_max_datagram_frame_size(1500);
+        let mut server_config = TestPair::new_test_config(true)?;
+        let mut pair = TestPair::new(&mut client_config, &mut server_config)?;
+        pair.handshake()?;
+        // Server didn't advertise → client must not attempt to send datagrams.
+        assert!(pair.client.datagram_max_send_size().is_none());
+        pair.client.datagram_send(b"should be dropped, not a protocol error")?;
+        pair.move_forward()?; // must not error out the connection
+        assert!(!pair.client.is_closed() && !pair.server.is_closed());
+        Ok(())
+    }
+
     #[test]
     fn version_negotiation_with_unknown_version() -> Result<()> {
         let mut test_pair = TestPair::new_with_test_config()?;

src/frame.rs

@@ -178,6 +178,15 @@ pub enum Frame {
         seq_num: u64,
         status: u64,
     },
+
+    /// DATAGRAM frame (type=0x30..0x31) carries application data with no
+    /// reliability or ordering guarantees (RFC 9221). `length` is Some when
+    /// the LEN bit (0x01) is set (an explicit length prefix, so more frames
+    /// may follow); None means the data runs to the end of the packet.
+    Datagram {
+        length: Option<usize>,
+        data: Bytes,
+    },
 }
 
 impl Frame {
@@ -359,6 +368,24 @@ impl Frame {
                 status: b.read_varint()?,
             },
 
+            // DATAGRAM frame (RFC 9221). 0x30 = no length (data to end of
+            // packet); 0x31 = explicit varint length prefix.
+            0x30 | 0x31 => {
+                let length = if frame_type & 0x01 != 0 {
+                    Some(b.read_varint()? as usize)
+                } else {
+                    None
+                };
+                let len = length.unwrap_or_else(|| b.len());
+                if len > b.len() {
+                    return Err(Error::BufferTooShort);
+                }
+                let start = buf.len() - b.len();
+                let data = buf.slice(start..(start + len));
+                b.skip(len)?;
+                Frame::Datagram { length, data }
+            }
+
             _ => return Err(Error::FrameEncodingError),
         };
 
@@ -609,6 +636,19 @@ impl Frame {
                 b.write_varint(*seq_num)?;
                 b.write_varint(*status)?;
             }
+
+            Frame::Datagram { length, data } => {
+                match length {
+                    Some(len) => {
+                        b.write_varint(0x31)?;
+                        b.write_varint(*len as u64)?;
+                    }
+                    None => {
+                        b.write_varint(0x30)?;
+                    }
+                }
+                b.write(data.as_ref())?;
+            }
         }
 
         Ok(len - b.len())
@@ -767,6 +807,14 @@ impl Frame {
                     + codec::encode_varint_len(*seq_num)
                     + codec::encode_varint_len(*status)
             }
+
+            Frame::Datagram { length, data } => {
+                // 1-byte frame type + optional length varint + data
+                1 + match length {
+                    Some(len) => codec::encode_varint_len(*len as u64),
+                    None => 0,
+                } + data.len()
+            }
         }
     }
 
@@ -933,6 +981,12 @@ impl Frame {
                 frame_type_value: None,
                 raw: None,
             },
+
+            Frame::Datagram { .. } => QuicFrame::Unknown {
+                raw_frame_type: 0x30,
+                frame_type_value: None,
+                raw: None,
+            },
         }
     }
 
@@ -1112,6 +1166,10 @@ impl std::fmt::Debug for Frame {
                     "PATH_STATUS dcid_seq_num={dcid_seq_num:x} seq_num={seq_num:x} status={status:x}",
                 )?;
             }
+
+            Frame::Datagram { length, data } => {
+                write!(f, "DATAGRAM length={length:?} len={}", data.len())?;
+            }
         }
 
         Ok(())
@@ -1845,6 +1903,45 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn datagram() -> Result<()> {
+        // with explicit length (type 0x31) — round-trips and is ack-eliciting
+        let payload = Bytes::from_static(b"hello bonded world");
+        let frame = Frame::Datagram {
+            length: Some(payload.len()),
+            data: payload.clone(),
+        };
+        assert!(frame.ack_eliciting());
+        assert!(!frame.probing());
+        let mut buf = [0; 128];
+        let len = frame.to_bytes(&mut buf[..])?;
+        assert_eq!(len, frame.wire_len());
+        let mut b = Bytes::copy_from_slice(&buf[..len]);
+        assert_eq!((frame, len), Frame::from_bytes(&mut b, PacketType::OneRTT)?);
+
+        // without length (type 0x30) — data runs to end of buffer
+        let frame2 = Frame::Datagram {
+            length: None,
+            data: payload.clone(),
+        };
+        let len2 = frame2.to_bytes(&mut buf[..])?;
+        let mut b2 = Bytes::copy_from_slice(&buf[..len2]);
+        let (decoded, n) = Frame::from_bytes(&mut b2, PacketType::OneRTT)?;
+        assert_eq!(n, len2);
+        match decoded {
+            Frame::Datagram { length, data } => {
+                assert_eq!(length, None);
+                assert_eq!(data, payload);
+            }
+            _ => panic!("expected Datagram"),
+        }
+
+        // DATAGRAM is forbidden in Initial/Handshake packets
+        let mut b3 = Bytes::copy_from_slice(&buf[..len]);
+        assert!(Frame::from_bytes(&mut b3, PacketType::Initial).is_err());
+        Ok(())
+    }
+
     #[test]
     fn special_frames() -> Result<()> {
         assert_eq!(

src/lib.rs

@@ -449,6 +449,13 @@ impl Config {
         self.local_transport_params.initial_max_data = cmp::min(v, self.max_connection_window);
     }
 
+    /// Set `max_datagram_frame_size` (RFC 9221). Non-zero enables receiving
+    /// unreliable DATAGRAM frames and advertises support to the peer; the peer
+    /// may then send us datagrams up to this size. 0 (default) disables it.
+    pub fn set_max_datagram_frame_size(&mut self, v: u64) {
+        self.local_transport_params.max_datagram_frame_size = cmp::min(v, VINT_MAX);
+    }
+
     /// Set the `initial_max_stream_data_bidi_local` transport parameter.
     /// The value is capped by the setting `max_stream_window`.
     /// The default value is `5242880`.