feat: ack receiver timestamps

quinn-proto/src/config.rs

@@ -53,6 +53,8 @@ pub struct TransportConfig {
     pub(crate) mtu_discovery_config: Option<MtuDiscoveryConfig>,
     pub(crate) ack_frequency_config: Option<AckFrequencyConfig>,
 
+    pub(crate) ack_timestamp_config: Option<AckTimestampsConfig>,
+
     pub(crate) persistent_congestion_threshold: u32,
     pub(crate) keep_alive_interval: Option<Duration>,
     pub(crate) crypto_buffer_size: usize,
@@ -223,6 +225,15 @@ impl TransportConfig {
         self
     }
 
+    /// Specifies the ACK timestamp config.
+    /// Defaults to `None`, which disables receiving acknowledgement timestamps from the sender.
+    /// If `Some`, TransportParameters are sent to the peer to enable acknowledgement timestamps
+    /// if supported.
+    pub fn ack_timestamp_config(&mut self, value: Option<AckTimestampsConfig>) -> &mut Self {
+        self.ack_timestamp_config = value;
+        self
+    }
+
     /// Number of consecutive PTOs after which network is considered to be experiencing persistent congestion.
     pub fn persistent_congestion_threshold(&mut self, value: u32) -> &mut Self {
         self.persistent_congestion_threshold = value;
@@ -360,6 +371,8 @@ impl Default for TransportConfig {
             congestion_controller_factory: Arc::new(congestion::CubicConfig::default()),
 
             enable_segmentation_offload: true,
+
+            ack_timestamp_config: None,
         }
     }
 }
@@ -390,6 +403,7 @@ impl fmt::Debug for TransportConfig {
                 deterministic_packet_numbers: _,
             congestion_controller_factory: _,
             enable_segmentation_offload,
+            ack_timestamp_config,
         } = self;
         fmt.debug_struct("TransportConfig")
             .field("max_concurrent_bidi_streams", max_concurrent_bidi_streams)
@@ -416,10 +430,51 @@ impl fmt::Debug for TransportConfig {
             .field("datagram_send_buffer_size", datagram_send_buffer_size)
             .field("congestion_controller_factory", &"[ opaque ]")
             .field("enable_segmentation_offload", enable_segmentation_offload)
+            .field("ack_timestamp_config", ack_timestamp_config)
             .finish()
     }
 }
 
+/// Parameters for controlling the peer's acknowledgements with receiver timestamps.
+#[derive(Clone, Debug)]
+pub struct AckTimestampsConfig {
+    pub(crate) max_timestamps_per_ack: VarInt,
+    pub(crate) exponent: VarInt,
+    pub(crate) basis: std::time::Instant,
+}
+
+impl AckTimestampsConfig {
+    /// Sets the maximum number of timestamp entries per ACK frame.
+    pub fn max_timestamps_per_ack(&mut self, value: VarInt) -> &mut Self {
+        self.max_timestamps_per_ack = value;
+        self
+    }
+
+    /// Timestamp values are divided by the exponent value provided. This reduces the size of the
+    /// VARINT for loss in precision. A exponent of 0 represents microsecond precision.
+    pub fn exponent(&mut self, value: VarInt) -> &mut Self {
+        self.exponent = value;
+        self
+    }
+
+    /// Sets the time base for which all timestamps are anchored on.
+    /// Defaults to Instant::now of when the default struct was created.
+    pub fn basis(&mut self, instant: std::time::Instant) -> &mut Self {
+        self.basis = instant;
+        self
+    }
+}
+
+impl Default for AckTimestampsConfig {
+    fn default() -> Self {
+        Self {
+            max_timestamps_per_ack: 10u32.into(),
+            exponent: 0u32.into(),
+            basis: std::time::Instant::now(),
+        }
+    }
+}
+
 /// Parameters for controlling the peer's acknowledgement frequency
 ///
 /// The parameters provided in this config will be sent to the peer at the beginning of the

quinn-proto/src/congestion.rs

@@ -34,6 +34,20 @@ pub trait Controller: Send + Sync {
     ) {
     }
 
+    #[allow(unused_variables)]
+    /// Packet deliveries were confirmed with timestamps information.
+    fn on_ack_packet(
+        &mut self,
+        pn: u64,
+        now: Instant,
+        sent: Instant,
+        received: Option<Instant>,
+        bytes: u64,
+        app_limited: bool,
+        rtt: &RttEstimator,
+    ) {
+    }
+
     /// Packets are acked in batches, all with the same `now` argument. This indicates one of those batches has completed.
     #[allow(unused_variables)]
     fn on_end_acks(

quinn-proto/src/connection/mod.rs

@@ -20,8 +20,7 @@ use crate::{
     coding::BufMutExt,
     config::{ServerConfig, TransportConfig},
     crypto::{self, KeyPair, Keys, PacketKey},
-    frame,
-    frame::{Close, Datagram, FrameStruct},
+    frame::{self, Close, Datagram, FrameStruct},
     packet::{
         FixedLengthConnectionIdParser, Header, InitialHeader, InitialPacket, LongType, Packet,
         PacketNumber, PartialDecode, SpaceId,
@@ -37,6 +36,8 @@ use crate::{
     VarInt, MAX_STREAM_COUNT, MIN_INITIAL_SIZE, TIMER_GRANULARITY,
 };
 
+use crate::config::AckTimestampsConfig;
+
 mod ack_frequency;
 use ack_frequency::AckFrequencyState;
 
@@ -65,7 +66,10 @@ use paths::{PathData, PathResponses};
 
 mod send_buffer;
 
-mod spaces;
+mod receiver_timestamps;
+pub(crate) use receiver_timestamps::{PacketTimestamp, ReceiverTimestamps};
+
+pub(crate) mod spaces;
 #[cfg(fuzzing)]
 pub use spaces::Retransmits;
 #[cfg(not(fuzzing))]
@@ -227,6 +231,11 @@ pub struct Connection {
     /// no outgoing application data.
     app_limited: bool,
 
+    //
+    // Ack Receive Timestamps
+    //
+    peer_ack_timestamp_cfg: Option<AckTimestampsConfig>,
+
     streams: StreamsState,
     /// Surplus remote CIDs for future use on new paths
     rem_cids: CidQueue,
@@ -238,6 +247,8 @@ pub struct Connection {
     stats: ConnectionStats,
     /// QUIC version used for the connection.
     version: u32,
+    /// Created at time instant.
+    epoch: Instant,
 }
 
 impl Connection {
@@ -337,6 +348,8 @@ impl Connection {
                 &TransportParameters::default(),
             )),
 
+            peer_ack_timestamp_cfg: None,
+
             pto_count: 0,
 
             app_limited: false,
@@ -357,6 +370,7 @@ impl Connection {
             rng,
             stats: ConnectionStats::default(),
             version,
+            epoch: now,
         };
         if side.is_client() {
             // Kick off the connection
@@ -817,6 +831,7 @@ impl Connection {
                         &mut self.spaces[space_id],
                         buf,
                         &mut self.stats,
+                        None,
                     );
                 }
 
@@ -1343,6 +1358,13 @@ impl Connection {
         if ack.largest >= self.spaces[space].next_packet_number {
             return Err(TransportError::PROTOCOL_VIOLATION("unsent packet acked"));
         }
+
+        if ack.timestamps.is_some() != self.config.ack_timestamp_config.is_some() {
+            return Err(TransportError::PROTOCOL_VIOLATION(
+                "ack with timestamps expectation mismatched",
+            ));
+        }
+
         let new_largest = {
             let space = &mut self.spaces[space];
             if space
@@ -1371,13 +1393,21 @@ impl Connection {
             }
         }
 
+        let mut timestamp_iter = self.config.ack_timestamp_config.as_ref().map(|cfg| {
+            let decoder = ack.timestamp_iter(cfg.basis, cfg.exponent.0).unwrap();
+            let mut v: tinyvec::TinyVec<[PacketTimestamp; 10]> = tinyvec::TinyVec::new();
+            decoder.for_each(|elt| v.push(elt));
+            v.reverse();
+            v.into_iter().peekable()
+        });
+
         if newly_acked.is_empty() {
             return Ok(());
         }
 
         let mut ack_eliciting_acked = false;
         for packet in newly_acked.elts() {
-            if let Some(info) = self.spaces[space].take(packet) {
+            if let Some(mut info) = self.spaces[space].take(packet) {
                 if let Some(acked) = info.largest_acked {
                     // Assume ACKs for all packets below the largest acknowledged in `packet` have
                     // been received. This can cause the peer to spuriously retransmit if some of
@@ -1399,6 +1429,25 @@ impl Connection {
                 // Notify ack frequency that a packet was acked, because it might contain an ACK_FREQUENCY frame
                 self.ack_frequency.on_acked(packet);
 
+                if let Some(timestamp_iter) = timestamp_iter.as_mut() {
+                    while let Some(v) = timestamp_iter.peek() {
+                        match v.packet_number.cmp(&packet) {
+                            cmp::Ordering::Less => {
+                                let _ = timestamp_iter.next();
+                            }
+                            cmp::Ordering::Equal => {
+                                // Unwrap safety is guaranteed because a value was validated
+                                // to exist using peek
+                                let ts = timestamp_iter.next().unwrap();
+                                info.time_received = Some(ts.timestamp);
+                            }
+                            cmp::Ordering::Greater => {
+                                break;
+                            }
+                        }
+                    }
+                }
+
                 self.on_packet_acked(now, packet, info);
             }
         }
@@ -1497,6 +1546,16 @@ impl Connection {
                 self.app_limited,
                 &self.path.rtt,
             );
+
+            self.path.congestion.on_ack_packet(
+                pn,
+                now,
+                info.time_sent,
+                info.time_received,
+                info.size.into(),
+                self.app_limited,
+                &self.path.rtt,
+            );
         }
 
         // Update state for confirmed delivery of frames
@@ -2184,8 +2243,10 @@ impl Connection {
             }
             Ok((packet, number)) => {
                 let span = match number {
-                    Some(pn) => trace_span!("recv", space = ?packet.header.space(), pn),
-                    None => trace_span!("recv", space = ?packet.header.space()),
+                    Some(pn) => {
+                        trace_span!("recv", space = ?packet.header.space(), pn, side=?self.side)
+                    }
+                    None => trace_span!("recv", space = ?packet.header.space(), side=?self.side),
                 };
                 let _guard = span.enter();
 
@@ -3047,6 +3108,7 @@ impl Connection {
                 space,
                 buf,
                 &mut self.stats,
+                self.peer_ack_timestamp_cfg.clone(),
             );
         }
 
@@ -3231,6 +3293,7 @@ impl Connection {
         space: &mut PacketSpace,
         buf: &mut Vec<u8>,
         stats: &mut ConnectionStats,
+        timestamp_config: Option<AckTimestampsConfig>,
     ) {
         debug_assert!(!space.pending_acks.ranges().is_empty());
 
@@ -3255,7 +3318,21 @@ impl Connection {
             delay_micros
         );
 
-        frame::Ack::encode(delay as _, space.pending_acks.ranges(), ecn, buf);
+        frame::Ack::encode(
+            delay as _,
+            space.pending_acks.ranges(),
+            ecn,
+            timestamp_config.map(|cfg| {
+                (
+                    space.pending_acks.receiver_timestamps_as_ref().unwrap(),
+                    cfg.basis,
+                    cfg.exponent.0,
+                    cfg.max_timestamps_per_ack.0,
+                )
+            }),
+            buf,
+        );
+
         stats.frame_tx.acks += 1;
     }
 
@@ -3309,6 +3386,26 @@ impl Connection {
         self.path.mtud.on_peer_max_udp_payload_size_received(
             u16::try_from(self.peer_params.max_udp_payload_size.into_inner()).unwrap_or(u16::MAX),
         );
+
+        {
+            self.peer_ack_timestamp_cfg = if let (Some(max_timestamps_per_ack), Some(exponent)) = (
+                params.max_recv_timestamps_per_ack,
+                params.receive_timestamps_exponent,
+            ) {
+                for space in self.spaces.iter_mut() {
+                    space
+                        .pending_acks
+                        .set_receiver_timestamp(max_timestamps_per_ack.0 as usize);
+                }
+                Some(AckTimestampsConfig {
+                    exponent,
+                    max_timestamps_per_ack,
+                    basis: self.epoch,
+                })
+            } else {
+                None
+            };
+        }
     }
 
     fn decrypt_packet(

quinn-proto/src/connection/packet_builder.rs

@@ -210,6 +210,8 @@ impl PacketBuilder {
             ack_eliciting,
             retransmits: sent.retransmits,
             stream_frames: sent.stream_frames,
+
+            time_received: None,
         };
 
         conn.path