diff --git a/include/tquic.h b/include/tquic.h index 973bbd1e4..a19f420bb 100644 --- a/include/tquic.h +++ b/include/tquic.h @@ -569,6 +569,13 @@ void quic_config_set_ack_delay_exponent(struct quic_config_t *config, uint64_t v */ void quic_config_set_max_ack_delay(struct quic_config_t *config, uint64_t v); +/** + * Enable ACK Frequency and advertise the minimum supported acknowledgment + * delay in microseconds. Returns 0 on success or a negative error code. + */ +int quic_config_enable_ack_frequency(struct quic_config_t *config, + uint64_t min_ack_delay); + /** * Set congestion control algorithm that the connection would use. */ @@ -1023,6 +1030,11 @@ bool quic_conn_is_in_early_data(struct quic_conn_t *conn); */ bool quic_conn_is_multipath(struct quic_conn_t *conn); +/** + * Check whether the peer advertised ACK Frequency support. + */ +bool quic_conn_peer_supports_ack_frequency(struct quic_conn_t *conn); + /** * Return the negotiated application level protocol. */ @@ -1055,6 +1067,19 @@ int quic_conn_early_data_reason_string(struct quic_conn_t *conn, */ int quic_conn_ping(struct quic_conn_t *conn); +/** + * Queue an ACK_FREQUENCY frame. requested_max_ack_delay is in microseconds. + */ +int quic_conn_update_ack_frequency(struct quic_conn_t *conn, + uint64_t ack_eliciting_threshold, + uint64_t requested_max_ack_delay, + uint64_t reordering_threshold); + +/** + * Queue an IMMEDIATE_ACK frame. + */ +int quic_conn_immediate_ack(struct quic_conn_t *conn); + /** * Send a Ping frame on the specified path for keep-alive. * The API is only applicable to multipath quic connections. diff --git a/src/ack_frequency.rs b/src/ack_frequency.rs new file mode 100644 index 000000000..9dffb19c3 --- /dev/null +++ b/src/ack_frequency.rs @@ -0,0 +1,177 @@ +// Copyright (c) 2026 The TQUIC Authors. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! State shared by the sending and receiving sides of ACK Frequency. + +use std::collections::BTreeMap; +use std::time::Duration; + +use crate::error::Error; +use crate::frame::Frame; +use crate::Result; + +/// Requested Max Ack Delay values are invalid when they are 2^14 milliseconds +/// or greater. +pub(crate) const MAX_REQUESTED_ACK_DELAY: u64 = (1 << 14) * 1000; + +/// The values carried by an ACK_FREQUENCY frame. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(crate) struct AckFrequencyParams { + pub sequence_number: u64, + pub ack_eliciting_threshold: u64, + pub requested_max_ack_delay: u64, + pub reordering_threshold: u64, +} + +impl AckFrequencyParams { + pub fn into_frame(self) -> Frame { + Frame::AckFrequency { + sequence_number: self.sequence_number, + ack_eliciting_threshold: self.ack_eliciting_threshold, + requested_max_ack_delay: self.requested_max_ack_delay, + reordering_threshold: self.reordering_threshold, + } + } +} + +/// The latest acknowledgment policy received from the peer. +#[derive(Clone, Debug, Default)] +pub(crate) struct AckFrequencyReceiverState { + params: Option, +} + +impl AckFrequencyReceiverState { + /// Process an ACK_FREQUENCY frame. Returns true when it supersedes the + /// current policy and false when it is stale. + pub fn on_frame( + &mut self, + params: AckFrequencyParams, + min_ack_delay: Option, + ) -> Result { + let min_ack_delay = min_ack_delay.ok_or(Error::ProtocolViolation)?; + if self + .params + .is_some_and(|current| params.sequence_number <= current.sequence_number) + { + return Ok(false); + } + + if params.requested_max_ack_delay < min_ack_delay + || params.requested_max_ack_delay >= MAX_REQUESTED_ACK_DELAY + { + return Err(Error::ProtocolViolation); + } + + self.params = Some(params); + Ok(true) + } + + pub fn params(&self) -> Option { + self.params + } +} + +/// Tracks acknowledged and in-flight ACK_FREQUENCY frames for PTO calculation. +#[derive(Clone, Debug, Default)] +pub(crate) struct AckFrequencySenderState { + acknowledged: Option, + in_flight: BTreeMap, +} + +impl AckFrequencySenderState { + pub fn on_frame_sent(&mut self, params: AckFrequencyParams) { + self.in_flight.insert(params.sequence_number, params); + } + + pub fn on_frame_acked(&mut self, params: AckFrequencyParams) { + if self.acknowledged.map_or(true, |current| { + params.sequence_number > current.sequence_number + }) { + self.acknowledged = Some(params); + } + + // Processing a newer frame supersedes all older acknowledgment policies. + self.in_flight + .retain(|sequence, _| *sequence > params.sequence_number); + } + + pub fn effective_max_ack_delay(&self, transport_max_ack_delay: Duration) -> Duration { + let acknowledged = self + .acknowledged + .map(|p| Duration::from_micros(p.requested_max_ack_delay)) + .unwrap_or(transport_max_ack_delay); + + self.in_flight.values().fold(acknowledged, |delay, p| { + delay.max(Duration::from_micros(p.requested_max_ack_delay)) + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn params(sequence_number: u64, delay: u64) -> AckFrequencyParams { + AckFrequencyParams { + sequence_number, + ack_eliciting_threshold: 1, + requested_max_ack_delay: delay, + reordering_threshold: 1, + } + } + + #[test] + fn receiver_rejects_invalid_delay_and_ignores_stale_frames() { + let mut state = AckFrequencyReceiverState::default(); + assert_eq!( + state.on_frame(params(1, 999), None), + Err(Error::ProtocolViolation) + ); + assert_eq!( + state.on_frame(params(1, 999), Some(1000)), + Err(Error::ProtocolViolation) + ); + assert_eq!( + state.on_frame(params(1, MAX_REQUESTED_ACK_DELAY), Some(0)), + Err(Error::ProtocolViolation) + ); + + assert_eq!(state.on_frame(params(2, 1000), Some(1000)), Ok(true)); + assert_eq!(state.on_frame(params(1, 0), Some(1000)), Ok(false)); + assert_eq!(state.params(), Some(params(2, 1000))); + } + + #[test] + fn sender_uses_largest_in_flight_delay_until_acknowledged() { + let mut state = AckFrequencySenderState::default(); + let transport_delay = Duration::from_millis(25); + assert_eq!( + state.effective_max_ack_delay(transport_delay), + transport_delay + ); + + state.on_frame_sent(params(0, 50_000)); + state.on_frame_sent(params(1, 10_000)); + assert_eq!( + state.effective_max_ack_delay(transport_delay), + Duration::from_millis(50) + ); + + state.on_frame_acked(params(1, 10_000)); + assert_eq!( + state.effective_max_ack_delay(transport_delay), + Duration::from_millis(10) + ); + } +} diff --git a/src/connection/connection.rs b/src/connection/connection.rs index 1b4190bc5..2f0ecaa06 100644 --- a/src/connection/connection.rs +++ b/src/connection/connection.rs @@ -41,6 +41,9 @@ use self::stream::Stream; use self::stream::StreamIter; use self::timer::Timer; use self::ConnectionFlags::*; +use crate::ack_frequency::AckFrequencyParams; +use crate::ack_frequency::AckFrequencyReceiverState; +use crate::ack_frequency::MAX_REQUESTED_ACK_DELAY; use crate::codec; use crate::codec::Decoder; use crate::codec::Encoder; @@ -124,6 +127,18 @@ pub struct Connection { /// Recovery and congestion control configurations. recovery_conf: RecoveryConfig, + /// ACK Frequency policy received from the peer. + ack_frequency_receiver: AckFrequencyReceiverState, + + /// Sequence number allocated to the next outgoing ACK_FREQUENCY frame. + next_ack_frequency_sequence: u64, + + /// The newest ACK_FREQUENCY frame waiting to be sent. + pending_ack_frequency: Option, + + /// Whether an IMMEDIATE_ACK frame is waiting to be sent. + pending_immediate_ack: bool, + /// Error to be sent to the peer in a CONNECTION_CLOSE frame. local_error: Option, @@ -263,6 +278,10 @@ impl Connection { peer_transport_params: TransportParams::default(), local_transport_params: conf.local_transport_params.clone(), recovery_conf: conf.recovery.clone(), + ack_frequency_receiver: AckFrequencyReceiverState::default(), + next_ack_frequency_sequence: 0, + pending_ack_frequency: None, + pending_immediate_ack: false, local_error: None, peer_error: None, timers: timer::TimerTable::default(), @@ -649,6 +668,7 @@ impl Connection { // Update packet number space let space = self.spaces.get_mut(space_id).ok_or(Error::InternalError)?; + let previous_largest_ack_eliciting = space.largest_rx_ack_eliciting_pkt_num; if space.recv_pkt_num_need_ack.max() < Some(pkt_num) { space.largest_rx_pkt_time = info.time; } @@ -661,11 +681,19 @@ impl Connection { // TODO: try to do connection migration } if ack_eliciting_pkt { - space.largest_rx_ack_eliciting_pkt_num = - cmp::max(space.largest_rx_ack_eliciting_pkt_num, pkt_num); + space.largest_rx_ack_eliciting_pkt_num = Some( + space + .largest_rx_ack_eliciting_pkt_num + .map_or(pkt_num, |largest| cmp::max(largest, pkt_num)), + ); } - self.try_schedule_ack_frame(space_id, pkt_num, ack_eliciting_pkt)?; + self.try_schedule_ack_frame( + space_id, + pkt_num, + ack_eliciting_pkt, + previous_largest_ack_eliciting, + )?; // An endpoint restarts its idle timer when a packet from its peer is // received and processed successfully. @@ -983,6 +1011,32 @@ impl Connection { Frame::StreamsBlocked { bidi, max } => { self.streams.on_streams_blocked_frame_received(max, bidi)?; } + + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => { + self.ack_frequency_receiver.on_frame( + AckFrequencyParams { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + }, + self.local_transport_params.min_ack_delay, + )?; + } + + Frame::ImmediateAck => { + if self.local_transport_params.min_ack_delay.is_none() { + return Err(Error::ProtocolViolation); + } + let space = self.spaces.get_mut(space_id).ok_or(Error::InternalError)?; + space.need_send_ack = true; + space.ack_timer = None; + } } Ok(()) @@ -1374,6 +1428,7 @@ impl Connection { space_id: SpaceId, pkt_num: u64, ack_eliciting: bool, + previous_largest_ack_eliciting: Option, ) -> Result<()> { if !ack_eliciting { return Ok(()); @@ -1391,36 +1446,80 @@ impl Connection { return Ok(()); } - // A receiver SHOULD send an ACK frame after receiving at least two - // ack-eliciting packets. + let ack_frequency = self.ack_frequency_receiver.params(); + + // ACK_FREQUENCY's threshold counts packets that can be received before + // an ACK is sent. Thus, a value of 1 acknowledges every second packet. space.ack_eliciting_pkts_since_last_sent_ack += 1; - let ack_eliciting_threshold = self.recovery_conf.ack_eliciting_threshold; - if space.ack_eliciting_pkts_since_last_sent_ack >= ack_eliciting_threshold { + let threshold_reached = ack_frequency.map_or_else( + || { + space.ack_eliciting_pkts_since_last_sent_ack + >= self.recovery_conf.ack_eliciting_threshold + }, + |params| space.ack_eliciting_pkts_since_last_sent_ack > params.ack_eliciting_threshold, + ); + if threshold_reached { space.need_send_ack = true; space.ack_timer = None; return Ok(()); } - // In order to assist loss detection at the sender, an endpoint SHOULD - // generate and send an ACK frame without delay when it receives an - // ack-eliciting packet either: - // - when the received packet has a packet number less than another - // ack-eliciting packet that has been received, or - // - when the packet has a packet number larger than the highest-numbered - // ack-eliciting packet that has been received and there are missing - // packets between that packet and this packet. - if pkt_num < space.largest_rx_ack_eliciting_pkt_num - || pkt_num > space.largest_rx_ack_eliciting_pkt_num + 1 + if let Some(params) = ack_frequency { + let reordering_threshold = params.reordering_threshold; + if reordering_threshold > 0 { + // A sufficiently old packet is acknowledged immediately. + let old_packet = space.largest_ack_sent.is_some_and(|largest_acked| { + largest_acked + .checked_sub(reordering_threshold) + .is_some_and(|limit| pkt_num <= limit) + }); + + // draft-14 also compares Largest Unacked with the smallest + // unreported missing packet. The receive window bounds this + // search without iterating over attacker-controlled gaps. + let missing_packet = space + .largest_rx_ack_eliciting_pkt_num + .and_then(|largest_unacked| { + let first_unreported = space + .largest_ack_sent + .and_then(|largest_acked| { + largest_acked.checked_sub(reordering_threshold) + }) + .map_or(0, |largest_reported_missing| { + largest_reported_missing.saturating_add(1) + }); + space + .recv_pkt_num_win + .first_missing(first_unreported, largest_unacked) + .map(|smallest_missing| (largest_unacked, smallest_missing)) + }) + .is_some_and(|(largest_unacked, smallest_missing)| { + largest_unacked.saturating_sub(smallest_missing) >= reordering_threshold + }); + + if old_packet || missing_packet { + space.need_send_ack = true; + space.ack_timer = None; + return Ok(()); + } + } + } else if previous_largest_ack_eliciting + .is_some_and(|largest| pkt_num < largest || pkt_num > largest.saturating_add(1)) { + // Preserve RFC 9000's immediate ACK behavior when ACK Frequency + // has not changed the reordering policy. space.need_send_ack = true; space.ack_timer = None; return Ok(()); } - // All ack-eliciting 0-RTT and 1-RTT packets within its advertised - // max_ack_delay. + // Acknowledge within the requested delay, or the transport parameter + // delay when ACK Frequency has not supplied an override. if space.ack_timer.is_none() { - let ack_delay = time::Duration::from_millis(self.peer_transport_params.max_ack_delay); + let ack_delay = ack_frequency.map_or_else( + || time::Duration::from_millis(self.peer_transport_params.max_ack_delay), + |params| time::Duration::from_micros(params.requested_max_ack_delay), + ); space.ack_timer = Some(time::Instant::now() + ack_delay); debug!( "{} set ack timer for space {:?}, timeout {:?} ", @@ -1432,6 +1531,7 @@ impl Connection { /// Process acknowledged frames in each packet number space fn try_process_acked_frames(&mut self) { + let mut acked_ack_frequency = Vec::new(); for (_, space) in self.spaces.iter_mut() { for acked_frame in space.acked.drain(..) { match acked_frame { @@ -1491,10 +1591,30 @@ impl Connection { } } + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => { + acked_ack_frequency.push(AckFrequencyParams { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + }); + } + _ => (), } } } + + for params in acked_ack_frequency { + for (_, path) in self.paths.iter_mut() { + path.recovery.on_ack_frequency_acked(params); + } + } } /// If any path doesn't has a DCID, try to allocate one for it. @@ -1827,6 +1947,25 @@ impl Connection { rate_sample_state: Default::default(), buffer_flags: write_status.buffer_flags, }; + let sent_ack_frequency = sent_pkt.frames.iter().find_map(|frame| match frame { + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => Some(AckFrequencyParams { + sequence_number: *sequence_number, + ack_eliciting_threshold: *ack_eliciting_threshold, + requested_max_ack_delay: *requested_max_ack_delay, + reordering_threshold: *reordering_threshold, + }), + _ => None, + }); + if let Some(params) = sent_ack_frequency { + for (_, path) in self.paths.iter_mut() { + path.recovery.on_ack_frequency_sent(params); + } + } debug!( "{} sent packet {:?} {:?} {:?}", self.trace_id, @@ -1978,6 +2117,10 @@ impl Connection { // simplicity. let out = &mut buf[..left]; + // ACK_FREQUENCY and IMMEDIATE_ACK are congestion controlled and can + // only be carried in 1-RTT packets. + self.try_write_ack_frequency_control_frames(out, st, pkt_type)?; + // Write PATH_CHALLENGE/PATH_RESPONSE frames self.try_write_path_validation_frames(out, st, pkt_type, path_id)?; @@ -2068,6 +2211,33 @@ impl Connection { Ok(()) } + fn try_write_ack_frequency_control_frames( + &mut self, + out: &mut [u8], + st: &mut FrameWriteStatus, + pkt_type: PacketType, + ) -> Result<()> { + if pkt_type != PacketType::OneRTT || self.is_closing() { + return Ok(()); + } + + if let Some(params) = self.pending_ack_frequency { + Connection::write_frame_to_packet(params.into_frame(), out, st)?; + self.pending_ack_frequency = None; + st.ack_eliciting = true; + st.in_flight = true; + } + + if self.pending_immediate_ack { + Connection::write_frame_to_packet(Frame::ImmediateAck, out, st)?; + self.pending_immediate_ack = false; + 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, @@ -2175,14 +2345,17 @@ impl Connection { let ack_delay_exp = self.local_transport_params.ack_delay_exponent as u32; let ack_delay = space.largest_rx_pkt_time.elapsed(); let ack_delay = ack_delay.as_micros() as u64 / 2_u64.pow(ack_delay_exp); + let largest_ack = space.recv_pkt_num_need_ack.max(); let frame = Frame::Ack { ack_delay, ack_ranges: space.recv_pkt_num_need_ack.clone(), ecn_counts: None, // ECN not supported }; Connection::write_frame_to_packet(frame, out, st)?; + space.largest_ack_sent = cmp::max(space.largest_ack_sent, largest_ack); space.need_send_ack = false; space.ack_eliciting_pkts_since_last_sent_ack = 0; + space.ack_timer = None; Ok(()) } @@ -2822,6 +2995,27 @@ impl Connection { space.need_send_ack = true; } + // A lost ACK_FREQUENCY frame can be retransmitted with the + // same sequence number unless a newer policy superseded it. + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => { + if self.next_ack_frequency_sequence <= sequence_number.saturating_add(1) { + self.pending_ack_frequency = Some(AckFrequencyParams { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + }); + } + } + + // IMMEDIATE_ACK frames are never retransmitted. + Frame::ImmediateAck => {} + // The HANDSHAKE_DONE frame MUST be retransmitted until it // is acknowledged. Frame::HandshakeDone if !self.flags.contains(HandshakeDoneAcked) => { @@ -3109,6 +3303,8 @@ impl Connection { || path.need_send_validation_frames(self.is_server) || path.dplpmtud.should_probe() || path.need_send_ping + || self.pending_ack_frequency.is_some() + || self.pending_immediate_ack || self.cids.need_send_cid_control_frames() || self.streams.need_send_stream_frames() || self.spaces.need_send_buffered_frames()) @@ -3127,6 +3323,8 @@ impl Connection { self.need_send_handshake_done_frame() || self.need_send_new_token_frame() || self.local_error.as_ref().is_some_and(|e| e.is_app) + || self.pending_ack_frequency.is_some() + || self.pending_immediate_ack || self.cids.need_send_cid_control_frames() || self.streams.need_send_stream_frames() } @@ -3175,6 +3373,11 @@ impl Connection { if self.cids.zero_length_scid() { cid_pid = None; } + let ack_frequency_state = self + .paths + .get_active() + .ok() + .map(|path| path.recovery.ack_frequency_state()); let mut path = path::Path::new( info.dst, info.src, @@ -3182,6 +3385,11 @@ impl Connection { &self.recovery_conf, &self.trace_id, ); + path.recovery.max_ack_delay = + time::Duration::from_millis(self.peer_transport_params.max_ack_delay); + if let Some(state) = ack_frequency_state { + path.recovery.set_ack_frequency_state(state); + } if self.is_server { path.anti_ampl_limit = buf_len * self.paths.anti_ampl_factor; } @@ -3721,6 +3929,72 @@ impl Connection { self.paths.mark_ping(path_addr) } + /// Return whether the peer advertised support for ACK Frequency. + pub fn peer_supports_ack_frequency(&self) -> bool { + self.peer_transport_params.min_ack_delay.is_some() + } + + /// Queue a new ACK_FREQUENCY frame for the peer. + /// + /// `requested_max_ack_delay` is expressed in microseconds. A newer call + /// supersedes an update that has not yet been sent. + pub fn update_ack_frequency( + &mut self, + ack_eliciting_threshold: u64, + requested_max_ack_delay: u64, + reordering_threshold: u64, + ) -> Result<()> { + if !self.is_established() { + return Err(Error::InvalidOperation( + "ACK Frequency requires an established connection".into(), + )); + } + + let min_ack_delay = self + .peer_transport_params + .min_ack_delay + .ok_or_else(|| Error::InvalidOperation("peer does not support ACK Frequency".into()))?; + if requested_max_ack_delay < min_ack_delay + || requested_max_ack_delay >= MAX_REQUESTED_ACK_DELAY + { + return Err(Error::InvalidConfig( + "requested_max_ack_delay is outside the peer's supported range".into(), + )); + } + if ack_eliciting_threshold > crate::VINT_MAX || reordering_threshold > crate::VINT_MAX { + return Err(Error::InvalidConfig( + "ACK Frequency thresholds exceed the QUIC varint limit".into(), + )); + } + if self.next_ack_frequency_sequence > crate::VINT_MAX { + return Err(Error::InvalidOperation( + "ACK Frequency sequence number exhausted".into(), + )); + } + + let params = AckFrequencyParams { + sequence_number: self.next_ack_frequency_sequence, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + }; + self.next_ack_frequency_sequence = self.next_ack_frequency_sequence.saturating_add(1); + self.pending_ack_frequency = Some(params); + Ok(()) + } + + /// Queue an IMMEDIATE_ACK frame for the peer. + pub fn immediate_ack(&mut self) -> Result<()> { + if !self.is_established() || !self.peer_supports_ack_frequency() { + return Err(Error::InvalidOperation( + "peer ACK Frequency support is unavailable".into(), + )); + } + + self.pending_immediate_ack = true; + Ok(()) + } + /// Client add a new path on the connection. pub fn add_path(&mut self, local_addr: SocketAddr, remote_addr: SocketAddr) -> Result { if self.is_server { @@ -3741,6 +4015,11 @@ impl Connection { self.cids.lowest_unused_dcid_seq() }; + let ack_frequency_state = self + .paths + .get_active() + .ok() + .map(|path| path.recovery.ack_frequency_state()); let mut path = path::Path::new( local_addr, remote_addr, @@ -3748,6 +4027,11 @@ impl Connection { &self.recovery_conf, &self.trace_id, ); + path.recovery.max_ack_delay = + time::Duration::from_millis(self.peer_transport_params.max_ack_delay); + if let Some(state) = ack_frequency_state { + path.recovery.set_ack_frequency_state(state); + } path.dcid_seq = dcid_seq; let pid = self.paths.insert_path(path)?; self.paths.get_mut(pid)?.update_trace_id(pid); @@ -7947,6 +8231,134 @@ pub(crate) mod tests { Ok(()) } + + #[test] + fn ack_frequency_negotiation_and_control_frames() -> Result<()> { + let mut client_config = TestPair::new_test_config(false)?; + client_config.enable_ack_frequency(1000)?; + let mut server_config = TestPair::new_test_config(true)?; + server_config.enable_ack_frequency(2000)?; + let mut test_pair = TestPair::new(&mut client_config, &mut server_config)?; + test_pair.handshake()?; + + assert_eq!( + test_pair.client.peer_transport_params.min_ack_delay, + Some(2000) + ); + assert_eq!( + test_pair.server.peer_transport_params.min_ack_delay, + Some(1000) + ); + + test_pair.client.update_ack_frequency(1, 10_000, 3)?; + let packets = TestPair::conn_packets_out(&mut test_pair.client)?; + TestPair::conn_packets_in(&mut test_pair.server, packets)?; + assert_eq!( + test_pair.server.ack_frequency_receiver.params(), + Some(AckFrequencyParams { + sequence_number: 0, + ack_eliciting_threshold: 1, + requested_max_ack_delay: 10_000, + reordering_threshold: 3, + }) + ); + + test_pair.client.immediate_ack()?; + let packets = TestPair::conn_packets_out(&mut test_pair.client)?; + TestPair::conn_packets_in(&mut test_pair.server, packets)?; + assert!( + test_pair + .server + .spaces + .get(SpaceId::Data) + .ok_or(Error::InternalError)? + .need_send_ack + ); + + Ok(()) + } + + #[test] + fn ack_frequency_requires_transport_parameter() -> Result<()> { + let mut test_pair = TestPair::new_with_test_config()?; + test_pair.handshake()?; + + assert!(matches!( + test_pair.client.update_ack_frequency(1, 10_000, 1), + Err(Error::InvalidOperation(_)) + )); + + let frame = Frame::AckFrequency { + sequence_number: 0, + ack_eliciting_threshold: 1, + requested_max_ack_delay: 10_000, + reordering_threshold: 1, + }; + assert_eq!( + test_pair.build_packet_and_send(PacketType::OneRTT, &[frame], false), + Err(Error::ProtocolViolation) + ); + + Ok(()) + } + + #[test] + fn ack_frequency_reordering_threshold_uses_draft_14_rules() -> Result<()> { + let mut conn = TestPair::new_with_test_config()?.server; + conn.ack_frequency_receiver.on_frame( + AckFrequencyParams { + sequence_number: 1, + ack_eliciting_threshold: 100, + requested_max_ack_delay: 10_000, + reordering_threshold: 3, + }, + Some(0), + )?; + + *conn + .spaces + .get_mut(SpaceId::Data) + .ok_or(Error::InternalError)? = PacketNumSpace::new(SpaceId::Data); + + // This is the reordering-threshold example from draft-14. With a + // threshold of 3, packets 5, 9, and 10 trigger immediate ACKs. + let packets = [0, 1, 3, 4, 5, 8, 9, 10]; + let expected = [false, false, false, false, true, false, true, true]; + for (pkt_num, expected_ack) in packets.into_iter().zip(expected) { + let previous = { + let space = conn.spaces.get_mut(SpaceId::Data).unwrap(); + let previous = space.largest_rx_ack_eliciting_pkt_num; + space.recv_pkt_num_win.insert(pkt_num); + space.largest_rx_ack_eliciting_pkt_num = Some(pkt_num); + previous + }; + conn.try_schedule_ack_frame(SpaceId::Data, pkt_num, true, previous)?; + let space = conn.spaces.get_mut(SpaceId::Data).unwrap(); + assert_eq!(space.need_send_ack, expected_ack, "packet {pkt_num}"); + if expected_ack { + space.largest_ack_sent = Some(pkt_num); + space.need_send_ack = false; + space.ack_eliciting_pkts_since_last_sent_ack = 0; + space.ack_timer = None; + } + } + + // A packet at or below Largest Acked - Reordering Threshold also + // triggers an immediate ACK. + let previous = conn + .spaces + .get(SpaceId::Data) + .and_then(|space| space.largest_rx_ack_eliciting_pkt_num); + conn.spaces + .get_mut(SpaceId::Data) + .unwrap() + .recv_pkt_num_win + .insert(7); + conn.try_schedule_ack_frame(SpaceId::Data, 7, true, previous)?; + assert!(conn.spaces.get(SpaceId::Data).unwrap().need_send_ack); + + Ok(()) + } } mod cid; diff --git a/src/connection/recovery.rs b/src/connection/recovery.rs index 007ba2c8a..8381c5ace 100644 --- a/src/connection/recovery.rs +++ b/src/connection/recovery.rs @@ -30,6 +30,8 @@ use super::space::SpaceId; use super::space::SpaceId::*; use super::Connection; use super::HandshakeStatus; +use crate::ack_frequency::AckFrequencyParams; +use crate::ack_frequency::AckFrequencySenderState; use crate::congestion_control; use crate::congestion_control::CongestionController; use crate::congestion_control::Pacer; @@ -60,6 +62,10 @@ pub struct Recovery { /// It is used for PTO calculation. pub max_ack_delay: Duration, + /// ACK_FREQUENCY frames that affect the peer's acknowledgment delay and + /// therefore the PTO calculation. + ack_frequency: AckFrequencySenderState, + /// The validated maximum size of outgoing UDP payloads in bytes. pub max_datagram_size: usize, @@ -127,6 +133,7 @@ impl Recovery { pub(super) fn new(conf: &RecoveryConfig) -> Self { Recovery { max_ack_delay: conf.max_ack_delay, + ack_frequency: AckFrequencySenderState::default(), max_datagram_size: crate::DEFAULT_SEND_UDP_PAYLOAD_SIZE, pto_linear_factor: conf.pto_linear_factor, max_pto: conf.max_pto, @@ -154,6 +161,22 @@ impl Recovery { self.trace_id = trace_id.to_string(); } + pub(super) fn on_ack_frequency_sent(&mut self, params: AckFrequencyParams) { + self.ack_frequency.on_frame_sent(params); + } + + pub(super) fn on_ack_frequency_acked(&mut self, params: AckFrequencyParams) { + self.ack_frequency.on_frame_acked(params); + } + + pub(super) fn ack_frequency_state(&self) -> AckFrequencySenderState { + self.ack_frequency.clone() + } + + pub(super) fn set_ack_frequency_state(&mut self, state: AckFrequencySenderState) { + self.ack_frequency = state; + } + /// Handle packet sent event. /// /// See RFC 9002 Section A.5. On Sending a Packet @@ -713,8 +736,11 @@ impl Recovery { .pto_count .saturating_sub(self.pto_linear_factor as usize); + let max_ack_delay = self + .ack_frequency + .effective_max_ack_delay(self.max_ack_delay); cmp::min( - duration + self.max_ack_delay * 2_u32.saturating_pow(backoff_factor as u32), + duration + max_ack_delay * 2_u32.saturating_pow(backoff_factor as u32), self.max_pto, ) } @@ -1655,4 +1681,34 @@ mod tests { Ok(()) } + + #[test] + fn ack_frequency_delay_is_included_in_pto() { + let conf = new_test_recovery_config(); + let mut recovery = Recovery::new(&conf); + let pto_base = Duration::from_millis(500); + let params = AckFrequencyParams { + sequence_number: 1, + ack_eliciting_threshold: 8, + requested_max_ack_delay: 250_000, + reordering_threshold: 3, + }; + + recovery.on_ack_frequency_sent(params); + assert_eq!( + recovery.pto_with_ack_delay(pto_base), + Duration::from_millis(750) + ); + + let acknowledged = AckFrequencyParams { + sequence_number: 2, + requested_max_ack_delay: 10_000, + ..params + }; + recovery.on_ack_frequency_acked(acknowledged); + assert_eq!( + recovery.pto_with_ack_delay(pto_base), + Duration::from_millis(510) + ); + } } diff --git a/src/connection/space.rs b/src/connection/space.rs index 154cc95c0..6bec8261e 100644 --- a/src/connection/space.rs +++ b/src/connection/space.rs @@ -90,7 +90,7 @@ pub struct PacketNumSpace { pub largest_rx_non_probing_pkt_num: u64, /// Highest received ack-eliciting packet number. - pub largest_rx_ack_eliciting_pkt_num: u64, + pub largest_rx_ack_eliciting_pkt_num: Option, /// The packet numbers to acknowledge. pub recv_pkt_num_need_ack: RangeSet, @@ -130,6 +130,9 @@ pub struct PacketNumSpace { /// The largest packet number acknowledged in the packet number space so far. pub largest_acked_pkt: u64, + /// The largest packet number included in an ACK frame sent to the peer. + pub largest_ack_sent: Option, + /// The number of times a PTO has been sent without receiving an acknowledgment. pub loss_probes: usize, @@ -156,7 +159,7 @@ impl PacketNumSpace { first_pkt_num_sent: None, largest_rx_pkt_time: Instant::now(), largest_rx_non_probing_pkt_num: 0, - largest_rx_ack_eliciting_pkt_num: 0, + largest_rx_ack_eliciting_pkt_num: None, recv_pkt_num_need_ack: RangeSet::new(crate::MAX_ACK_RANGES), recv_pkt_num_win: SeqNumWindow::default(), need_send_ack: false, @@ -169,6 +172,7 @@ impl PacketNumSpace { time_of_last_sent_ack_eliciting_pkt: None, loss_time: None, largest_acked_pkt: u64::MAX, + largest_ack_sent: None, loss_probes: 0, bytes_in_flight: 0, ack_eliciting_in_flight: 0, diff --git a/src/ffi.rs b/src/ffi.rs index ae5d7314c..65ec23d34 100644 --- a/src/ffi.rs +++ b/src/ffi.rs @@ -296,6 +296,18 @@ pub extern "C" fn quic_config_set_max_ack_delay(config: &mut Config, v: u64) { config.set_max_ack_delay(v); } +/// Enable ACK Frequency and advertise `min_ack_delay` in microseconds. +#[no_mangle] +pub extern "C" fn quic_config_enable_ack_frequency( + config: &mut Config, + min_ack_delay: u64, +) -> c_int { + match config.enable_ack_frequency(min_ack_delay) { + Ok(_) => 0, + Err(e) => e.to_errno() as c_int, + } +} + /// Set congestion control algorithm that the connection would use. #[no_mangle] pub extern "C" fn quic_config_set_congestion_control_algorithm( @@ -1080,6 +1092,12 @@ pub extern "C" fn quic_conn_is_multipath(conn: &mut Connection) -> bool { conn.is_multipath() } +/// Check whether the peer advertised ACK Frequency support. +#[no_mangle] +pub extern "C" fn quic_conn_peer_supports_ack_frequency(conn: &mut Connection) -> bool { + conn.peer_supports_ack_frequency() +} + /// Return the negotiated application level protocol. #[no_mangle] pub extern "C" fn quic_conn_application_proto( @@ -1161,6 +1179,33 @@ pub extern "C" fn quic_conn_ping(conn: &mut Connection) -> c_int { } } +/// Queue an ACK_FREQUENCY frame. Delay is expressed in microseconds. +#[no_mangle] +pub extern "C" fn quic_conn_update_ack_frequency( + conn: &mut Connection, + ack_eliciting_threshold: u64, + requested_max_ack_delay: u64, + reordering_threshold: u64, +) -> c_int { + match conn.update_ack_frequency( + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + ) { + Ok(_) => 0, + Err(e) => e.to_errno() as c_int, + } +} + +/// Queue an IMMEDIATE_ACK frame. +#[no_mangle] +pub extern "C" fn quic_conn_immediate_ack(conn: &mut Connection) -> c_int { + match conn.immediate_ack() { + Ok(_) => 0, + Err(e) => e.to_errno() as c_int, + } +} + /// Send a Ping frame on the specified path for keep-alive. /// The API is only applicable to multipath quic connections. #[no_mangle] diff --git a/src/frame.rs b/src/frame.rs index a5ec9657e..620999c1b 100644 --- a/src/frame.rs +++ b/src/frame.rs @@ -161,6 +161,19 @@ pub enum Frame { /// confirmation of the handshake to the client. HandshakeDone, + /// IMMEDIATE_ACK frame (type=0x1f) requests that the peer acknowledge the + /// containing packet without delay. + ImmediateAck, + + /// ACK_FREQUENCY frame (type=0xaf) updates the peer's acknowledgment + /// policy. All delay values are expressed in microseconds. + AckFrequency { + sequence_number: u64, + ack_eliciting_threshold: u64, + requested_max_ack_delay: u64, + reordering_threshold: u64, + }, + /// PATH_ABANDON frame informs the peer to abandon a path. /// See draft-ietf-quic-multipath-05. PathAbandon { @@ -347,6 +360,15 @@ impl Frame { 0x1e => Frame::HandshakeDone, + 0x1f => Frame::ImmediateAck, + + 0xaf => Frame::AckFrequency { + sequence_number: b.read_varint()?, + ack_eliciting_threshold: b.read_varint()?, + requested_max_ack_delay: b.read_varint()?, + reordering_threshold: b.read_varint()?, + }, + 0x15228c05 => Frame::PathAbandon { dcid_seq_num: b.read_varint()?, error_code: b.read_varint()?, @@ -374,6 +396,13 @@ impl Frame { // PADDING and PING are allowed on all packet types. (_, Frame::Paddings { .. }) | (_, Frame::Ping { .. }) => true, + // ACK_FREQUENCY and IMMEDIATE_ACK are only permitted in 1-RTT + // packets after support has been negotiated. + (PacketType::OneRTT, Frame::AckFrequency { .. }) => true, + (PacketType::OneRTT, Frame::ImmediateAck) => true, + (_, Frame::AckFrequency { .. }) => false, + (_, Frame::ImmediateAck) => false, + // ACK, CRYPTO, HANDSHAKE_DONE, NEW_TOKEN, PATH_RESPONSE, and // RETIRE_CONNECTION_ID can't be sent on 0-RTT packets. (PacketType::ZeroRTT, Frame::Ack { .. }) => false, @@ -587,6 +616,23 @@ impl Frame { b.write_varint(0x1e)?; } + Frame::ImmediateAck => { + b.write_varint(0x1f)?; + } + + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => { + b.write_varint(0xaf)?; + b.write_varint(*sequence_number)?; + b.write_varint(*ack_eliciting_threshold)?; + b.write_varint(*requested_max_ack_delay)?; + b.write_varint(*reordering_threshold)?; + } + Frame::PathAbandon { dcid_seq_num, error_code, @@ -743,6 +789,21 @@ impl Frame { Frame::HandshakeDone => 1, + Frame::ImmediateAck => 1, + + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => { + codec::encode_varint_len(0xaf) + + codec::encode_varint_len(*sequence_number) + + codec::encode_varint_len(*ack_eliciting_threshold) + + codec::encode_varint_len(*requested_max_ack_delay) + + codec::encode_varint_len(*reordering_threshold) + } + Frame::PathAbandon { dcid_seq_num, error_code, @@ -922,6 +983,20 @@ impl Frame { Frame::HandshakeDone => QuicFrame::HandshakeDone, + Frame::ImmediateAck => QuicFrame::ImmediateAck, + + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => QuicFrame::AckFrequency { + sequence_number: *sequence_number, + ack_eliciting_threshold: *ack_eliciting_threshold, + requested_max_ack_delay: *requested_max_ack_delay, + reordering_threshold: *reordering_threshold, + }, + Frame::PathAbandon { .. } => QuicFrame::Unknown { raw_frame_type: 0x15228c05, frame_type_value: None, @@ -1091,6 +1166,26 @@ impl std::fmt::Debug for Frame { write!(f, "HANDSHAKE_DONE")?; } + Frame::ImmediateAck => { + write!(f, "IMMEDIATE_ACK")?; + } + + Frame::AckFrequency { + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold, + } => { + write!( + f, + "ACK_FREQUENCY sequence={} threshold={} delay={} reordering={}", + sequence_number, + ack_eliciting_threshold, + requested_max_ack_delay, + reordering_threshold + )?; + } + Frame::PathAbandon { dcid_seq_num, error_code, @@ -1792,6 +1887,66 @@ mod tests { Ok(()) } + #[test] + fn immediate_ack() -> Result<()> { + let frame = Frame::ImmediateAck; + assert_eq!(format!("{:?}", &frame), "IMMEDIATE_ACK"); + + let mut buf = [0; 16]; + let len = frame.to_bytes(&mut buf)?; + assert_eq!(len, frame.wire_len()); + assert_eq!(&buf[..len], &[0x1f]); + + let mut encoded = Bytes::copy_from_slice(&buf[..len]); + assert_eq!( + (frame, len), + Frame::from_bytes(&mut encoded, PacketType::OneRTT)? + ); + for pkt_type in [ + PacketType::Initial, + PacketType::Handshake, + PacketType::ZeroRTT, + ] { + let mut encoded = Bytes::copy_from_slice(&buf[..len]); + assert!(Frame::from_bytes(&mut encoded, pkt_type).is_err()); + } + Ok(()) + } + + #[test] + fn ack_frequency() -> Result<()> { + let frame = Frame::AckFrequency { + sequence_number: 1, + ack_eliciting_threshold: 2, + requested_max_ack_delay: 3, + reordering_threshold: 4, + }; + assert_eq!( + format!("{:?}", &frame), + "ACK_FREQUENCY sequence=1 threshold=2 delay=3 reordering=4" + ); + + let mut buf = [0; 32]; + let len = frame.to_bytes(&mut buf)?; + assert_eq!(len, frame.wire_len()); + assert_eq!(len, 6); + + let mut encoded = Bytes::copy_from_slice(&buf[..len]); + assert_eq!( + (frame, len), + Frame::from_bytes(&mut encoded, PacketType::OneRTT)? + ); + for pkt_type in [ + PacketType::Initial, + PacketType::Handshake, + PacketType::ZeroRTT, + ] { + let mut encoded = Bytes::copy_from_slice(&buf[..len]); + assert!(Frame::from_bytes(&mut encoded, pkt_type).is_err()); + } + Ok(()) + } + #[test] fn path_abandon() -> Result<()> { let frame = Frame::PathAbandon { diff --git a/src/lib.rs b/src/lib.rs index d1d259dc5..29a0a95ea 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -497,6 +497,25 @@ impl Config { self.local_transport_params.max_ack_delay = cmp::min(v, VINT_MAX); } + /// Enable the ACK Frequency extension and advertise `min_ack_delay` in + /// microseconds. This must be configured before creating a connection. + pub fn enable_ack_frequency(&mut self, min_ack_delay: u64) -> Result<()> { + if min_ack_delay > VINT_MAX + || min_ack_delay + > self + .local_transport_params + .max_ack_delay + .saturating_mul(1000) + { + return Err(Error::InvalidConfig( + "min_ack_delay is outside the supported range".into(), + )); + } + + self.local_transport_params.min_ack_delay = Some(min_ack_delay); + Ok(()) + } + /// Set the maximum number of ack-eliciting packets the endpoint receives before /// sending an acknowledgment. /// The default value is `2`. @@ -1221,6 +1240,25 @@ mod tests { Ok(()) } + + #[test] + fn enable_ack_frequency_validates_min_ack_delay() -> Result<()> { + let mut config = Config::new()?; + config.set_max_ack_delay(25); + config.enable_ack_frequency(25_000)?; + assert_eq!(config.local_transport_params.min_ack_delay, Some(25_000)); + assert!(matches!( + config.enable_ack_frequency(25_001), + Err(Error::InvalidConfig(_)) + )); + + config.set_max_ack_delay(VINT_MAX); + assert!(matches!( + config.enable_ack_frequency(VINT_MAX + 1), + Err(Error::InvalidConfig(_)) + )); + Ok(()) + } } pub use crate::congestion_control::CongestionControlAlgorithm; @@ -1263,6 +1301,7 @@ mod ffi; #[path = "h3/connection.rs"] mod h3_connection; +mod ack_frequency; mod codec; pub mod endpoint; pub mod error; diff --git a/src/qlog/events.rs b/src/qlog/events.rs index 94b7c059c..b35063739 100644 --- a/src/qlog/events.rs +++ b/src/qlog/events.rs @@ -212,6 +212,7 @@ pub enum EventData { max_udp_payload_size: Option, ack_delay_exponent: Option, max_ack_delay: Option, + min_ack_delay: Option, active_connection_id_limit: Option, initial_max_data: Option, initial_max_stream_data_bidi_local: Option, @@ -1106,6 +1107,8 @@ pub enum QuicFrameTypeName { ConnectionClose, ApplicationClose, HandshakeDone, + ImmediateAck, + AckFrequency, Datagram, Unknown, } @@ -1213,6 +1216,15 @@ pub enum QuicFrame { HandshakeDone, + ImmediateAck, + + AckFrequency { + sequence_number: u64, + ack_eliciting_threshold: u64, + requested_max_ack_delay: u64, + reordering_threshold: u64, + }, + Datagram { length: u64, raw: Option, diff --git a/src/trans_param.rs b/src/trans_param.rs index b2db8262f..9a57f0ba7 100644 --- a/src/trans_param.rs +++ b/src/trans_param.rs @@ -89,6 +89,10 @@ pub struct TransportParams { /// in milliseconds by which the endpoint will delay sending acknowledgments. pub max_ack_delay: u64, + /// The minimum acknowledgment delay in microseconds that this endpoint can + /// honor. Presence of this parameter enables ACK Frequency negotiation. + pub min_ack_delay: Option, + /// The parameter is included if the endpoint does not support active /// connection migration on the address being used during the handshake. pub disable_active_migration: bool, @@ -266,11 +270,24 @@ impl TransportParams { tp.disable_encryption = true; } + 0xff04de1b => { + tp.min_ack_delay = Some(val.read_varint()?); + } + // Ignore unknown parameters. _ => (), } } + // Parameters can appear in any order, so validate this relationship + // only after all values have been decoded. + if tp + .min_ack_delay + .is_some_and(|v| v > tp.max_ack_delay.saturating_mul(1000)) + { + return Err(Error::TransportParameterError); + } + Ok((tp, len - buf.len())) } @@ -280,6 +297,15 @@ impl TransportParams { is_server: bool, mut buf: &mut [u8], ) -> Result { + if tp + .min_ack_delay + .is_some_and(|v| v > tp.max_ack_delay.saturating_mul(1000)) + { + return Err(Error::InvalidConfig( + "min_ack_delay exceeds max_ack_delay".into(), + )); + } + let len = buf.len(); if is_server { @@ -362,6 +388,12 @@ impl TransportParams { buf.write_varint(tp.max_ack_delay)?; } + if let Some(min_ack_delay) = tp.min_ack_delay { + buf.write_varint(0xff04de1b)?; + buf.write_varint(codec::encode_varint_len(min_ack_delay) as u64)?; + buf.write_varint(min_ack_delay)?; + } + if tp.disable_active_migration { buf.write_varint(0x000c)?; buf.write_varint(0)?; @@ -430,6 +462,7 @@ impl TransportParams { max_udp_payload_size: Some(self.max_udp_payload_size as u32), ack_delay_exponent: Some(self.ack_delay_exponent as u16), max_ack_delay: Some(self.max_ack_delay as u16), + min_ack_delay: self.min_ack_delay, active_connection_id_limit: Some(self.active_conn_id_limit as u32), initial_max_data: Some(self.initial_max_data), initial_max_stream_data_bidi_local: Some(self.initial_max_stream_data_bidi_local), @@ -470,6 +503,8 @@ impl Default for TransportParams { // milliseconds is assumed. max_ack_delay: 25, + min_ack_delay: None, + disable_active_migration: false, preferred_address: None, @@ -576,6 +611,7 @@ mod tests { initial_max_streams_uni: 100, ack_delay_exponent: 10, max_ack_delay: 2_u64.pow(8), + min_ack_delay: Some(1000), disable_active_migration: true, preferred_address: None, active_conn_id_limit: 12, @@ -620,6 +656,7 @@ mod tests { initial_max_streams_uni: 100, ack_delay_exponent: 10, max_ack_delay: 2_u64.pow(8), + min_ack_delay: Some(2000), disable_active_migration: true, preferred_address, active_conn_id_limit: 12, @@ -678,4 +715,35 @@ mod tests { Ok(()) } + + #[test] + fn min_ack_delay_must_not_exceed_max_ack_delay() { + let tp = TransportParams { + max_ack_delay: 1, + min_ack_delay: Some(1001), + ..TransportParams::default() + }; + let mut raw = [0; 128]; + assert!(matches!( + TransportParams::encode(&tp, false, &mut raw), + Err(Error::InvalidConfig(_)) + )); + + // min_ack_delay precedes max_ack_delay to exercise order-independent + // validation during decoding. + let mut raw = [0; 64]; + let raw_len = raw.len(); + let mut out = &mut raw[..]; + out.write_varint(0xff04de1b).unwrap(); + out.write_varint(2).unwrap(); + out.write_varint(1001).unwrap(); + out.write_varint(0x000b).unwrap(); + out.write_varint(1).unwrap(); + out.write_varint(1).unwrap(); + let written = raw_len - out.len(); + assert_eq!( + TransportParams::decode(&raw[..written], true), + Err(Error::TransportParameterError) + ); + } } diff --git a/src/window.rs b/src/window.rs index 52966f646..c723d94e1 100644 --- a/src/window.rs +++ b/src/window.rs @@ -46,7 +46,7 @@ impl SeqNumWindow { } /// Check whether the packet number exist or not - pub fn contains(&mut self, seq: u64) -> bool { + pub fn contains(&self, seq: u64) -> bool { // Sequence number is on the right end of the window. if seq > self.upper() { return false; @@ -61,6 +61,19 @@ impl SeqNumWindow { self.window & mask != 0 } + /// Return the first missing sequence number in the inclusive range that is + /// still represented by this window. Sequence numbers older than the + /// window are treated as already processed, matching `contains()`. + pub fn first_missing(&self, start: u64, end: u64) -> Option { + let start = start.max(self.lower); + let end = end.min(self.upper()); + if start > end { + return None; + } + + (start..=end).find(|seq| !self.contains(*seq)) + } + /// Return the largest sequence number fn upper(&self) -> u64 { self.lower.saturating_add(SEQ_NUM_WINDOW_SIZE) - 1 @@ -73,7 +86,7 @@ mod tests { #[test] fn seq_num_window_default() { - let mut win = SeqNumWindow::default(); + let win = SeqNumWindow::default(); assert!(!win.contains(0)); assert!(!win.contains(1)); } @@ -125,4 +138,19 @@ mod tests { assert!(!win.contains(max_seq - 1)); assert!(win.contains(max_seq - 128)); } + + #[test] + fn seq_num_window_first_missing() { + let mut win = SeqNumWindow::default(); + for seq in [0, 1, 3, 4] { + win.insert(seq); + } + assert_eq!(win.first_missing(0, 4), Some(2)); + win.insert(2); + assert_eq!(win.first_missing(0, 4), None); + + win.insert(200); + assert_eq!(win.first_missing(0, 72), None); + assert_eq!(win.first_missing(73, 200), Some(73)); + } }