1#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)]
9use std::{
10 cmp,
11 collections::VecDeque,
12 convert::TryFrom,
13 fmt, io, mem,
14 net::{IpAddr, SocketAddr},
15 sync::Arc,
16};
17
18use bytes::{Bytes, BytesMut};
19use frame::StreamMetaVec;
20use rand::{Rng, SeedableRng, rngs::StdRng};
23use thiserror::Error;
24use tracing::{debug, error, info, trace, trace_span, warn};
25
26use crate::{
27 Dir, Duration, EndpointConfig, Frame, INITIAL_MTU, Instant, MAX_CID_SIZE, MAX_STREAM_COUNT,
28 MIN_INITIAL_SIZE, MtuDiscoveryConfig, Side, StreamId, TIMER_GRANULARITY, TokenStore, Transmit,
29 TransportError, TransportErrorCode, VarInt, VarIntBoundsExceeded,
30 cid_generator::ConnectionIdGenerator,
31 cid_queue::CidQueue,
32 coding::BufMutExt,
33 config::{ServerConfig, TransportConfig},
34 crypto::{self, KeyPair, Keys, PacketKey},
35 endpoint::AddressDiscoveryStats,
36 frame::{self, Close, Datagram, FrameStruct, NewToken},
37 nat_traversal_api::PeerId,
38 packet::{
39 FixedLengthConnectionIdParser, Header, InitialHeader, InitialPacket, LongType, Packet,
40 PacketNumber, PartialDecode, SpaceId,
41 },
42 range_set::ArrayRangeSet,
43 shared::{
44 ConnectionEvent, ConnectionEventInner, ConnectionId, DatagramConnectionEvent, EcnCodepoint,
45 EndpointEvent, EndpointEventInner,
46 },
47 token::ResetToken,
48 transport_parameters::TransportParameters,
49};
50
51fn allow_loopback_from_env() -> bool {
52 matches!(
53 std::env::var("ANT_QUIC_ALLOW_LOOPBACK")
54 .unwrap_or_default()
55 .trim()
56 .to_ascii_lowercase()
57 .as_str(),
58 "1" | "true" | "yes"
59 )
60}
61
62mod ack_frequency;
63use ack_frequency::AckFrequencyState;
64
65pub mod port_prediction;
66pub use self::port_prediction::{PortPredictor, PortPredictorConfig};
67
68pub(crate) mod nat_traversal;
69use nat_traversal::NatTraversalState;
70pub(crate) use nat_traversal::{CoordinationPhase, NatTraversalError};
72
73mod assembler;
74pub use assembler::Chunk;
75
76mod cid_state;
77use cid_state::CidState;
78
79mod datagrams;
80use datagrams::DatagramState;
81pub use datagrams::{Datagrams, SendDatagramError};
82
83mod mtud;
84
85mod pacing;
86
87mod packet_builder;
88use packet_builder::PacketBuilder;
89
90mod packet_crypto;
91use packet_crypto::{PrevCrypto, ZeroRttCrypto};
92
93mod paths;
94pub use paths::RttEstimator;
95use paths::{PathData, PathResponses};
96
97mod send_buffer;
98
99mod spaces;
100#[cfg(fuzzing)]
101pub use spaces::Retransmits;
102#[cfg(not(fuzzing))]
103use spaces::Retransmits;
104use spaces::{PacketNumberFilter, PacketSpace, SendableFrames, SentPacket, ThinRetransmits};
105
106mod stats;
107pub use stats::{ConnectionStats, DatagramDropStats, FrameStats, PathStats, UdpStats};
108
109mod streams;
110#[cfg(fuzzing)]
111pub use streams::StreamsState;
112#[cfg(not(fuzzing))]
113use streams::StreamsState;
114pub use streams::{
115 Chunks, ClosedStream, FinishError, ReadError, ReadableError, RecvStream, SendStream,
116 ShouldTransmit, StreamEvent, Streams, WriteError, Written,
117};
118
119mod timer;
120use crate::congestion::Controller;
121use timer::{Timer, TimerTable};
122
123pub struct Connection {
163 endpoint_config: Arc<EndpointConfig>,
164 config: Arc<TransportConfig>,
165 rng: StdRng,
166 crypto: Box<dyn crypto::Session>,
167 handshake_cid: ConnectionId,
169 rem_handshake_cid: ConnectionId,
171 local_ip: Option<IpAddr>,
174 path: PathData,
175 allow_mtud: bool,
177 prev_path: Option<(ConnectionId, PathData)>,
178 state: State,
179 side: ConnectionSide,
180 zero_rtt_enabled: bool,
182 zero_rtt_crypto: Option<ZeroRttCrypto>,
184 key_phase: bool,
185 key_phase_size: u64,
187 peer_params: TransportParameters,
189 orig_rem_cid: ConnectionId,
191 initial_dst_cid: ConnectionId,
193 retry_src_cid: Option<ConnectionId>,
196 lost_packets: u64,
198 events: VecDeque<Event>,
199 endpoint_events: VecDeque<EndpointEventInner>,
200 spin_enabled: bool,
202 spin: bool,
204 spaces: [PacketSpace; 3],
206 highest_space: SpaceId,
208 prev_crypto: Option<PrevCrypto>,
210 next_crypto: Option<KeyPair<Box<dyn PacketKey>>>,
215 accepted_0rtt: bool,
216 permit_idle_reset: bool,
218 idle_timeout: Option<Duration>,
220 timers: TimerTable,
221 authentication_failures: u64,
223 error: Option<ConnectionError>,
225 packet_number_filter: PacketNumberFilter,
227
228 path_responses: PathResponses,
233 close: bool,
234
235 ack_frequency: AckFrequencyState,
239
240 pto_count: u32,
245
246 receiving_ecn: bool,
251 total_authed_packets: u64,
253 app_limited: bool,
256
257 streams: StreamsState,
258 rem_cids: CidQueue,
260 local_cid_state: CidState,
262 datagrams: DatagramState,
264 stats: ConnectionStats,
266 version: u32,
268
269 nat_traversal: Option<NatTraversalState>,
271
272 nat_traversal_frame_config: frame::nat_traversal_unified::NatTraversalFrameConfig,
274
275 address_discovery_state: Option<AddressDiscoveryState>,
277
278 pqc_state: PqcState,
280
281 #[cfg(feature = "trace")]
283 trace_context: crate::tracing::TraceContext,
284
285 #[cfg(feature = "trace")]
287 event_log: Arc<crate::tracing::EventLog>,
288
289 #[cfg(feature = "__qlog")]
291 qlog_streamer: Option<Box<dyn std::io::Write + Send + Sync>>,
292
293 peer_id_for_tokens: Option<PeerId>,
295 delay_new_token_until_binding: bool,
297}
298
299impl Connection {
300 pub(crate) fn new(
301 endpoint_config: Arc<EndpointConfig>,
302 config: Arc<TransportConfig>,
303 init_cid: ConnectionId,
304 loc_cid: ConnectionId,
305 rem_cid: ConnectionId,
306 remote: SocketAddr,
307 local_ip: Option<IpAddr>,
308 crypto: Box<dyn crypto::Session>,
309 cid_gen: &dyn ConnectionIdGenerator,
310 now: Instant,
311 version: u32,
312 allow_mtud: bool,
313 rng_seed: [u8; 32],
314 side_args: SideArgs,
315 ) -> Self {
316 let pref_addr_cid = side_args.pref_addr_cid();
317 let path_validated = side_args.path_validated();
318 let connection_side = ConnectionSide::from(side_args);
319 let side = connection_side.side();
320 let initial_space = PacketSpace {
321 crypto: Some(crypto.initial_keys(&init_cid, side)),
322 ..PacketSpace::new(now)
323 };
324 let state = State::Handshake(state::Handshake {
325 rem_cid_set: side.is_server(),
326 expected_token: Bytes::new(),
327 client_hello: None,
328 });
329 let mut rng = StdRng::from_seed(rng_seed);
330 let mut this = Self {
331 endpoint_config,
332 crypto,
333 handshake_cid: loc_cid,
334 rem_handshake_cid: rem_cid,
335 local_cid_state: CidState::new(
336 cid_gen.cid_len(),
337 cid_gen.cid_lifetime(),
338 now,
339 if pref_addr_cid.is_some() { 2 } else { 1 },
340 ),
341 path: PathData::new(remote, allow_mtud, None, now, &config),
342 allow_mtud,
343 local_ip,
344 prev_path: None,
345 state,
346 side: connection_side,
347 zero_rtt_enabled: false,
348 zero_rtt_crypto: None,
349 key_phase: false,
350 key_phase_size: rng.gen_range(10..1000),
357 peer_params: TransportParameters::default(),
358 orig_rem_cid: rem_cid,
359 initial_dst_cid: init_cid,
360 retry_src_cid: None,
361 lost_packets: 0,
362 events: VecDeque::new(),
363 endpoint_events: VecDeque::new(),
364 spin_enabled: config.allow_spin && rng.gen_ratio(7, 8),
365 spin: false,
366 spaces: [initial_space, PacketSpace::new(now), PacketSpace::new(now)],
367 highest_space: SpaceId::Initial,
368 prev_crypto: None,
369 next_crypto: None,
370 accepted_0rtt: false,
371 permit_idle_reset: true,
372 idle_timeout: match config.max_idle_timeout {
373 None | Some(VarInt(0)) => None,
374 Some(dur) => Some(Duration::from_millis(dur.0)),
375 },
376 timers: TimerTable::default(),
377 authentication_failures: 0,
378 error: None,
379 #[cfg(test)]
380 packet_number_filter: match config.deterministic_packet_numbers {
381 false => PacketNumberFilter::new(&mut rng),
382 true => PacketNumberFilter::disabled(),
383 },
384 #[cfg(not(test))]
385 packet_number_filter: PacketNumberFilter::new(&mut rng),
386
387 path_responses: PathResponses::default(),
388 close: false,
389
390 ack_frequency: AckFrequencyState::new(get_max_ack_delay(
391 &TransportParameters::default(),
392 )),
393
394 pto_count: 0,
395
396 app_limited: false,
397 receiving_ecn: false,
398 total_authed_packets: 0,
399
400 streams: StreamsState::new(
401 side,
402 config.max_concurrent_uni_streams,
403 config.max_concurrent_bidi_streams,
404 config.send_window,
405 config.receive_window,
406 config.stream_receive_window,
407 ),
408 datagrams: DatagramState::default(),
409 config,
410 rem_cids: CidQueue::new(rem_cid),
411 rng,
412 stats: ConnectionStats::default(),
413 version,
414 nat_traversal: None, nat_traversal_frame_config:
416 frame::nat_traversal_unified::NatTraversalFrameConfig::default(),
417 address_discovery_state: {
418 Some(AddressDiscoveryState::new(
421 &crate::transport_parameters::AddressDiscoveryConfig::default(),
422 now,
423 ))
424 },
425 pqc_state: PqcState::new(),
426
427 #[cfg(feature = "trace")]
428 trace_context: crate::tracing::TraceContext::new(crate::tracing::TraceId::new()),
429
430 #[cfg(feature = "trace")]
431 event_log: crate::tracing::global_log(),
432
433 #[cfg(feature = "__qlog")]
434 qlog_streamer: None,
435
436 peer_id_for_tokens: None,
437 delay_new_token_until_binding: false,
438 };
439
440 #[cfg(feature = "trace")]
442 {
443 use crate::trace_event;
444 use crate::tracing::{Event, EventData, socket_addr_to_bytes, timestamp_now};
445 let _peer_id = {
447 let mut id = [0u8; 32];
448 let addr_bytes = match remote {
449 SocketAddr::V4(addr) => addr.ip().octets().to_vec(),
450 SocketAddr::V6(addr) => addr.ip().octets().to_vec(),
451 };
452 id[..addr_bytes.len().min(32)]
453 .copy_from_slice(&addr_bytes[..addr_bytes.len().min(32)]);
454 id
455 };
456
457 let (addr_bytes, addr_type) = socket_addr_to_bytes(remote);
458 trace_event!(
459 &this.event_log,
460 Event {
461 timestamp: timestamp_now(),
462 trace_id: this.trace_context.trace_id(),
463 sequence: 0,
464 _padding: 0,
465 node_id: [0u8; 32], event_data: EventData::ConnInit {
467 endpoint_bytes: addr_bytes,
468 addr_type,
469 _padding: [0u8; 45],
470 },
471 }
472 );
473 }
474
475 if path_validated {
476 this.on_path_validated();
477 }
478 if side.is_client() {
479 this.write_crypto();
481 this.init_0rtt();
482 }
483 this
484 }
485
486 #[cfg(feature = "__qlog")]
488 pub fn set_qlog(
489 &mut self,
490 writer: Box<dyn std::io::Write + Send + Sync>,
491 _title: Option<String>,
492 _description: Option<String>,
493 _now: Instant,
494 ) {
495 self.qlog_streamer = Some(writer);
496 }
497
498 #[cfg(feature = "__qlog")]
500 fn emit_qlog_recovery_metrics(&mut self, _now: Instant) {
501 }
504
505 #[must_use]
513 pub fn poll_timeout(&mut self) -> Option<Instant> {
514 let mut next_timeout = self.timers.next_timeout();
515
516 if let Some(nat_state) = &self.nat_traversal {
518 if let Some(nat_timeout) = nat_state.get_next_timeout(Instant::now()) {
519 self.timers.set(Timer::NatTraversal, nat_timeout);
521 next_timeout = Some(next_timeout.map_or(nat_timeout, |t| t.min(nat_timeout)));
522 }
523 }
524
525 next_timeout
526 }
527
528 #[must_use]
534 pub fn poll(&mut self) -> Option<Event> {
535 if let Some(x) = self.events.pop_front() {
536 return Some(x);
537 }
538
539 if let Some(event) = self.streams.poll() {
540 return Some(Event::Stream(event));
541 }
542
543 if let Some(err) = self.error.take() {
544 return Some(Event::ConnectionLost { reason: err });
545 }
546
547 None
548 }
549
550 #[must_use]
552 pub fn poll_endpoint_events(&mut self) -> Option<EndpointEvent> {
553 self.endpoint_events.pop_front().map(EndpointEvent)
554 }
555
556 #[must_use]
558 pub fn streams(&mut self) -> Streams<'_> {
559 Streams {
560 state: &mut self.streams,
561 conn_state: &self.state,
562 }
563 }
564
565 #[must_use]
569 pub fn recv_stream(&mut self, id: StreamId) -> RecvStream<'_> {
570 assert!(id.dir() == Dir::Bi || id.initiator() != self.side.side());
571 RecvStream {
572 id,
573 state: &mut self.streams,
574 pending: &mut self.spaces[SpaceId::Data].pending,
575 }
576 }
577
578 #[must_use]
580 pub fn send_stream(&mut self, id: StreamId) -> SendStream<'_> {
581 assert!(id.dir() == Dir::Bi || id.initiator() == self.side.side());
582 SendStream {
583 id,
584 state: &mut self.streams,
585 pending: &mut self.spaces[SpaceId::Data].pending,
586 conn_state: &self.state,
587 }
588 }
589
590 #[must_use]
600 pub fn poll_transmit(
601 &mut self,
602 now: Instant,
603 max_datagrams: usize,
604 buf: &mut Vec<u8>,
605 ) -> Option<Transmit> {
606 assert!(max_datagrams != 0);
607 let max_datagrams = match self.config.enable_segmentation_offload {
608 false => 1,
609 true => max_datagrams,
610 };
611
612 let mut num_datagrams = 0;
613 let mut datagram_start = 0;
616 let mut segment_size = usize::from(self.path.current_mtu());
617
618 if let Some(nat_traversal) = &mut self.nat_traversal {
620 if nat_traversal.check_coordination_timeout(now) {
621 trace!("NAT traversal coordination timed out, may retry");
622 }
623 let expired = nat_traversal.check_validation_timeouts(now);
625 if !expired.is_empty() {
626 debug!(
627 "Cleaned up {} expired NAT traversal validations",
628 expired.len()
629 );
630 }
631 }
632
633 self.check_for_address_observations(now);
635
636 if let Some(challenge) = self.send_nat_traversal_challenge(now, buf) {
638 return Some(challenge);
639 }
640
641 if let Some(challenge) = self.send_path_challenge(now, buf) {
642 return Some(challenge);
643 }
644
645 for space in SpaceId::iter() {
647 let request_immediate_ack =
648 space == SpaceId::Data && self.peer_supports_ack_frequency();
649 self.spaces[space].maybe_queue_probe(request_immediate_ack, &self.streams);
650 }
651
652 let close = match self.state {
654 State::Drained => {
655 self.app_limited = true;
656 return None;
657 }
658 State::Draining | State::Closed(_) => {
659 if !self.close {
662 self.app_limited = true;
663 return None;
664 }
665 true
666 }
667 _ => false,
668 };
669
670 if let Some(config) = &self.config.ack_frequency_config {
672 self.spaces[SpaceId::Data].pending.ack_frequency = self
673 .ack_frequency
674 .should_send_ack_frequency(self.path.rtt.get(), config, &self.peer_params)
675 && self.highest_space == SpaceId::Data
676 && self.peer_supports_ack_frequency();
677 }
678
679 let mut buf_capacity = 0;
683
684 let mut coalesce = true;
685 let mut builder_storage: Option<PacketBuilder> = None;
686 let mut sent_frames = None;
687 let mut pad_datagram = false;
688 let mut pad_datagram_to_mtu = false;
689 let mut congestion_blocked = false;
690
691 let mut space_idx = 0;
693 let spaces = [SpaceId::Initial, SpaceId::Handshake, SpaceId::Data];
694 while space_idx < spaces.len() {
697 let space_id = spaces[space_idx];
698 let pn = self.packet_number_filter.peek(&self.spaces[SpaceId::Data]);
705 let frame_space_1rtt =
706 segment_size.saturating_sub(self.predict_1rtt_overhead(Some(pn)));
707
708 let can_send = self.space_can_send(space_id, frame_space_1rtt);
710 if can_send.is_empty() && (!close || self.spaces[space_id].crypto.is_none()) {
711 space_idx += 1;
712 continue;
713 }
714
715 let mut ack_eliciting = !self.spaces[space_id].pending.is_empty(&self.streams)
716 || self.spaces[space_id].ping_pending
717 || self.spaces[space_id].immediate_ack_pending;
718 if space_id == SpaceId::Data {
719 ack_eliciting |= self.can_send_1rtt(frame_space_1rtt);
720 }
721
722 pad_datagram_to_mtu |= space_id == SpaceId::Data && self.config.pad_to_mtu;
723
724 let buf_end = if let Some(builder) = &builder_storage {
728 buf.len().max(builder.min_size) + builder.tag_len
729 } else {
730 buf.len()
731 };
732
733 let tag_len = if let Some(ref crypto) = self.spaces[space_id].crypto {
734 crypto.packet.local.tag_len()
735 } else if space_id == SpaceId::Data {
736 match self.zero_rtt_crypto.as_ref() {
737 Some(crypto) => crypto.packet.tag_len(),
738 None => {
739 error!(
741 "sending packets in the application data space requires known 0-RTT or 1-RTT keys"
742 );
743 return None;
744 }
745 }
746 } else {
747 unreachable!("tried to send {:?} packet without keys", space_id)
748 };
749 if !coalesce || buf_capacity - buf_end < MIN_PACKET_SPACE + tag_len {
750 if num_datagrams >= max_datagrams {
754 break;
756 }
757
758 if self
765 .path
766 .anti_amplification_blocked(segment_size as u64 * (num_datagrams as u64) + 1)
767 {
768 trace!("blocked by anti-amplification");
769 break;
770 }
771
772 if ack_eliciting && self.spaces[space_id].loss_probes == 0 {
775 let untracked_bytes = if let Some(builder) = &builder_storage {
777 buf_capacity - builder.partial_encode.start
778 } else {
779 0
780 } as u64;
781 debug_assert!(untracked_bytes <= segment_size as u64);
782
783 let bytes_to_send = segment_size as u64 + untracked_bytes;
784 if self.path.in_flight.bytes + bytes_to_send >= self.path.congestion.window() {
785 space_idx += 1;
786 congestion_blocked = true;
787 trace!("blocked by congestion control");
790 continue;
791 }
792
793 let smoothed_rtt = self.path.rtt.get();
795 if let Some(delay) = self.path.pacing.delay(
796 smoothed_rtt,
797 bytes_to_send,
798 self.path.current_mtu(),
799 self.path.congestion.window(),
800 now,
801 ) {
802 self.timers.set(Timer::Pacing, delay);
803 congestion_blocked = true;
804 trace!("blocked by pacing");
807 break;
808 }
809 }
810
811 if let Some(mut builder) = builder_storage.take() {
813 if pad_datagram {
814 let min_size = self.pqc_state.min_initial_size();
815 builder.pad_to(min_size);
816 }
817
818 if num_datagrams > 1 || pad_datagram_to_mtu {
819 const MAX_PADDING: usize = 16;
832 let packet_len_unpadded = cmp::max(builder.min_size, buf.len())
833 - datagram_start
834 + builder.tag_len;
835 if (packet_len_unpadded + MAX_PADDING < segment_size
836 && !pad_datagram_to_mtu)
837 || datagram_start + segment_size > buf_capacity
838 {
839 trace!(
840 "GSO truncated by demand for {} padding bytes or loss probe",
841 segment_size - packet_len_unpadded
842 );
843 builder_storage = Some(builder);
844 break;
845 }
846
847 builder.pad_to(segment_size as u16);
850 }
851
852 builder.finish_and_track(now, self, sent_frames.take(), buf);
853
854 if num_datagrams == 1 {
855 segment_size = buf.len();
862 buf_capacity = buf.len();
865
866 if space_id == SpaceId::Data {
873 let frame_space_1rtt =
874 segment_size.saturating_sub(self.predict_1rtt_overhead(Some(pn)));
875 if self.space_can_send(space_id, frame_space_1rtt).is_empty() {
876 break;
877 }
878 }
879 }
880 }
881
882 let next_datagram_size_limit = match self.spaces[space_id].loss_probes {
884 0 => segment_size,
885 _ => {
886 self.spaces[space_id].loss_probes -= 1;
887 std::cmp::min(segment_size, usize::from(INITIAL_MTU))
891 }
892 };
893 buf_capacity += next_datagram_size_limit;
894 if buf.capacity() < buf_capacity {
895 buf.reserve(max_datagrams * segment_size);
904 }
905 num_datagrams += 1;
906 coalesce = true;
907 pad_datagram = false;
908 datagram_start = buf.len();
909
910 debug_assert_eq!(
911 datagram_start % segment_size,
912 0,
913 "datagrams in a GSO batch must be aligned to the segment size"
914 );
915 } else {
916 if let Some(builder) = builder_storage.take() {
920 builder.finish_and_track(now, self, sent_frames.take(), buf);
921 }
922 }
923
924 debug_assert!(buf_capacity - buf.len() >= MIN_PACKET_SPACE);
925
926 if self.spaces[SpaceId::Initial].crypto.is_some()
931 && space_id == SpaceId::Handshake
932 && self.side.is_client()
933 {
934 self.discard_space(now, SpaceId::Initial);
937 }
938 if let Some(ref mut prev) = self.prev_crypto {
939 prev.update_unacked = false;
940 }
941
942 debug_assert!(
943 builder_storage.is_none() && sent_frames.is_none(),
944 "Previous packet must have been finished"
945 );
946
947 let builder = builder_storage.insert(PacketBuilder::new(
948 now,
949 space_id,
950 self.rem_cids.active(),
951 buf,
952 buf_capacity,
953 datagram_start,
954 ack_eliciting,
955 self,
956 )?);
957 coalesce = coalesce && !builder.short_header;
958
959 let should_adjust_coalescing = self
961 .pqc_state
962 .should_adjust_coalescing(buf.len() - datagram_start, space_id);
963
964 if should_adjust_coalescing {
965 coalesce = false;
966 trace!("Disabling coalescing for PQC handshake in {:?}", space_id);
967 }
968
969 pad_datagram |=
971 space_id == SpaceId::Initial && (self.side.is_client() || ack_eliciting);
972
973 if close {
974 trace!("sending CONNECTION_CLOSE");
975 if !self.spaces[space_id].pending_acks.ranges().is_empty() {
980 if Self::populate_acks(
981 now,
982 self.receiving_ecn,
983 &mut SentFrames::default(),
984 &mut self.spaces[space_id],
985 buf,
986 &mut self.stats,
987 )
988 .is_err()
989 {
990 self.handle_encode_error(now, "ACK (close)");
991 return None;
992 }
993 }
994
995 debug_assert!(
999 buf.len() + frame::ConnectionClose::SIZE_BOUND < builder.max_size,
1000 "ACKs should leave space for ConnectionClose"
1001 );
1002 if buf.len() + frame::ConnectionClose::SIZE_BOUND < builder.max_size {
1003 let max_frame_size = builder.max_size - buf.len();
1004 match self.state {
1005 State::Closed(state::Closed { ref reason }) => {
1006 let result = if space_id == SpaceId::Data || reason.is_transport_layer()
1007 {
1008 reason.try_encode(buf, max_frame_size)
1009 } else {
1010 frame::ConnectionClose {
1011 error_code: TransportErrorCode::APPLICATION_ERROR,
1012 frame_type: None,
1013 reason: Bytes::new(),
1014 }
1015 .try_encode(buf, max_frame_size)
1016 };
1017 if result.is_err() {
1018 self.handle_encode_error(now, "ConnectionClose");
1019 return None;
1020 }
1021 }
1022 State::Draining => {
1023 if (frame::ConnectionClose {
1024 error_code: TransportErrorCode::NO_ERROR,
1025 frame_type: None,
1026 reason: Bytes::new(),
1027 })
1028 .try_encode(buf, max_frame_size)
1029 .is_err()
1030 {
1031 self.handle_encode_error(now, "ConnectionClose (draining)");
1032 return None;
1033 }
1034 }
1035 _ => unreachable!(
1036 "tried to make a close packet when the connection wasn't closed"
1037 ),
1038 }
1039 }
1040 if space_id == self.highest_space {
1041 self.close = false;
1043 break;
1045 } else {
1046 space_idx += 1;
1050 continue;
1051 }
1052 }
1053
1054 if space_id == SpaceId::Data && num_datagrams == 1 {
1057 if let Some((token, remote)) = self.path_responses.pop_off_path(self.path.remote) {
1058 let mut builder = builder_storage.take().unwrap();
1061 trace!("PATH_RESPONSE {:08x} (off-path)", token);
1062 if !self.encode_or_close(
1063 now,
1064 frame::FrameType::PATH_RESPONSE.try_encode(buf),
1065 "PATH_RESPONSE (off-path)",
1066 ) {
1067 return None;
1068 }
1069 buf.write(token);
1070 self.stats.frame_tx.path_response += 1;
1071 let min_size = self.pqc_state.min_initial_size();
1072 builder.pad_to(min_size);
1073 builder.finish_and_track(
1074 now,
1075 self,
1076 Some(SentFrames {
1077 non_retransmits: true,
1078 ..SentFrames::default()
1079 }),
1080 buf,
1081 );
1082 self.stats.udp_tx.on_sent(1, buf.len());
1083
1084 #[cfg(feature = "trace")]
1086 {
1087 use crate::trace_packet_sent;
1088 trace_packet_sent!(
1090 &self.event_log,
1091 self.trace_context.trace_id(),
1092 buf.len() as u32,
1093 0 );
1095 }
1096
1097 return Some(Transmit {
1098 destination: remote,
1099 size: buf.len(),
1100 ecn: None,
1101 segment_size: None,
1102 src_ip: self.local_ip,
1103 });
1104 }
1105 }
1106
1107 if space_id == SpaceId::Data && self.address_discovery_state.is_some() {
1109 let peer_supports = self.peer_params.address_discovery.is_some();
1110
1111 if let Some(state) = &mut self.address_discovery_state {
1112 if peer_supports {
1113 if let Some(frame) = state.queue_observed_address_frame(0, self.path.remote)
1114 {
1115 self.spaces[space_id]
1116 .pending
1117 .outbound_observations
1118 .push(frame);
1119 }
1120 }
1121 }
1122 }
1123
1124 let sent =
1125 self.populate_packet(now, space_id, buf, builder.max_size, builder.exact_number);
1126
1127 debug_assert!(
1134 !(sent.is_ack_only(&self.streams)
1135 && !can_send.acks
1136 && can_send.other
1137 && (buf_capacity - builder.datagram_start) == self.path.current_mtu() as usize
1138 && self.datagrams.outgoing.is_empty()),
1139 "SendableFrames was {can_send:?}, but only ACKs have been written"
1140 );
1141 pad_datagram |= sent.requires_padding;
1142
1143 if sent.largest_acked.is_some() {
1144 self.spaces[space_id].pending_acks.acks_sent();
1145 self.timers.stop(Timer::MaxAckDelay);
1146 }
1147
1148 sent_frames = Some(sent);
1150
1151 }
1154
1155 if let Some(mut builder) = builder_storage {
1157 if pad_datagram {
1158 let min_size = self.pqc_state.min_initial_size();
1159 builder.pad_to(min_size);
1160 }
1161
1162 if pad_datagram_to_mtu && buf_capacity >= datagram_start + segment_size {
1168 builder.pad_to(segment_size as u16);
1169 }
1170
1171 let last_packet_number = builder.exact_number;
1172 builder.finish_and_track(now, self, sent_frames, buf);
1173 self.path
1174 .congestion
1175 .on_sent(now, buf.len() as u64, last_packet_number);
1176
1177 #[cfg(feature = "__qlog")]
1178 self.emit_qlog_recovery_metrics(now);
1179 }
1180
1181 self.app_limited = buf.is_empty() && !congestion_blocked;
1182
1183 if buf.is_empty() && self.state.is_established() {
1185 let space_id = SpaceId::Data;
1186 let probe_size = self
1187 .path
1188 .mtud
1189 .poll_transmit(now, self.packet_number_filter.peek(&self.spaces[space_id]))?;
1190
1191 let buf_capacity = probe_size as usize;
1192 buf.reserve(buf_capacity);
1193
1194 let mut builder = PacketBuilder::new(
1195 now,
1196 space_id,
1197 self.rem_cids.active(),
1198 buf,
1199 buf_capacity,
1200 0,
1201 true,
1202 self,
1203 )?;
1204
1205 if !self.encode_or_close(now, frame::FrameType::PING.try_encode(buf), "PING (MTU)") {
1207 return None;
1208 }
1209 self.stats.frame_tx.ping += 1;
1210
1211 if self.peer_supports_ack_frequency() {
1213 if !self.encode_or_close(
1214 now,
1215 frame::FrameType::IMMEDIATE_ACK.try_encode(buf),
1216 "IMMEDIATE_ACK (MTU)",
1217 ) {
1218 return None;
1219 }
1220 self.stats.frame_tx.immediate_ack += 1;
1221 }
1222
1223 builder.pad_to(probe_size);
1224 let sent_frames = SentFrames {
1225 non_retransmits: true,
1226 ..Default::default()
1227 };
1228 builder.finish_and_track(now, self, Some(sent_frames), buf);
1229
1230 self.stats.path.sent_plpmtud_probes += 1;
1231 num_datagrams = 1;
1232
1233 trace!(?probe_size, "writing MTUD probe");
1234 }
1235
1236 if buf.is_empty() {
1237 return None;
1238 }
1239
1240 trace!("sending {} bytes in {} datagrams", buf.len(), num_datagrams);
1241 self.path.total_sent = self.path.total_sent.saturating_add(buf.len() as u64);
1242
1243 self.stats.udp_tx.on_sent(num_datagrams as u64, buf.len());
1244
1245 #[cfg(feature = "trace")]
1247 {
1248 use crate::trace_packet_sent;
1249 let packet_num = self.spaces[SpaceId::Data]
1252 .next_packet_number
1253 .saturating_sub(1);
1254 trace_packet_sent!(
1255 &self.event_log,
1256 self.trace_context.trace_id(),
1257 buf.len() as u32,
1258 packet_num
1259 );
1260 }
1261
1262 Some(Transmit {
1263 destination: self.path.remote,
1264 size: buf.len(),
1265 ecn: if self.path.sending_ecn {
1266 Some(EcnCodepoint::Ect0)
1267 } else {
1268 None
1269 },
1270 segment_size: match num_datagrams {
1271 1 => None,
1272 _ => Some(segment_size),
1273 },
1274 src_ip: self.local_ip,
1275 })
1276 }
1277
1278 fn send_coordination_request(&mut self, _now: Instant, _buf: &mut Vec<u8>) -> Option<Transmit> {
1280 let nat = self.nat_traversal.as_mut()?;
1282 if !nat.should_send_punch_request() {
1283 return None;
1284 }
1285
1286 let coord = nat.coordination.as_ref()?;
1287 let round = coord.round;
1288 if coord.punch_targets.is_empty() {
1289 return None;
1290 }
1291
1292 trace!(
1293 "queuing PUNCH_ME_NOW round {} with {} targets",
1294 round,
1295 coord.punch_targets.len()
1296 );
1297
1298 for target in &coord.punch_targets {
1300 let punch = frame::PunchMeNow {
1301 round,
1302 paired_with_sequence_number: target.remote_sequence,
1303 address: target.remote_addr,
1304 target_peer_id: None,
1305 };
1306 self.spaces[SpaceId::Data].pending.punch_me_now.push(punch);
1307 }
1308
1309 nat.mark_punch_request_sent();
1311
1312 None
1314 }
1315
1316 fn send_coordinated_path_challenge(
1318 &mut self,
1319 now: Instant,
1320 buf: &mut Vec<u8>,
1321 ) -> Option<Transmit> {
1322 if let Some(nat_traversal) = &mut self.nat_traversal {
1324 if nat_traversal.should_start_punching(now) {
1325 nat_traversal.start_punching_phase(now);
1326 }
1327 }
1328
1329 let (target_addr, challenge) = {
1331 let nat_traversal = self.nat_traversal.as_ref()?;
1332 match nat_traversal.get_coordination_phase() {
1333 Some(CoordinationPhase::Punching) => {
1334 let targets = nat_traversal.get_punch_targets_from_coordination()?;
1335 if targets.is_empty() {
1336 return None;
1337 }
1338 let target = &targets[0];
1340 (target.remote_addr, target.challenge)
1341 }
1342 _ => return None,
1343 }
1344 };
1345
1346 debug_assert_eq!(
1347 self.highest_space,
1348 SpaceId::Data,
1349 "PATH_CHALLENGE queued without 1-RTT keys"
1350 );
1351
1352 buf.reserve(self.pqc_state.min_initial_size() as usize);
1353 let buf_capacity = buf.capacity();
1354
1355 let mut builder = PacketBuilder::new(
1356 now,
1357 SpaceId::Data,
1358 self.rem_cids.active(),
1359 buf,
1360 buf_capacity,
1361 0,
1362 false,
1363 self,
1364 )?;
1365
1366 trace!(
1367 "sending coordinated PATH_CHALLENGE {:08x} to {}",
1368 challenge, target_addr
1369 );
1370 if !self.encode_or_close(
1371 now,
1372 frame::FrameType::PATH_CHALLENGE.try_encode(buf),
1373 "PATH_CHALLENGE (coordination)",
1374 ) {
1375 return None;
1376 }
1377 buf.write(challenge);
1378 self.stats.frame_tx.path_challenge += 1;
1379
1380 let min_size = self.pqc_state.min_initial_size();
1381 builder.pad_to(min_size);
1382 builder.finish_and_track(now, self, None, buf);
1383
1384 if let Some(nat_traversal) = &mut self.nat_traversal {
1386 nat_traversal.mark_coordination_validating();
1387 }
1388
1389 Some(Transmit {
1390 destination: target_addr,
1391 size: buf.len(),
1392 ecn: if self.path.sending_ecn {
1393 Some(EcnCodepoint::Ect0)
1394 } else {
1395 None
1396 },
1397 segment_size: None,
1398 src_ip: self.local_ip,
1399 })
1400 }
1401
1402 fn send_nat_traversal_challenge(
1404 &mut self,
1405 now: Instant,
1406 buf: &mut Vec<u8>,
1407 ) -> Option<Transmit> {
1408 if let Some(request) = self.send_coordination_request(now, buf) {
1410 return Some(request);
1411 }
1412
1413 if let Some(punch) = self.send_coordinated_path_challenge(now, buf) {
1415 return Some(punch);
1416 }
1417
1418 let (remote_addr, remote_sequence) = {
1420 let nat_traversal = self.nat_traversal.as_ref()?;
1421 let candidates = nat_traversal.get_validation_candidates();
1422 if candidates.is_empty() {
1423 return None;
1424 }
1425 let (sequence, candidate) = candidates[0];
1427 (candidate.address, sequence)
1428 };
1429
1430 let challenge = self.rng.r#gen::<u64>();
1431
1432 if let Err(e) =
1434 self.nat_traversal
1435 .as_mut()?
1436 .start_validation(remote_sequence, challenge, now)
1437 {
1438 warn!("Failed to start NAT traversal validation: {}", e);
1439 return None;
1440 }
1441
1442 debug_assert_eq!(
1443 self.highest_space,
1444 SpaceId::Data,
1445 "PATH_CHALLENGE queued without 1-RTT keys"
1446 );
1447
1448 buf.reserve(self.pqc_state.min_initial_size() as usize);
1449 let buf_capacity = buf.capacity();
1450
1451 let mut builder = PacketBuilder::new(
1453 now,
1454 SpaceId::Data,
1455 self.rem_cids.active(),
1456 buf,
1457 buf_capacity,
1458 0,
1459 false,
1460 self,
1461 )?;
1462
1463 trace!(
1464 "sending PATH_CHALLENGE {:08x} to NAT candidate {}",
1465 challenge, remote_addr
1466 );
1467 if !self.encode_or_close(
1468 now,
1469 frame::FrameType::PATH_CHALLENGE.try_encode(buf),
1470 "PATH_CHALLENGE (nat)",
1471 ) {
1472 return None;
1473 }
1474 buf.write(challenge);
1475 self.stats.frame_tx.path_challenge += 1;
1476
1477 let min_size = self.pqc_state.min_initial_size();
1479 builder.pad_to(min_size);
1480
1481 builder.finish_and_track(now, self, None, buf);
1482
1483 Some(Transmit {
1484 destination: remote_addr,
1485 size: buf.len(),
1486 ecn: if self.path.sending_ecn {
1487 Some(EcnCodepoint::Ect0)
1488 } else {
1489 None
1490 },
1491 segment_size: None,
1492 src_ip: self.local_ip,
1493 })
1494 }
1495
1496 fn send_path_challenge(&mut self, now: Instant, buf: &mut Vec<u8>) -> Option<Transmit> {
1498 let (prev_cid, prev_path) = self.prev_path.as_mut()?;
1499 if !prev_path.challenge_pending {
1500 return None;
1501 }
1502 prev_path.challenge_pending = false;
1503 let token = prev_path
1504 .challenge
1505 .expect("previous path challenge pending without token");
1506 let destination = prev_path.remote;
1507 debug_assert_eq!(
1508 self.highest_space,
1509 SpaceId::Data,
1510 "PATH_CHALLENGE queued without 1-RTT keys"
1511 );
1512 buf.reserve(self.pqc_state.min_initial_size() as usize);
1513
1514 let buf_capacity = buf.capacity();
1515
1516 let mut builder = PacketBuilder::new(
1522 now,
1523 SpaceId::Data,
1524 *prev_cid,
1525 buf,
1526 buf_capacity,
1527 0,
1528 false,
1529 self,
1530 )?;
1531 trace!("validating previous path with PATH_CHALLENGE {:08x}", token);
1532 if !self.encode_or_close(
1533 now,
1534 frame::FrameType::PATH_CHALLENGE.try_encode(buf),
1535 "PATH_CHALLENGE (prev path)",
1536 ) {
1537 return None;
1538 }
1539 buf.write(token);
1540 self.stats.frame_tx.path_challenge += 1;
1541
1542 let min_size = self.pqc_state.min_initial_size();
1547 builder.pad_to(min_size);
1548
1549 builder.finish(self, buf);
1550 self.stats.udp_tx.on_sent(1, buf.len());
1551
1552 Some(Transmit {
1553 destination,
1554 size: buf.len(),
1555 ecn: None,
1556 segment_size: None,
1557 src_ip: self.local_ip,
1558 })
1559 }
1560
1561 fn space_can_send(&self, space_id: SpaceId, frame_space_1rtt: usize) -> SendableFrames {
1563 if self.spaces[space_id].crypto.is_none()
1564 && (space_id != SpaceId::Data
1565 || self.zero_rtt_crypto.is_none()
1566 || self.side.is_server())
1567 {
1568 return SendableFrames::empty();
1570 }
1571 let mut can_send = self.spaces[space_id].can_send(&self.streams);
1572 if space_id == SpaceId::Data {
1573 can_send.other |= self.can_send_1rtt(frame_space_1rtt);
1574 }
1575 can_send
1576 }
1577
1578 pub fn handle_event(&mut self, event: ConnectionEvent) {
1584 use ConnectionEventInner::*;
1585 match event.0 {
1586 Datagram(DatagramConnectionEvent {
1587 now,
1588 remote,
1589 ecn,
1590 first_decode,
1591 remaining,
1592 }) => {
1593 if remote != self.path.remote && !self.side.remote_may_migrate() {
1597 trace!("discarding packet from unrecognized peer {}", remote);
1598 return;
1599 }
1600
1601 let was_anti_amplification_blocked = self.path.anti_amplification_blocked(1);
1602
1603 self.stats.udp_rx.datagrams += 1;
1604 self.stats.udp_rx.bytes += first_decode.len() as u64;
1605 let data_len = first_decode.len();
1606
1607 self.handle_decode(now, remote, ecn, first_decode);
1608 self.path.total_recvd = self.path.total_recvd.saturating_add(data_len as u64);
1613
1614 if let Some(data) = remaining {
1615 self.stats.udp_rx.bytes += data.len() as u64;
1616 self.handle_coalesced(now, remote, ecn, data);
1617 }
1618
1619 #[cfg(feature = "__qlog")]
1620 self.emit_qlog_recovery_metrics(now);
1621
1622 if was_anti_amplification_blocked {
1623 self.set_loss_detection_timer(now);
1627 }
1628 }
1629 NewIdentifiers(ids, now) => {
1630 self.local_cid_state.new_cids(&ids, now);
1631 ids.into_iter().rev().for_each(|frame| {
1632 self.spaces[SpaceId::Data].pending.new_cids.push(frame);
1633 });
1634 if self.timers.get(Timer::PushNewCid).is_none_or(|x| x <= now) {
1636 self.reset_cid_retirement();
1637 }
1638 }
1639 QueueAddAddress(add) => {
1640 self.spaces[SpaceId::Data].pending.add_addresses.push(add);
1642 }
1643 QueuePunchMeNow(punch) => {
1644 self.spaces[SpaceId::Data].pending.punch_me_now.push(punch);
1646 }
1647 }
1648 }
1649
1650 pub fn handle_timeout(&mut self, now: Instant) {
1660 for &timer in &Timer::VALUES {
1661 if !self.timers.is_expired(timer, now) {
1662 continue;
1663 }
1664 self.timers.stop(timer);
1665 trace!(timer = ?timer, "timeout");
1666 match timer {
1667 Timer::Close => {
1668 self.state = State::Drained;
1669 self.endpoint_events.push_back(EndpointEventInner::Drained);
1670 }
1671 Timer::Idle => {
1672 self.kill(ConnectionError::TimedOut);
1673 }
1674 Timer::KeepAlive => {
1675 trace!("sending keep-alive");
1676 self.ping();
1677 }
1678 Timer::LossDetection => {
1679 self.on_loss_detection_timeout(now);
1680
1681 #[cfg(feature = "__qlog")]
1682 self.emit_qlog_recovery_metrics(now);
1683 }
1684 Timer::KeyDiscard => {
1685 self.zero_rtt_crypto = None;
1686 self.prev_crypto = None;
1687 }
1688 Timer::PathValidation => {
1689 debug!("path validation failed");
1690 if let Some((_, prev)) = self.prev_path.take() {
1691 self.path = prev;
1692 }
1693 self.path.challenge = None;
1694 self.path.challenge_pending = false;
1695 }
1696 Timer::Pacing => trace!("pacing timer expired"),
1697 Timer::NatTraversal => {
1698 self.handle_nat_traversal_timeout(now);
1699 }
1700 Timer::PushNewCid => {
1701 let num_new_cid = self.local_cid_state.on_cid_timeout().into();
1703 if !self.state.is_closed() {
1704 trace!(
1705 "push a new cid to peer RETIRE_PRIOR_TO field {}",
1706 self.local_cid_state.retire_prior_to()
1707 );
1708 self.endpoint_events
1709 .push_back(EndpointEventInner::NeedIdentifiers(now, num_new_cid));
1710 }
1711 }
1712 Timer::MaxAckDelay => {
1713 trace!("max ack delay reached");
1714 self.spaces[SpaceId::Data]
1716 .pending_acks
1717 .on_max_ack_delay_timeout()
1718 }
1719 }
1720 }
1721 }
1722
1723 pub fn close(&mut self, now: Instant, error_code: VarInt, reason: Bytes) {
1735 self.close_inner(
1736 now,
1737 Close::Application(frame::ApplicationClose { error_code, reason }),
1738 )
1739 }
1740
1741 fn close_inner(&mut self, now: Instant, reason: Close) {
1742 let was_closed = self.state.is_closed();
1743 if !was_closed {
1744 self.close_common();
1745 self.set_close_timer(now);
1746 self.close = true;
1747 self.state = State::Closed(state::Closed { reason });
1748 }
1749 }
1750
1751 pub fn datagrams(&mut self) -> Datagrams<'_> {
1753 Datagrams { conn: self }
1754 }
1755
1756 pub fn stats(&self) -> ConnectionStats {
1758 let mut stats = self.stats;
1759 stats.path.rtt = self.path.rtt.get();
1760 stats.path.cwnd = self.path.congestion.window();
1761 stats.path.current_mtu = self.path.mtud.current_mtu();
1762
1763 stats
1764 }
1765
1766 pub fn nat_traversal_uses_rfc_frame_format(&self) -> bool {
1768 self.nat_traversal_frame_config.use_rfc_format
1769 }
1770
1771 pub fn nat_traversal_accepts_legacy_frame_format(&self) -> bool {
1773 self.nat_traversal_frame_config.accept_legacy
1774 }
1775
1776 pub fn set_token_binding_peer_id(&mut self, pid: PeerId) {
1778 self.peer_id_for_tokens = Some(pid);
1779 }
1780
1781 pub fn set_delay_new_token_until_binding(&mut self, v: bool) {
1783 self.delay_new_token_until_binding = v;
1784 }
1785
1786 pub fn ping(&mut self) {
1790 self.spaces[self.highest_space].ping_pending = true;
1791 }
1792
1793 pub(crate) fn is_pqc(&self) -> bool {
1795 self.pqc_state.using_pqc
1796 }
1797
1798 pub fn force_key_update(&mut self) {
1802 if !self.state.is_established() {
1803 debug!("ignoring forced key update in illegal state");
1804 return;
1805 }
1806 if self.prev_crypto.is_some() {
1807 debug!("ignoring redundant forced key update");
1810 return;
1811 }
1812 self.update_keys(None, false);
1813 }
1814
1815 pub fn crypto_session(&self) -> &dyn crypto::Session {
1817 &*self.crypto
1818 }
1819
1820 pub fn is_handshaking(&self) -> bool {
1825 self.state.is_handshake()
1826 }
1827
1828 pub fn is_closed(&self) -> bool {
1836 self.state.is_closed()
1837 }
1838
1839 pub fn is_drained(&self) -> bool {
1844 self.state.is_drained()
1845 }
1846
1847 pub fn accepted_0rtt(&self) -> bool {
1851 self.accepted_0rtt
1852 }
1853
1854 pub fn has_0rtt(&self) -> bool {
1856 self.zero_rtt_enabled
1857 }
1858
1859 pub fn has_pending_retransmits(&self) -> bool {
1861 !self.spaces[SpaceId::Data].pending.is_empty(&self.streams)
1862 }
1863
1864 pub fn side(&self) -> Side {
1866 self.side.side()
1867 }
1868
1869 pub fn remote_address(&self) -> SocketAddr {
1871 self.path.remote
1872 }
1873
1874 pub fn local_ip(&self) -> Option<IpAddr> {
1884 self.local_ip
1885 }
1886
1887 pub fn rtt(&self) -> Duration {
1889 self.path.rtt.get()
1890 }
1891
1892 pub fn congestion_state(&self) -> &dyn Controller {
1894 self.path.congestion.as_ref()
1895 }
1896
1897 pub fn path_changed(&mut self, now: Instant) {
1908 self.path.reset(now, &self.config);
1909 }
1910
1911 pub fn set_max_concurrent_streams(&mut self, dir: Dir, count: VarInt) {
1916 self.streams.set_max_concurrent(dir, count);
1917 let pending = &mut self.spaces[SpaceId::Data].pending;
1920 self.streams.queue_max_stream_id(pending);
1921 }
1922
1923 pub fn max_concurrent_streams(&self, dir: Dir) -> u64 {
1929 self.streams.max_concurrent(dir)
1930 }
1931
1932 pub fn set_receive_window(&mut self, receive_window: VarInt) {
1934 if self.streams.set_receive_window(receive_window) {
1935 self.spaces[SpaceId::Data].pending.max_data = true;
1936 }
1937 }
1938
1939 pub fn set_address_discovery_enabled(&mut self, enabled: bool) {
1941 if let Some(ref mut state) = self.address_discovery_state {
1942 state.enabled = enabled;
1943 }
1944 }
1945
1946 pub fn address_discovery_enabled(&self) -> bool {
1948 self.address_discovery_state
1949 .as_ref()
1950 .is_some_and(|state| state.enabled)
1951 }
1952
1953 pub fn observed_address(&self) -> Option<SocketAddr> {
1958 self.address_discovery_state
1959 .as_ref()
1960 .and_then(|state| state.get_observed_address(0)) }
1962
1963 pub fn all_observed_addresses(&self) -> Vec<SocketAddr> {
1968 self.address_discovery_state
1969 .as_ref()
1970 .map(|state| state.get_all_received_history())
1971 .unwrap_or_default()
1972 }
1973
1974 #[allow(dead_code)]
1976 pub(crate) fn address_discovery_state(&self) -> Option<&AddressDiscoveryState> {
1977 self.address_discovery_state.as_ref()
1978 }
1979
1980 fn on_ack_received(
1981 &mut self,
1982 now: Instant,
1983 space: SpaceId,
1984 ack: frame::Ack,
1985 ) -> Result<(), TransportError> {
1986 if ack.largest >= self.spaces[space].next_packet_number {
1987 return Err(TransportError::PROTOCOL_VIOLATION("unsent packet acked"));
1988 }
1989 let new_largest = {
1990 let space = &mut self.spaces[space];
1991 if space.largest_acked_packet.is_none_or(|pn| ack.largest > pn) {
1992 space.largest_acked_packet = Some(ack.largest);
1993 if let Some(info) = space.sent_packets.get(&ack.largest) {
1994 space.largest_acked_packet_sent = info.time_sent;
1998 }
1999 true
2000 } else {
2001 false
2002 }
2003 };
2004
2005 let mut newly_acked = ArrayRangeSet::new();
2007 for range in ack.iter() {
2008 self.packet_number_filter.check_ack(space, range.clone())?;
2009 for (&pn, _) in self.spaces[space].sent_packets.range(range) {
2010 newly_acked.insert_one(pn);
2011 }
2012 }
2013
2014 if newly_acked.is_empty() {
2015 return Ok(());
2016 }
2017
2018 let mut ack_eliciting_acked = false;
2019 for packet in newly_acked.elts() {
2020 if let Some(info) = self.spaces[space].take(packet) {
2021 if let Some(acked) = info.largest_acked {
2022 self.spaces[space].pending_acks.subtract_below(acked);
2028 }
2029 ack_eliciting_acked |= info.ack_eliciting;
2030
2031 let mtu_updated = self.path.mtud.on_acked(space, packet, info.size);
2033 if mtu_updated {
2034 self.path
2035 .congestion
2036 .on_mtu_update(self.path.mtud.current_mtu());
2037 }
2038
2039 self.ack_frequency.on_acked(packet);
2041
2042 self.on_packet_acked(now, packet, info);
2043 }
2044 }
2045
2046 self.path.congestion.on_end_acks(
2047 now,
2048 self.path.in_flight.bytes,
2049 self.app_limited,
2050 self.spaces[space].largest_acked_packet,
2051 );
2052
2053 if new_largest && ack_eliciting_acked {
2054 let ack_delay = if space != SpaceId::Data {
2055 Duration::from_micros(0)
2056 } else {
2057 cmp::min(
2058 self.ack_frequency.peer_max_ack_delay,
2059 Duration::from_micros(ack.delay << self.peer_params.ack_delay_exponent.0),
2060 )
2061 };
2062 let rtt = instant_saturating_sub(now, self.spaces[space].largest_acked_packet_sent);
2063 self.path.rtt.update(ack_delay, rtt);
2064 if self.path.first_packet_after_rtt_sample.is_none() {
2065 self.path.first_packet_after_rtt_sample =
2066 Some((space, self.spaces[space].next_packet_number));
2067 }
2068 }
2069
2070 self.detect_lost_packets(now, space, true);
2072
2073 if self.peer_completed_address_validation() {
2074 self.pto_count = 0;
2075 }
2076
2077 if self.path.sending_ecn {
2079 if let Some(ecn) = ack.ecn {
2080 if new_largest {
2085 let sent = self.spaces[space].largest_acked_packet_sent;
2086 self.process_ecn(now, space, newly_acked.len() as u64, ecn, sent);
2087 }
2088 } else {
2089 debug!("ECN not acknowledged by peer");
2091 self.path.sending_ecn = false;
2092 }
2093 }
2094
2095 self.set_loss_detection_timer(now);
2096 Ok(())
2097 }
2098
2099 fn process_ecn(
2101 &mut self,
2102 now: Instant,
2103 space: SpaceId,
2104 newly_acked: u64,
2105 ecn: frame::EcnCounts,
2106 largest_sent_time: Instant,
2107 ) {
2108 match self.spaces[space].detect_ecn(newly_acked, ecn) {
2109 Err(e) => {
2110 debug!("halting ECN due to verification failure: {}", e);
2111 self.path.sending_ecn = false;
2112 self.spaces[space].ecn_feedback = frame::EcnCounts::ZERO;
2115 }
2116 Ok(false) => {}
2117 Ok(true) => {
2118 self.stats.path.congestion_events += 1;
2119 self.path
2120 .congestion
2121 .on_congestion_event(now, largest_sent_time, false, 0);
2122 }
2123 }
2124 }
2125
2126 fn on_packet_acked(&mut self, now: Instant, pn: u64, info: SentPacket) {
2129 self.remove_in_flight(pn, &info);
2130 if info.ack_eliciting && self.path.challenge.is_none() {
2131 self.path.congestion.on_ack(
2134 now,
2135 info.time_sent,
2136 info.size.into(),
2137 self.app_limited,
2138 &self.path.rtt,
2139 );
2140 }
2141
2142 if let Some(retransmits) = info.retransmits.get() {
2144 for (id, _) in retransmits.reset_stream.iter() {
2145 self.streams.reset_acked(*id);
2146 }
2147 }
2148
2149 for frame in info.stream_frames {
2150 self.streams.received_ack_of(frame);
2151 }
2152 }
2153
2154 fn set_key_discard_timer(&mut self, now: Instant, space: SpaceId) {
2155 let start = if self.zero_rtt_crypto.is_some() {
2156 now
2157 } else {
2158 self.prev_crypto
2159 .as_ref()
2160 .expect("no previous keys")
2161 .end_packet
2162 .as_ref()
2163 .expect("update not acknowledged yet")
2164 .1
2165 };
2166 self.timers
2167 .set(Timer::KeyDiscard, start + self.pto(space) * 3);
2168 }
2169
2170 fn on_loss_detection_timeout(&mut self, now: Instant) {
2171 if let Some((_, pn_space)) = self.loss_time_and_space() {
2172 self.detect_lost_packets(now, pn_space, false);
2174 self.set_loss_detection_timer(now);
2175 return;
2176 }
2177
2178 let (_, space) = match self.pto_time_and_space(now) {
2179 Some(x) => x,
2180 None => {
2181 error!("PTO expired while unset");
2182 return;
2183 }
2184 };
2185 trace!(
2186 in_flight = self.path.in_flight.bytes,
2187 count = self.pto_count,
2188 ?space,
2189 "PTO fired"
2190 );
2191
2192 let count = match self.path.in_flight.ack_eliciting {
2193 0 => {
2196 debug_assert!(!self.peer_completed_address_validation());
2197 1
2198 }
2199 _ => 2,
2201 };
2202 self.spaces[space].loss_probes = self.spaces[space].loss_probes.saturating_add(count);
2203 self.pto_count = self.pto_count.saturating_add(1);
2204 self.set_loss_detection_timer(now);
2205 }
2206
2207 fn detect_lost_packets(&mut self, now: Instant, pn_space: SpaceId, due_to_ack: bool) {
2208 let mut lost_packets = Vec::<u64>::new();
2209 let mut lost_mtu_probe = None;
2210 let in_flight_mtu_probe = self.path.mtud.in_flight_mtu_probe();
2211 let rtt = self.path.rtt.conservative();
2212 let loss_delay = cmp::max(rtt.mul_f32(self.config.time_threshold), TIMER_GRANULARITY);
2213
2214 let largest_acked_packet = self.spaces[pn_space].largest_acked_packet.unwrap();
2215 let packet_threshold = self.config.packet_threshold as u64;
2216 let mut size_of_lost_packets = 0u64;
2217
2218 let congestion_period =
2222 self.pto(SpaceId::Data) * self.config.persistent_congestion_threshold;
2223 let mut persistent_congestion_start: Option<Instant> = None;
2224 let mut prev_packet = None;
2225 let mut in_persistent_congestion = false;
2226
2227 let space = &mut self.spaces[pn_space];
2228 space.loss_time = None;
2229
2230 for (&packet, info) in space.sent_packets.range(0..largest_acked_packet) {
2231 if prev_packet != Some(packet.wrapping_sub(1)) {
2232 persistent_congestion_start = None;
2234 }
2235
2236 if now.saturating_duration_since(info.time_sent) >= loss_delay
2240 || largest_acked_packet >= packet + packet_threshold
2241 {
2242 if Some(packet) == in_flight_mtu_probe {
2243 lost_mtu_probe = in_flight_mtu_probe;
2246 } else {
2247 lost_packets.push(packet);
2248 size_of_lost_packets += info.size as u64;
2249 if info.ack_eliciting && due_to_ack {
2250 match persistent_congestion_start {
2251 Some(start) if info.time_sent - start > congestion_period => {
2254 in_persistent_congestion = true;
2255 }
2256 None if self
2258 .path
2259 .first_packet_after_rtt_sample
2260 .is_some_and(|x| x < (pn_space, packet)) =>
2261 {
2262 persistent_congestion_start = Some(info.time_sent);
2263 }
2264 _ => {}
2265 }
2266 }
2267 }
2268 } else {
2269 let next_loss_time = info.time_sent + loss_delay;
2270 space.loss_time = Some(
2271 space
2272 .loss_time
2273 .map_or(next_loss_time, |x| cmp::min(x, next_loss_time)),
2274 );
2275 persistent_congestion_start = None;
2276 }
2277
2278 prev_packet = Some(packet);
2279 }
2280
2281 if let Some(largest_lost) = lost_packets.last().cloned() {
2283 let old_bytes_in_flight = self.path.in_flight.bytes;
2284 let largest_lost_sent = self.spaces[pn_space].sent_packets[&largest_lost].time_sent;
2285 self.lost_packets += lost_packets.len() as u64;
2286 self.stats.path.lost_packets += lost_packets.len() as u64;
2287 self.stats.path.lost_bytes += size_of_lost_packets;
2288 trace!(
2289 "packets lost: {:?}, bytes lost: {}",
2290 lost_packets, size_of_lost_packets
2291 );
2292
2293 for &packet in &lost_packets {
2294 let info = self.spaces[pn_space].take(packet).unwrap(); self.remove_in_flight(packet, &info);
2296 for frame in info.stream_frames {
2297 self.streams.retransmit(frame);
2298 }
2299 self.spaces[pn_space].pending |= info.retransmits;
2300 self.path.mtud.on_non_probe_lost(packet, info.size);
2301 }
2302
2303 if self.path.mtud.black_hole_detected(now) {
2304 self.stats.path.black_holes_detected += 1;
2305 self.path
2306 .congestion
2307 .on_mtu_update(self.path.mtud.current_mtu());
2308 if let Some(max_datagram_size) = self.datagrams().max_size() {
2309 self.datagrams.drop_oversized(max_datagram_size);
2310 }
2311 }
2312
2313 let lost_ack_eliciting = old_bytes_in_flight != self.path.in_flight.bytes;
2315
2316 if lost_ack_eliciting {
2317 self.stats.path.congestion_events += 1;
2318 self.path.congestion.on_congestion_event(
2319 now,
2320 largest_lost_sent,
2321 in_persistent_congestion,
2322 size_of_lost_packets,
2323 );
2324 }
2325 }
2326
2327 if let Some(packet) = lost_mtu_probe {
2329 let info = self.spaces[SpaceId::Data].take(packet).unwrap(); self.remove_in_flight(packet, &info);
2331 self.path.mtud.on_probe_lost();
2332 self.stats.path.lost_plpmtud_probes += 1;
2333 }
2334 }
2335
2336 fn loss_time_and_space(&self) -> Option<(Instant, SpaceId)> {
2337 SpaceId::iter()
2338 .filter_map(|id| Some((self.spaces[id].loss_time?, id)))
2339 .min_by_key(|&(time, _)| time)
2340 }
2341
2342 fn pto_time_and_space(&self, now: Instant) -> Option<(Instant, SpaceId)> {
2343 let backoff = 2u32.pow(self.pto_count.min(MAX_BACKOFF_EXPONENT));
2344 let mut duration = self.path.rtt.pto_base() * backoff;
2345
2346 if self.path.in_flight.ack_eliciting == 0 {
2347 debug_assert!(!self.peer_completed_address_validation());
2348 let space = match self.highest_space {
2349 SpaceId::Handshake => SpaceId::Handshake,
2350 _ => SpaceId::Initial,
2351 };
2352 return Some((now + duration, space));
2353 }
2354
2355 let mut result = None;
2356 for space in SpaceId::iter() {
2357 if self.spaces[space].in_flight == 0 {
2358 continue;
2359 }
2360 if space == SpaceId::Data {
2361 if self.is_handshaking() {
2363 return result;
2364 }
2365 duration += self.ack_frequency.max_ack_delay_for_pto() * backoff;
2367 }
2368 let last_ack_eliciting = match self.spaces[space].time_of_last_ack_eliciting_packet {
2369 Some(time) => time,
2370 None => continue,
2371 };
2372 let pto = last_ack_eliciting + duration;
2373 if result.is_none_or(|(earliest_pto, _)| pto < earliest_pto) {
2374 result = Some((pto, space));
2375 }
2376 }
2377 result
2378 }
2379
2380 fn peer_completed_address_validation(&self) -> bool {
2381 if self.side.is_server() || self.state.is_closed() {
2382 return true;
2383 }
2384 self.spaces[SpaceId::Handshake]
2387 .largest_acked_packet
2388 .is_some()
2389 || self.spaces[SpaceId::Data].largest_acked_packet.is_some()
2390 || (self.spaces[SpaceId::Data].crypto.is_some()
2391 && self.spaces[SpaceId::Handshake].crypto.is_none())
2392 }
2393
2394 fn set_loss_detection_timer(&mut self, now: Instant) {
2395 if self.state.is_closed() {
2396 return;
2400 }
2401
2402 if let Some((loss_time, _)) = self.loss_time_and_space() {
2403 self.timers.set(Timer::LossDetection, loss_time);
2405 return;
2406 }
2407
2408 if self.path.anti_amplification_blocked(1) {
2409 self.timers.stop(Timer::LossDetection);
2411 return;
2412 }
2413
2414 if self.path.in_flight.ack_eliciting == 0 && self.peer_completed_address_validation() {
2415 self.timers.stop(Timer::LossDetection);
2418 return;
2419 }
2420
2421 if let Some((timeout, _)) = self.pto_time_and_space(now) {
2424 self.timers.set(Timer::LossDetection, timeout);
2425 } else {
2426 self.timers.stop(Timer::LossDetection);
2427 }
2428 }
2429
2430 fn pto(&self, space: SpaceId) -> Duration {
2432 let max_ack_delay = match space {
2433 SpaceId::Initial | SpaceId::Handshake => Duration::ZERO,
2434 SpaceId::Data => self.ack_frequency.max_ack_delay_for_pto(),
2435 };
2436 self.path.rtt.pto_base() + max_ack_delay
2437 }
2438
2439 fn on_packet_authenticated(
2440 &mut self,
2441 now: Instant,
2442 space_id: SpaceId,
2443 ecn: Option<EcnCodepoint>,
2444 packet: Option<u64>,
2445 spin: bool,
2446 is_1rtt: bool,
2447 ) {
2448 self.total_authed_packets += 1;
2449 self.reset_keep_alive(now);
2450 self.reset_idle_timeout(now, space_id);
2451 self.permit_idle_reset = true;
2452 self.receiving_ecn |= ecn.is_some();
2453 if let Some(x) = ecn {
2454 let space = &mut self.spaces[space_id];
2455 space.ecn_counters += x;
2456
2457 if x.is_ce() {
2458 space.pending_acks.set_immediate_ack_required();
2459 }
2460 }
2461
2462 let packet = match packet {
2463 Some(x) => x,
2464 None => return,
2465 };
2466 if self.side.is_server() {
2467 if self.spaces[SpaceId::Initial].crypto.is_some() && space_id == SpaceId::Handshake {
2468 self.discard_space(now, SpaceId::Initial);
2470 }
2471 if self.zero_rtt_crypto.is_some() && is_1rtt {
2472 self.set_key_discard_timer(now, space_id)
2474 }
2475 }
2476 let space = &mut self.spaces[space_id];
2477 space.pending_acks.insert_one(packet, now);
2478 if packet >= space.rx_packet {
2479 space.rx_packet = packet;
2480 self.spin = self.side.is_client() ^ spin;
2482 }
2483 }
2484
2485 fn reset_idle_timeout(&mut self, now: Instant, space: SpaceId) {
2486 let timeout = match self.idle_timeout {
2487 None => return,
2488 Some(dur) => dur,
2489 };
2490 if self.state.is_closed() {
2491 self.timers.stop(Timer::Idle);
2492 return;
2493 }
2494 let dt = cmp::max(timeout, 3 * self.pto(space));
2495 self.timers.set(Timer::Idle, now + dt);
2496 }
2497
2498 fn reset_keep_alive(&mut self, now: Instant) {
2499 let interval = match self.config.keep_alive_interval {
2500 Some(x) if self.state.is_established() => x,
2501 _ => return,
2502 };
2503 self.timers.set(Timer::KeepAlive, now + interval);
2504 }
2505
2506 fn reset_cid_retirement(&mut self) {
2507 if let Some(t) = self.local_cid_state.next_timeout() {
2508 self.timers.set(Timer::PushNewCid, t);
2509 }
2510 }
2511
2512 pub(crate) fn handle_first_packet(
2517 &mut self,
2518 now: Instant,
2519 remote: SocketAddr,
2520 ecn: Option<EcnCodepoint>,
2521 packet_number: u64,
2522 packet: InitialPacket,
2523 remaining: Option<BytesMut>,
2524 ) -> Result<(), ConnectionError> {
2525 let span = trace_span!("first recv");
2526 let _guard = span.enter();
2527 debug_assert!(self.side.is_server());
2528 let len = packet.header_data.len() + packet.payload.len();
2529 self.path.total_recvd = len as u64;
2530
2531 match self.state {
2532 State::Handshake(ref mut state) => {
2533 state.expected_token = packet.header.token.clone();
2534 }
2535 _ => unreachable!("first packet must be delivered in Handshake state"),
2536 }
2537
2538 self.on_packet_authenticated(
2539 now,
2540 SpaceId::Initial,
2541 ecn,
2542 Some(packet_number),
2543 false,
2544 false,
2545 );
2546
2547 self.process_decrypted_packet(now, remote, Some(packet_number), packet.into())?;
2548 if let Some(data) = remaining {
2549 self.handle_coalesced(now, remote, ecn, data);
2550 }
2551
2552 #[cfg(feature = "__qlog")]
2553 self.emit_qlog_recovery_metrics(now);
2554
2555 Ok(())
2556 }
2557
2558 fn init_0rtt(&mut self) {
2559 let (header, packet) = match self.crypto.early_crypto() {
2560 Some(x) => x,
2561 None => return,
2562 };
2563 if self.side.is_client() {
2564 match self.crypto.transport_parameters() {
2565 Ok(params) => {
2566 let params = params
2567 .expect("crypto layer didn't supply transport parameters with ticket");
2568 let params = TransportParameters {
2570 initial_src_cid: None,
2571 original_dst_cid: None,
2572 preferred_address: None,
2573 retry_src_cid: None,
2574 stateless_reset_token: None,
2575 min_ack_delay: None,
2576 ack_delay_exponent: TransportParameters::default().ack_delay_exponent,
2577 max_ack_delay: TransportParameters::default().max_ack_delay,
2578 ..params
2579 };
2580 self.set_peer_params(params);
2581 }
2582 Err(e) => {
2583 error!("session ticket has malformed transport parameters: {}", e);
2584 return;
2585 }
2586 }
2587 }
2588 trace!("0-RTT enabled");
2589 self.zero_rtt_enabled = true;
2590 self.zero_rtt_crypto = Some(ZeroRttCrypto { header, packet });
2591 }
2592
2593 fn read_crypto(
2594 &mut self,
2595 space: SpaceId,
2596 crypto: &frame::Crypto,
2597 payload_len: usize,
2598 ) -> Result<(), TransportError> {
2599 let expected = if !self.state.is_handshake() {
2600 SpaceId::Data
2601 } else if self.highest_space == SpaceId::Initial {
2602 SpaceId::Initial
2603 } else {
2604 SpaceId::Handshake
2607 };
2608 debug_assert!(space <= expected, "received out-of-order CRYPTO data");
2612
2613 let end = crypto.offset + crypto.data.len() as u64;
2614 if space < expected && end > self.spaces[space].crypto_stream.bytes_read() {
2615 warn!(
2616 "received new {:?} CRYPTO data when expecting {:?}",
2617 space, expected
2618 );
2619 return Err(TransportError::PROTOCOL_VIOLATION(
2620 "new data at unexpected encryption level",
2621 ));
2622 }
2623
2624 self.pqc_state.detect_pqc_from_crypto(&crypto.data, space);
2626
2627 if self.pqc_state.should_trigger_mtu_discovery() {
2629 self.path
2631 .mtud
2632 .reset(self.pqc_state.min_initial_size(), self.config.min_mtu);
2633 trace!("Triggered MTU discovery for PQC handshake");
2634 }
2635
2636 let space = &mut self.spaces[space];
2637 let max = end.saturating_sub(space.crypto_stream.bytes_read());
2638 if max > self.config.crypto_buffer_size as u64 {
2639 return Err(TransportError::CRYPTO_BUFFER_EXCEEDED(""));
2640 }
2641
2642 space
2643 .crypto_stream
2644 .insert(crypto.offset, crypto.data.clone(), payload_len)
2645 .map_err(|_| TransportError::INTERNAL_ERROR("too many gaps in crypto stream buffer"))?;
2646
2647 while let Some(chunk) = space.crypto_stream.read(usize::MAX, true) {
2648 trace!("consumed {} CRYPTO bytes", chunk.bytes.len());
2649 if self.crypto.read_handshake(&chunk.bytes)? {
2650 self.events.push_back(Event::HandshakeDataReady);
2651 }
2652 }
2653
2654 Ok(())
2655 }
2656
2657 fn write_crypto(&mut self) {
2658 loop {
2659 let space = self.highest_space;
2660 let mut outgoing = Vec::new();
2661 if let Some(crypto) = self.crypto.write_handshake(&mut outgoing) {
2662 match space {
2663 SpaceId::Initial => {
2664 self.upgrade_crypto(SpaceId::Handshake, crypto);
2665 }
2666 SpaceId::Handshake => {
2667 self.upgrade_crypto(SpaceId::Data, crypto);
2668 }
2669 _ => unreachable!("got updated secrets during 1-RTT"),
2670 }
2671 }
2672 if outgoing.is_empty() {
2673 if space == self.highest_space {
2674 break;
2675 } else {
2676 continue;
2678 }
2679 }
2680 let offset = self.spaces[space].crypto_offset;
2681 let outgoing = Bytes::from(outgoing);
2682 if let State::Handshake(ref mut state) = self.state {
2683 if space == SpaceId::Initial && offset == 0 && self.side.is_client() {
2684 state.client_hello = Some(outgoing.clone());
2685 }
2686 }
2687 self.spaces[space].crypto_offset += outgoing.len() as u64;
2688 trace!("wrote {} {:?} CRYPTO bytes", outgoing.len(), space);
2689
2690 let use_pqc_fragmentation = self.pqc_state.using_pqc && outgoing.len() > 1200;
2692
2693 if use_pqc_fragmentation {
2694 let frames = self.pqc_state.packet_handler.fragment_crypto_data(
2696 &outgoing,
2697 offset,
2698 self.pqc_state.min_initial_size() as usize,
2699 );
2700 for frame in frames {
2701 self.spaces[space].pending.crypto.push_back(frame);
2702 }
2703 } else {
2704 self.spaces[space].pending.crypto.push_back(frame::Crypto {
2706 offset,
2707 data: outgoing,
2708 });
2709 }
2710 }
2711 }
2712
2713 fn upgrade_crypto(&mut self, space: SpaceId, crypto: Keys) {
2715 debug_assert!(
2716 self.spaces[space].crypto.is_none(),
2717 "already reached packet space {space:?}"
2718 );
2719 trace!("{:?} keys ready", space);
2720 if space == SpaceId::Data {
2721 self.next_crypto = Some(
2723 self.crypto
2724 .next_1rtt_keys()
2725 .expect("handshake should be complete"),
2726 );
2727 }
2728
2729 self.spaces[space].crypto = Some(crypto);
2730 debug_assert!(space as usize > self.highest_space as usize);
2731 self.highest_space = space;
2732 if space == SpaceId::Data && self.side.is_client() {
2733 self.zero_rtt_crypto = None;
2735 }
2736 }
2737
2738 fn discard_space(&mut self, now: Instant, space_id: SpaceId) {
2739 debug_assert!(space_id != SpaceId::Data);
2740 trace!("discarding {:?} keys", space_id);
2741 if space_id == SpaceId::Initial {
2742 if let ConnectionSide::Client { token, .. } = &mut self.side {
2744 *token = Bytes::new();
2745 }
2746 }
2747 let space = &mut self.spaces[space_id];
2748 space.crypto = None;
2749 space.time_of_last_ack_eliciting_packet = None;
2750 space.loss_time = None;
2751 space.in_flight = 0;
2752 let sent_packets = mem::take(&mut space.sent_packets);
2753 for (pn, packet) in sent_packets.into_iter() {
2754 self.remove_in_flight(pn, &packet);
2755 }
2756 self.set_loss_detection_timer(now)
2757 }
2758
2759 fn handle_coalesced(
2760 &mut self,
2761 now: Instant,
2762 remote: SocketAddr,
2763 ecn: Option<EcnCodepoint>,
2764 data: BytesMut,
2765 ) {
2766 self.path.total_recvd = self.path.total_recvd.saturating_add(data.len() as u64);
2767 let mut remaining = Some(data);
2768 while let Some(data) = remaining {
2769 match PartialDecode::new(
2770 data,
2771 &FixedLengthConnectionIdParser::new(self.local_cid_state.cid_len()),
2772 &[self.version],
2773 self.endpoint_config.grease_quic_bit,
2774 ) {
2775 Ok((partial_decode, rest)) => {
2776 remaining = rest;
2777 self.handle_decode(now, remote, ecn, partial_decode);
2778 }
2779 Err(e) => {
2780 trace!("malformed header: {}", e);
2781 return;
2782 }
2783 }
2784 }
2785 }
2786
2787 fn handle_decode(
2788 &mut self,
2789 now: Instant,
2790 remote: SocketAddr,
2791 ecn: Option<EcnCodepoint>,
2792 partial_decode: PartialDecode,
2793 ) {
2794 if let Some(decoded) = packet_crypto::unprotect_header(
2795 partial_decode,
2796 &self.spaces,
2797 self.zero_rtt_crypto.as_ref(),
2798 self.peer_params.stateless_reset_token,
2799 ) {
2800 self.handle_packet(now, remote, ecn, decoded.packet, decoded.stateless_reset);
2801 }
2802 }
2803
2804 fn handle_packet(
2805 &mut self,
2806 now: Instant,
2807 remote: SocketAddr,
2808 ecn: Option<EcnCodepoint>,
2809 packet: Option<Packet>,
2810 stateless_reset: bool,
2811 ) {
2812 self.stats.udp_rx.ios += 1;
2813 if let Some(ref packet) = packet {
2814 trace!(
2815 "got {:?} packet ({} bytes) from {} using id {}",
2816 packet.header.space(),
2817 packet.payload.len() + packet.header_data.len(),
2818 remote,
2819 packet.header.dst_cid(),
2820 );
2821
2822 #[cfg(feature = "trace")]
2824 {
2825 use crate::trace_packet_received;
2826 let packet_size = packet.payload.len() + packet.header_data.len();
2828 trace_packet_received!(
2829 &self.event_log,
2830 self.trace_context.trace_id(),
2831 packet_size as u32,
2832 0 );
2834 }
2835 }
2836
2837 if self.is_handshaking() && remote != self.path.remote {
2838 debug!("discarding packet with unexpected remote during handshake");
2839 return;
2840 }
2841
2842 let was_closed = self.state.is_closed();
2843 let was_drained = self.state.is_drained();
2844
2845 let decrypted = match packet {
2846 None => Err(None),
2847 Some(mut packet) => self
2848 .decrypt_packet(now, &mut packet)
2849 .map(move |number| (packet, number)),
2850 };
2851 let result = match decrypted {
2852 _ if stateless_reset => {
2853 debug!("got stateless reset");
2854 Err(ConnectionError::Reset)
2855 }
2856 Err(Some(e)) => {
2857 warn!("illegal packet: {}", e);
2858 Err(e.into())
2859 }
2860 Err(None) => {
2861 debug!("failed to authenticate packet");
2862 self.authentication_failures += 1;
2863 let integrity_limit = self.spaces[self.highest_space]
2864 .crypto
2865 .as_ref()
2866 .unwrap()
2867 .packet
2868 .local
2869 .integrity_limit();
2870 if self.authentication_failures > integrity_limit {
2871 Err(TransportError::AEAD_LIMIT_REACHED("integrity limit violated").into())
2872 } else {
2873 return;
2874 }
2875 }
2876 Ok((packet, number)) => {
2877 let span = match number {
2878 Some(pn) => trace_span!("recv", space = ?packet.header.space(), pn),
2879 None => trace_span!("recv", space = ?packet.header.space()),
2880 };
2881 let _guard = span.enter();
2882
2883 let is_duplicate = |n| self.spaces[packet.header.space()].dedup.insert(n);
2884 if number.is_some_and(is_duplicate) {
2885 debug!("discarding possible duplicate packet");
2886 return;
2887 } else if self.state.is_handshake() && packet.header.is_short() {
2888 trace!("dropping short packet during handshake");
2890 return;
2891 } else {
2892 if let Header::Initial(InitialHeader { ref token, .. }) = packet.header {
2893 if let State::Handshake(ref hs) = self.state {
2894 if self.side.is_server() && token != &hs.expected_token {
2895 warn!("discarding Initial with invalid retry token");
2899 return;
2900 }
2901 }
2902 }
2903
2904 if !self.state.is_closed() {
2905 let spin = match packet.header {
2906 Header::Short { spin, .. } => spin,
2907 _ => false,
2908 };
2909 self.on_packet_authenticated(
2910 now,
2911 packet.header.space(),
2912 ecn,
2913 number,
2914 spin,
2915 packet.header.is_1rtt(),
2916 );
2917 }
2918
2919 self.process_decrypted_packet(now, remote, number, packet)
2920 }
2921 }
2922 };
2923
2924 if let Err(conn_err) = result {
2926 self.error = Some(conn_err.clone());
2927 self.state = match conn_err {
2928 ConnectionError::ApplicationClosed(reason) => State::closed(reason),
2929 ConnectionError::ConnectionClosed(reason) => State::closed(reason),
2930 ConnectionError::Reset
2931 | ConnectionError::TransportError(TransportError {
2932 code: TransportErrorCode::AEAD_LIMIT_REACHED,
2933 ..
2934 }) => State::Drained,
2935 ConnectionError::TimedOut => {
2936 unreachable!("timeouts aren't generated by packet processing");
2937 }
2938 ConnectionError::TransportError(err) => {
2939 debug!("closing connection due to transport error: {}", err);
2940 State::closed(err)
2941 }
2942 ConnectionError::VersionMismatch => State::Draining,
2943 ConnectionError::LocallyClosed => {
2944 unreachable!("LocallyClosed isn't generated by packet processing");
2945 }
2946 ConnectionError::CidsExhausted => {
2947 unreachable!("CidsExhausted isn't generated by packet processing");
2948 }
2949 };
2950 }
2951
2952 if !was_closed && self.state.is_closed() {
2953 self.close_common();
2954 if !self.state.is_drained() {
2955 self.set_close_timer(now);
2956 }
2957 }
2958 if !was_drained && self.state.is_drained() {
2959 self.endpoint_events.push_back(EndpointEventInner::Drained);
2960 self.timers.stop(Timer::Close);
2963 }
2964
2965 if let State::Closed(_) = self.state {
2967 self.close = remote == self.path.remote;
2968 }
2969 }
2970
2971 fn process_decrypted_packet(
2972 &mut self,
2973 now: Instant,
2974 remote: SocketAddr,
2975 number: Option<u64>,
2976 packet: Packet,
2977 ) -> Result<(), ConnectionError> {
2978 let state = match self.state {
2979 State::Established => {
2980 match packet.header.space() {
2981 SpaceId::Data => self.process_payload(now, remote, number.unwrap(), packet)?,
2982 _ if packet.header.has_frames() => self.process_early_payload(now, packet)?,
2983 _ => {
2984 trace!("discarding unexpected pre-handshake packet");
2985 }
2986 }
2987 return Ok(());
2988 }
2989 State::Closed(_) => {
2990 for result in frame::Iter::new(packet.payload.freeze())? {
2991 let frame = match result {
2992 Ok(frame) => frame,
2993 Err(err) => {
2994 debug!("frame decoding error: {err:?}");
2995 continue;
2996 }
2997 };
2998
2999 if let Frame::Padding = frame {
3000 continue;
3001 };
3002
3003 self.stats.frame_rx.record(&frame);
3004
3005 if let Frame::Close(_) = frame {
3006 trace!("draining");
3007 self.state = State::Draining;
3008 break;
3009 }
3010 }
3011 return Ok(());
3012 }
3013 State::Draining | State::Drained => return Ok(()),
3014 State::Handshake(ref mut state) => state,
3015 };
3016
3017 match packet.header {
3018 Header::Retry {
3019 src_cid: rem_cid, ..
3020 } => {
3021 if self.side.is_server() {
3022 return Err(TransportError::PROTOCOL_VIOLATION("client sent Retry").into());
3023 }
3024
3025 if self.total_authed_packets > 1
3026 || packet.payload.len() <= 16 || !self.crypto.is_valid_retry(
3028 &self.rem_cids.active(),
3029 &packet.header_data,
3030 &packet.payload,
3031 )
3032 {
3033 trace!("discarding invalid Retry");
3034 return Ok(());
3042 }
3043
3044 trace!("retrying with CID {}", rem_cid);
3045 let client_hello = state.client_hello.take().unwrap();
3046 self.retry_src_cid = Some(rem_cid);
3047 self.rem_cids.update_initial_cid(rem_cid);
3048 self.rem_handshake_cid = rem_cid;
3049
3050 let space = &mut self.spaces[SpaceId::Initial];
3051 if let Some(info) = space.take(0) {
3052 self.on_packet_acked(now, 0, info);
3053 };
3054
3055 self.discard_space(now, SpaceId::Initial); self.spaces[SpaceId::Initial] = PacketSpace {
3057 crypto: Some(self.crypto.initial_keys(&rem_cid, self.side.side())),
3058 next_packet_number: self.spaces[SpaceId::Initial].next_packet_number,
3059 crypto_offset: client_hello.len() as u64,
3060 ..PacketSpace::new(now)
3061 };
3062 self.spaces[SpaceId::Initial]
3063 .pending
3064 .crypto
3065 .push_back(frame::Crypto {
3066 offset: 0,
3067 data: client_hello,
3068 });
3069
3070 let zero_rtt = mem::take(&mut self.spaces[SpaceId::Data].sent_packets);
3072 for (pn, info) in zero_rtt {
3073 self.remove_in_flight(pn, &info);
3074 self.spaces[SpaceId::Data].pending |= info.retransmits;
3075 }
3076 self.streams.retransmit_all_for_0rtt();
3077
3078 let token_len = packet.payload.len() - 16;
3079 let ConnectionSide::Client { ref mut token, .. } = self.side else {
3080 unreachable!("we already short-circuited if we're server");
3081 };
3082 *token = packet.payload.freeze().split_to(token_len);
3083 self.state = State::Handshake(state::Handshake {
3084 expected_token: Bytes::new(),
3085 rem_cid_set: false,
3086 client_hello: None,
3087 });
3088 Ok(())
3089 }
3090 Header::Long {
3091 ty: LongType::Handshake,
3092 src_cid: rem_cid,
3093 ..
3094 } => {
3095 if rem_cid != self.rem_handshake_cid {
3096 debug!(
3097 "discarding packet with mismatched remote CID: {} != {}",
3098 self.rem_handshake_cid, rem_cid
3099 );
3100 return Ok(());
3101 }
3102 self.on_path_validated();
3103
3104 self.process_early_payload(now, packet)?;
3105 if self.state.is_closed() {
3106 return Ok(());
3107 }
3108
3109 if self.crypto.is_handshaking() {
3110 trace!("handshake ongoing");
3111 return Ok(());
3112 }
3113
3114 if self.side.is_client() {
3115 let params =
3117 self.crypto
3118 .transport_parameters()?
3119 .ok_or_else(|| TransportError {
3120 code: TransportErrorCode::crypto(0x6d),
3121 frame: None,
3122 reason: "transport parameters missing".into(),
3123 })?;
3124
3125 if self.has_0rtt() {
3126 if !self.crypto.early_data_accepted().unwrap() {
3127 debug_assert!(self.side.is_client());
3128 debug!("0-RTT rejected");
3129 self.accepted_0rtt = false;
3130 self.streams.zero_rtt_rejected();
3131
3132 self.spaces[SpaceId::Data].pending = Retransmits::default();
3134
3135 let sent_packets =
3137 mem::take(&mut self.spaces[SpaceId::Data].sent_packets);
3138 for (pn, packet) in sent_packets {
3139 self.remove_in_flight(pn, &packet);
3140 }
3141 } else {
3142 self.accepted_0rtt = true;
3143 params.validate_resumption_from(&self.peer_params)?;
3144 }
3145 }
3146 if let Some(token) = params.stateless_reset_token {
3147 self.endpoint_events
3148 .push_back(EndpointEventInner::ResetToken(self.path.remote, token));
3149 }
3150 self.handle_peer_params(params)?;
3151 self.issue_first_cids(now);
3152 } else {
3153 self.spaces[SpaceId::Data].pending.handshake_done = true;
3155 self.discard_space(now, SpaceId::Handshake);
3156 }
3157
3158 self.events.push_back(Event::Connected);
3159 self.state = State::Established;
3160 trace!("established");
3161 Ok(())
3162 }
3163 Header::Initial(InitialHeader {
3164 src_cid: rem_cid, ..
3165 }) => {
3166 if !state.rem_cid_set {
3167 trace!("switching remote CID to {}", rem_cid);
3168 let mut state = state.clone();
3169 self.rem_cids.update_initial_cid(rem_cid);
3170 self.rem_handshake_cid = rem_cid;
3171 self.orig_rem_cid = rem_cid;
3172 state.rem_cid_set = true;
3173 self.state = State::Handshake(state);
3174 } else if rem_cid != self.rem_handshake_cid {
3175 debug!(
3176 "discarding packet with mismatched remote CID: {} != {}",
3177 self.rem_handshake_cid, rem_cid
3178 );
3179 return Ok(());
3180 }
3181
3182 let starting_space = self.highest_space;
3183 self.process_early_payload(now, packet)?;
3184
3185 if self.side.is_server()
3186 && starting_space == SpaceId::Initial
3187 && self.highest_space != SpaceId::Initial
3188 {
3189 let params =
3190 self.crypto
3191 .transport_parameters()?
3192 .ok_or_else(|| TransportError {
3193 code: TransportErrorCode::crypto(0x6d),
3194 frame: None,
3195 reason: "transport parameters missing".into(),
3196 })?;
3197 self.handle_peer_params(params)?;
3198 self.issue_first_cids(now);
3199 self.init_0rtt();
3200 }
3201 Ok(())
3202 }
3203 Header::Long {
3204 ty: LongType::ZeroRtt,
3205 ..
3206 } => {
3207 self.process_payload(now, remote, number.unwrap(), packet)?;
3208 Ok(())
3209 }
3210 Header::VersionNegotiate { .. } => {
3211 if self.total_authed_packets > 1 {
3212 return Ok(());
3213 }
3214 let supported = packet
3215 .payload
3216 .chunks(4)
3217 .any(|x| match <[u8; 4]>::try_from(x) {
3218 Ok(version) => self.version == u32::from_be_bytes(version),
3219 Err(_) => false,
3220 });
3221 if supported {
3222 return Ok(());
3223 }
3224 debug!("remote doesn't support our version");
3225 Err(ConnectionError::VersionMismatch)
3226 }
3227 Header::Short { .. } => unreachable!(
3228 "short packets received during handshake are discarded in handle_packet"
3229 ),
3230 }
3231 }
3232
3233 fn process_early_payload(
3235 &mut self,
3236 now: Instant,
3237 packet: Packet,
3238 ) -> Result<(), TransportError> {
3239 debug_assert_ne!(packet.header.space(), SpaceId::Data);
3240 let payload_len = packet.payload.len();
3241 let mut ack_eliciting = false;
3242 for result in frame::Iter::new(packet.payload.freeze())? {
3243 let frame = result?;
3244 let span = match frame {
3245 Frame::Padding => continue,
3246 _ => Some(trace_span!("frame", ty = %frame.ty())),
3247 };
3248
3249 self.stats.frame_rx.record(&frame);
3250
3251 let _guard = span.as_ref().map(|x| x.enter());
3252 ack_eliciting |= frame.is_ack_eliciting();
3253
3254 match frame {
3256 Frame::Padding | Frame::Ping => {}
3257 Frame::Crypto(frame) => {
3258 self.read_crypto(packet.header.space(), &frame, payload_len)?;
3259 }
3260 Frame::Ack(ack) => {
3261 self.on_ack_received(now, packet.header.space(), ack)?;
3262 }
3263 Frame::Close(reason) => {
3264 self.error = Some(reason.into());
3265 self.state = State::Draining;
3266 return Ok(());
3267 }
3268 _ => {
3269 let mut err =
3270 TransportError::PROTOCOL_VIOLATION("illegal frame type in handshake");
3271 err.frame = Some(frame.ty());
3272 return Err(err);
3273 }
3274 }
3275 }
3276
3277 if ack_eliciting {
3278 self.spaces[packet.header.space()]
3280 .pending_acks
3281 .set_immediate_ack_required();
3282 }
3283
3284 self.write_crypto();
3285 Ok(())
3286 }
3287
3288 fn process_payload(
3289 &mut self,
3290 now: Instant,
3291 remote: SocketAddr,
3292 number: u64,
3293 packet: Packet,
3294 ) -> Result<(), TransportError> {
3295 let payload = packet.payload.freeze();
3296 let mut is_probing_packet = true;
3297 let mut close = None;
3298 let payload_len = payload.len();
3299 let mut ack_eliciting = false;
3300 for result in frame::Iter::new(payload)? {
3301 let frame = result?;
3302 let span = match frame {
3303 Frame::Padding => continue,
3304 _ => Some(trace_span!("frame", ty = %frame.ty())),
3305 };
3306
3307 self.stats.frame_rx.record(&frame);
3308 match &frame {
3311 Frame::Crypto(f) => {
3312 trace!(offset = f.offset, len = f.data.len(), "got crypto frame");
3313 }
3314 Frame::Stream(f) => {
3315 trace!(id = %f.id, offset = f.offset, len = f.data.len(), fin = f.fin, "got stream frame");
3316 }
3317 Frame::Datagram(f) => {
3318 trace!(len = f.data.len(), "got datagram frame");
3319 }
3320 f => {
3321 trace!("got frame {:?}", f);
3322 }
3323 }
3324
3325 let _guard = span.as_ref().map(|x| x.enter());
3326 if packet.header.is_0rtt() {
3327 match frame {
3328 Frame::Crypto(_) | Frame::Close(Close::Application(_)) => {
3329 return Err(TransportError::PROTOCOL_VIOLATION(
3330 "illegal frame type in 0-RTT",
3331 ));
3332 }
3333 _ => {}
3334 }
3335 }
3336 ack_eliciting |= frame.is_ack_eliciting();
3337
3338 match frame {
3340 Frame::Padding
3341 | Frame::PathChallenge(_)
3342 | Frame::PathResponse(_)
3343 | Frame::NewConnectionId(_) => {}
3344 _ => {
3345 is_probing_packet = false;
3346 }
3347 }
3348 match frame {
3349 Frame::Crypto(frame) => {
3350 self.read_crypto(SpaceId::Data, &frame, payload_len)?;
3351 }
3352 Frame::Stream(frame) => {
3353 if self.streams.received(frame, payload_len)?.should_transmit() {
3354 self.spaces[SpaceId::Data].pending.max_data = true;
3355 }
3356 }
3357 Frame::Ack(ack) => {
3358 self.on_ack_received(now, SpaceId::Data, ack)?;
3359 }
3360 Frame::Padding | Frame::Ping => {}
3361 Frame::Close(reason) => {
3362 close = Some(reason);
3363 }
3364 Frame::PathChallenge(token) => {
3365 self.path_responses.push(number, token, remote);
3366 if remote == self.path.remote {
3367 match self.peer_supports_ack_frequency() {
3370 true => self.immediate_ack(),
3371 false => self.ping(),
3372 }
3373 }
3374 }
3375 Frame::PathResponse(token) => {
3376 if self.path.challenge == Some(token) && remote == self.path.remote {
3377 trace!("new path validated");
3378 self.timers.stop(Timer::PathValidation);
3379 self.path.challenge = None;
3380 self.path.validated = true;
3381 if let Some((_, ref mut prev_path)) = self.prev_path {
3382 prev_path.challenge = None;
3383 prev_path.challenge_pending = false;
3384 }
3385 self.on_path_validated();
3386 } else if let Some(nat_traversal) = &mut self.nat_traversal {
3387 match nat_traversal.handle_validation_success(remote, token, now) {
3389 Ok(sequence) => {
3390 trace!(
3391 "NAT traversal candidate {} validated for sequence {}",
3392 remote, sequence
3393 );
3394
3395 if nat_traversal.handle_coordination_success(remote, now) {
3397 trace!("Coordination succeeded via {}", remote);
3398
3399 let can_migrate = match &self.side {
3401 ConnectionSide::Client { .. } => true, ConnectionSide::Server { server_config } => {
3403 server_config.migration
3404 }
3405 };
3406
3407 if can_migrate {
3408 let best_pairs = nat_traversal.get_best_succeeded_pairs();
3410 if let Some(best) = best_pairs.first() {
3411 if best.remote_addr == remote
3412 && best.remote_addr != self.path.remote
3413 {
3414 debug!(
3415 "NAT traversal found better path, initiating migration"
3416 );
3417 if let Err(e) =
3419 self.migrate_to_nat_traversal_path(now)
3420 {
3421 warn!(
3422 "Failed to migrate to NAT traversal path: {:?}",
3423 e
3424 );
3425 }
3426 }
3427 }
3428 }
3429 } else {
3430 if nat_traversal.mark_pair_succeeded(remote) {
3432 trace!("NAT traversal pair succeeded for {}", remote);
3433 }
3434 }
3435 }
3436 Err(NatTraversalError::ChallengeMismatch) => {
3437 debug!(
3438 "PATH_RESPONSE challenge mismatch for NAT candidate {}",
3439 remote
3440 );
3441 }
3442 Err(e) => {
3443 debug!("NAT traversal validation error: {}", e);
3444 }
3445 }
3446 } else {
3447 debug!(token, "ignoring invalid PATH_RESPONSE");
3448 }
3449 }
3450 Frame::MaxData(bytes) => {
3451 self.streams.received_max_data(bytes);
3452 }
3453 Frame::MaxStreamData { id, offset } => {
3454 self.streams.received_max_stream_data(id, offset)?;
3455 }
3456 Frame::MaxStreams { dir, count } => {
3457 self.streams.received_max_streams(dir, count)?;
3458 }
3459 Frame::ResetStream(frame) => {
3460 if self.streams.received_reset(frame)?.should_transmit() {
3461 self.spaces[SpaceId::Data].pending.max_data = true;
3462 }
3463 }
3464 Frame::DataBlocked { offset } => {
3465 debug!(offset, "peer claims to be blocked at connection level");
3466 }
3467 Frame::StreamDataBlocked { id, offset } => {
3468 if id.initiator() == self.side.side() && id.dir() == Dir::Uni {
3469 debug!("got STREAM_DATA_BLOCKED on send-only {}", id);
3470 return Err(TransportError::STREAM_STATE_ERROR(
3471 "STREAM_DATA_BLOCKED on send-only stream",
3472 ));
3473 }
3474 debug!(
3475 stream = %id,
3476 offset, "peer claims to be blocked at stream level"
3477 );
3478 }
3479 Frame::StreamsBlocked { dir, limit } => {
3480 if limit > MAX_STREAM_COUNT {
3481 return Err(TransportError::FRAME_ENCODING_ERROR(
3482 "unrepresentable stream limit",
3483 ));
3484 }
3485 debug!(
3486 "peer claims to be blocked opening more than {} {} streams",
3487 limit, dir
3488 );
3489 }
3490 Frame::StopSending(frame::StopSending { id, error_code }) => {
3491 if id.initiator() != self.side.side() {
3492 if id.dir() == Dir::Uni {
3493 debug!("got STOP_SENDING on recv-only {}", id);
3494 return Err(TransportError::STREAM_STATE_ERROR(
3495 "STOP_SENDING on recv-only stream",
3496 ));
3497 }
3498 } else if self.streams.is_local_unopened(id) {
3499 return Err(TransportError::STREAM_STATE_ERROR(
3500 "STOP_SENDING on unopened stream",
3501 ));
3502 }
3503 self.streams.received_stop_sending(id, error_code);
3504 }
3505 Frame::RetireConnectionId { sequence } => {
3506 let allow_more_cids = self
3507 .local_cid_state
3508 .on_cid_retirement(sequence, self.peer_params.issue_cids_limit())?;
3509 self.endpoint_events
3510 .push_back(EndpointEventInner::RetireConnectionId(
3511 now,
3512 sequence,
3513 allow_more_cids,
3514 ));
3515 }
3516 Frame::NewConnectionId(frame) => {
3517 trace!(
3518 sequence = frame.sequence,
3519 id = %frame.id,
3520 retire_prior_to = frame.retire_prior_to,
3521 );
3522 if self.rem_cids.active().is_empty() {
3523 return Err(TransportError::PROTOCOL_VIOLATION(
3524 "NEW_CONNECTION_ID when CIDs aren't in use",
3525 ));
3526 }
3527 if frame.retire_prior_to > frame.sequence {
3528 return Err(TransportError::PROTOCOL_VIOLATION(
3529 "NEW_CONNECTION_ID retiring unissued CIDs",
3530 ));
3531 }
3532
3533 use crate::cid_queue::InsertError;
3534 match self.rem_cids.insert(frame) {
3535 Ok(None) => {}
3536 Ok(Some((retired, reset_token))) => {
3537 let pending_retired =
3538 &mut self.spaces[SpaceId::Data].pending.retire_cids;
3539 const MAX_PENDING_RETIRED_CIDS: u64 = CidQueue::LEN as u64 * 10;
3542 if (pending_retired.len() as u64)
3545 .saturating_add(retired.end.saturating_sub(retired.start))
3546 > MAX_PENDING_RETIRED_CIDS
3547 {
3548 return Err(TransportError::CONNECTION_ID_LIMIT_ERROR(
3549 "queued too many retired CIDs",
3550 ));
3551 }
3552 pending_retired.extend(retired);
3553 self.set_reset_token(reset_token);
3554 }
3555 Err(InsertError::ExceedsLimit) => {
3556 return Err(TransportError::CONNECTION_ID_LIMIT_ERROR(""));
3557 }
3558 Err(InsertError::Retired) => {
3559 trace!("discarding already-retired");
3560 self.spaces[SpaceId::Data]
3564 .pending
3565 .retire_cids
3566 .push(frame.sequence);
3567 continue;
3568 }
3569 };
3570
3571 if self.side.is_server() && self.rem_cids.active_seq() == 0 {
3572 self.update_rem_cid();
3575 }
3576 }
3577 Frame::NewToken(NewToken { token }) => {
3578 let ConnectionSide::Client {
3579 token_store,
3580 server_name,
3581 ..
3582 } = &self.side
3583 else {
3584 return Err(TransportError::PROTOCOL_VIOLATION("client sent NEW_TOKEN"));
3585 };
3586 if token.is_empty() {
3587 return Err(TransportError::FRAME_ENCODING_ERROR("empty token"));
3588 }
3589 trace!("got new token");
3590 token_store.insert(server_name, token);
3591 }
3592 Frame::Datagram(datagram) => {
3593 let result = self
3594 .datagrams
3595 .received(datagram, &self.config.datagram_receive_buffer_size)?;
3596 if result.was_empty {
3597 self.events.push_back(Event::DatagramReceived);
3598 }
3599 if result.dropped_count > 0 {
3600 let drop_counts = DatagramDropStats {
3601 datagrams: result.dropped_count as u64,
3602 bytes: result.dropped_bytes as u64,
3603 };
3604 self.stats
3605 .datagram_drops
3606 .record(drop_counts.datagrams, drop_counts.bytes);
3607 self.events.push_back(Event::DatagramDropped(drop_counts));
3608 }
3609 }
3610 Frame::AckFrequency(ack_frequency) => {
3611 let space = &mut self.spaces[SpaceId::Data];
3613
3614 if !self
3615 .ack_frequency
3616 .ack_frequency_received(&ack_frequency, &mut space.pending_acks)?
3617 {
3618 continue;
3620 }
3621
3622 if let Some(timeout) = space
3625 .pending_acks
3626 .max_ack_delay_timeout(self.ack_frequency.max_ack_delay)
3627 {
3628 self.timers.set(Timer::MaxAckDelay, timeout);
3629 }
3630 }
3631 Frame::ImmediateAck => {
3632 self.spaces[SpaceId::Data]
3634 .pending_acks
3635 .set_immediate_ack_required();
3636 }
3637 Frame::HandshakeDone => {
3638 if self.side.is_server() {
3639 return Err(TransportError::PROTOCOL_VIOLATION(
3640 "client sent HANDSHAKE_DONE",
3641 ));
3642 }
3643 if self.spaces[SpaceId::Handshake].crypto.is_some() {
3644 self.discard_space(now, SpaceId::Handshake);
3645 }
3646 }
3647 Frame::AddAddress(add_address) => {
3648 self.handle_add_address(&add_address, now)?;
3649 }
3650 Frame::PunchMeNow(punch_me_now) => {
3651 self.handle_punch_me_now(&punch_me_now, now)?;
3652 }
3653 Frame::RemoveAddress(remove_address) => {
3654 self.handle_remove_address(&remove_address)?;
3655 }
3656 Frame::ObservedAddress(observed_address) => {
3657 self.handle_observed_address_frame(&observed_address, now)?;
3658 }
3659 Frame::TryConnectTo(try_connect_to) => {
3660 self.handle_try_connect_to(&try_connect_to, now)?;
3661 }
3662 Frame::TryConnectToResponse(response) => {
3663 self.handle_try_connect_to_response(&response)?;
3664 }
3665 }
3666 }
3667
3668 let space = &mut self.spaces[SpaceId::Data];
3669 if space
3670 .pending_acks
3671 .packet_received(now, number, ack_eliciting, &space.dedup)
3672 {
3673 self.timers
3674 .set(Timer::MaxAckDelay, now + self.ack_frequency.max_ack_delay);
3675 }
3676
3677 let pending = &mut self.spaces[SpaceId::Data].pending;
3682 self.streams.queue_max_stream_id(pending);
3683
3684 if let Some(reason) = close {
3685 self.error = Some(reason.into());
3686 self.state = State::Draining;
3687 self.close = true;
3688 }
3689
3690 if remote != self.path.remote
3691 && !is_probing_packet
3692 && number == self.spaces[SpaceId::Data].rx_packet
3693 {
3694 let ConnectionSide::Server { ref server_config } = self.side else {
3695 return Err(TransportError::PROTOCOL_VIOLATION(
3696 "packets from unknown remote should be dropped by clients",
3697 ));
3698 };
3699 debug_assert!(
3700 server_config.migration,
3701 "migration-initiating packets should have been dropped immediately"
3702 );
3703 self.migrate(now, remote);
3704 self.update_rem_cid();
3706 self.spin = false;
3707 }
3708
3709 Ok(())
3710 }
3711
3712 fn migrate(&mut self, now: Instant, remote: SocketAddr) {
3713 trace!(%remote, "migration initiated");
3714 let mut new_path = if remote.is_ipv4() && remote.ip() == self.path.remote.ip() {
3718 PathData::from_previous(remote, &self.path, now)
3719 } else {
3720 let peer_max_udp_payload_size =
3721 u16::try_from(self.peer_params.max_udp_payload_size.into_inner())
3722 .unwrap_or(u16::MAX);
3723 PathData::new(
3724 remote,
3725 self.allow_mtud,
3726 Some(peer_max_udp_payload_size),
3727 now,
3728 &self.config,
3729 )
3730 };
3731 new_path.challenge = Some(self.rng.r#gen());
3732 new_path.challenge_pending = true;
3733 let prev_pto = self.pto(SpaceId::Data);
3734
3735 let mut prev = mem::replace(&mut self.path, new_path);
3736 if prev.challenge.is_none() {
3738 prev.challenge = Some(self.rng.r#gen());
3739 prev.challenge_pending = true;
3740 self.prev_path = Some((self.rem_cids.active(), prev));
3743 }
3744
3745 self.timers.set(
3746 Timer::PathValidation,
3747 now + 3 * cmp::max(self.pto(SpaceId::Data), prev_pto),
3748 );
3749 }
3750
3751 pub fn local_address_changed(&mut self) {
3753 self.update_rem_cid();
3754 self.ping();
3755 }
3756
3757 pub fn migrate_to_nat_traversal_path(&mut self, now: Instant) -> Result<(), TransportError> {
3759 let (remote_addr, local_addr) = {
3761 let nat_state = self
3762 .nat_traversal
3763 .as_ref()
3764 .ok_or_else(|| TransportError::PROTOCOL_VIOLATION("NAT traversal not enabled"))?;
3765
3766 let best_pairs = nat_state.get_best_succeeded_pairs();
3768 if best_pairs.is_empty() {
3769 return Err(TransportError::PROTOCOL_VIOLATION(
3770 "No validated NAT traversal paths",
3771 ));
3772 }
3773
3774 let best_path = best_pairs
3776 .iter()
3777 .find(|pair| pair.remote_addr != self.path.remote)
3778 .or_else(|| best_pairs.first());
3779
3780 let best_path = best_path.ok_or_else(|| {
3781 TransportError::PROTOCOL_VIOLATION("No suitable NAT traversal path")
3782 })?;
3783
3784 debug!(
3785 "Migrating to NAT traversal path: {} -> {} (priority: {})",
3786 self.path.remote, best_path.remote_addr, best_path.priority
3787 );
3788
3789 (best_path.remote_addr, best_path.local_addr)
3790 };
3791
3792 self.migrate(now, remote_addr);
3794
3795 if local_addr != SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0) {
3797 self.local_ip = Some(local_addr.ip());
3798 }
3799
3800 self.path.challenge_pending = true;
3802
3803 Ok(())
3804 }
3805
3806 fn update_rem_cid(&mut self) {
3808 let (reset_token, retired) = match self.rem_cids.next() {
3809 Some(x) => x,
3810 None => return,
3811 };
3812
3813 self.spaces[SpaceId::Data]
3815 .pending
3816 .retire_cids
3817 .extend(retired);
3818 self.set_reset_token(reset_token);
3819 }
3820
3821 fn set_reset_token(&mut self, reset_token: ResetToken) {
3822 self.endpoint_events
3823 .push_back(EndpointEventInner::ResetToken(
3824 self.path.remote,
3825 reset_token,
3826 ));
3827 self.peer_params.stateless_reset_token = Some(reset_token);
3828 }
3829
3830 fn handle_encode_error(&mut self, now: Instant, context: &'static str) {
3831 tracing::error!("VarInt overflow while encoding {context}");
3832 self.close_inner(
3833 now,
3834 Close::from(TransportError::INTERNAL_ERROR(
3835 "varint overflow during encoding",
3836 )),
3837 );
3838 }
3839
3840 fn encode_or_close(
3841 &mut self,
3842 now: Instant,
3843 result: Result<(), VarIntBoundsExceeded>,
3844 context: &'static str,
3845 ) -> bool {
3846 if result.is_err() {
3847 self.handle_encode_error(now, context);
3848 return false;
3849 }
3850 true
3851 }
3852
3853 fn issue_first_cids(&mut self, now: Instant) {
3855 if self.local_cid_state.cid_len() == 0 {
3856 return;
3857 }
3858
3859 let mut n = self.peer_params.issue_cids_limit() - 1;
3861 if let ConnectionSide::Server { server_config } = &self.side {
3862 if server_config.has_preferred_address() {
3863 n -= 1;
3865 }
3866 }
3867 self.endpoint_events
3868 .push_back(EndpointEventInner::NeedIdentifiers(now, n));
3869 }
3870
3871 fn populate_packet(
3872 &mut self,
3873 now: Instant,
3874 space_id: SpaceId,
3875 buf: &mut Vec<u8>,
3876 max_size: usize,
3877 pn: u64,
3878 ) -> SentFrames {
3879 let mut sent = SentFrames::default();
3880 let space = &mut self.spaces[space_id];
3881 let is_0rtt = space_id == SpaceId::Data && space.crypto.is_none();
3882 space.pending_acks.maybe_ack_non_eliciting();
3883 macro_rules! encode_or_close {
3884 ($result:expr, $context:expr) => {{
3885 if $result.is_err() {
3886 drop(space);
3887 self.handle_encode_error(now, $context);
3888 return sent;
3889 }
3890 }};
3891 }
3892
3893 if !is_0rtt && mem::replace(&mut space.pending.handshake_done, false) {
3895 encode_or_close!(
3896 frame::FrameType::HANDSHAKE_DONE.try_encode(buf),
3897 "HANDSHAKE_DONE"
3898 );
3899 sent.retransmits.get_or_create().handshake_done = true;
3900 self.stats.frame_tx.handshake_done =
3902 self.stats.frame_tx.handshake_done.saturating_add(1);
3903 }
3904
3905 if mem::replace(&mut space.ping_pending, false) {
3907 trace!("PING");
3908 encode_or_close!(frame::FrameType::PING.try_encode(buf), "PING");
3909 sent.non_retransmits = true;
3910 self.stats.frame_tx.ping += 1;
3911 }
3912
3913 if mem::replace(&mut space.immediate_ack_pending, false) {
3915 trace!("IMMEDIATE_ACK");
3916 encode_or_close!(
3917 frame::FrameType::IMMEDIATE_ACK.try_encode(buf),
3918 "IMMEDIATE_ACK"
3919 );
3920 sent.non_retransmits = true;
3921 self.stats.frame_tx.immediate_ack += 1;
3922 }
3923
3924 if space.pending_acks.can_send() {
3926 let ack_result = Self::populate_acks(
3927 now,
3928 self.receiving_ecn,
3929 &mut sent,
3930 space,
3931 buf,
3932 &mut self.stats,
3933 );
3934 encode_or_close!(ack_result, "ACK");
3935 }
3936
3937 if mem::replace(&mut space.pending.ack_frequency, false) {
3939 let sequence_number = self.ack_frequency.next_sequence_number();
3940
3941 let config = self.config.ack_frequency_config.as_ref().unwrap();
3943
3944 let max_ack_delay = self.ack_frequency.candidate_max_ack_delay(
3946 self.path.rtt.get(),
3947 config,
3948 &self.peer_params,
3949 );
3950
3951 trace!(?max_ack_delay, "ACK_FREQUENCY");
3952
3953 encode_or_close!(
3954 (frame::AckFrequency {
3955 sequence: sequence_number,
3956 ack_eliciting_threshold: config.ack_eliciting_threshold,
3957 request_max_ack_delay: max_ack_delay
3958 .as_micros()
3959 .try_into()
3960 .unwrap_or(VarInt::MAX),
3961 reordering_threshold: config.reordering_threshold,
3962 })
3963 .try_encode(buf),
3964 "ACK_FREQUENCY"
3965 );
3966
3967 sent.retransmits.get_or_create().ack_frequency = true;
3968
3969 self.ack_frequency.ack_frequency_sent(pn, max_ack_delay);
3970 self.stats.frame_tx.ack_frequency += 1;
3971 }
3972
3973 if buf.len() + 9 < max_size && space_id == SpaceId::Data {
3975 if let Some(token) = self.path.challenge {
3977 self.path.challenge_pending = false;
3979 sent.non_retransmits = true;
3980 sent.requires_padding = true;
3981 trace!("PATH_CHALLENGE {:08x}", token);
3982 encode_or_close!(
3983 frame::FrameType::PATH_CHALLENGE.try_encode(buf),
3984 "PATH_CHALLENGE"
3985 );
3986 buf.write(token);
3987 self.stats.frame_tx.path_challenge += 1;
3988 }
3989
3990 }
3993
3994 if buf.len() + 9 < max_size && space_id == SpaceId::Data {
3996 if let Some(token) = self.path_responses.pop_on_path(self.path.remote) {
3997 sent.non_retransmits = true;
3998 sent.requires_padding = true;
3999 trace!("PATH_RESPONSE {:08x}", token);
4000 encode_or_close!(
4001 frame::FrameType::PATH_RESPONSE.try_encode(buf),
4002 "PATH_RESPONSE"
4003 );
4004 buf.write(token);
4005 self.stats.frame_tx.path_response += 1;
4006 }
4007 }
4008
4009 while buf.len() + frame::Crypto::SIZE_BOUND < max_size && !is_0rtt {
4011 let mut frame = match space.pending.crypto.pop_front() {
4012 Some(x) => x,
4013 None => break,
4014 };
4015
4016 let max_crypto_data_size = max_size
4021 - buf.len()
4022 - 1 - VarInt::size(unsafe { VarInt::from_u64_unchecked(frame.offset) })
4024 - 2; let available_space = max_size - buf.len();
4028 let remaining_data = frame.data.len();
4029 let optimal_size = self
4030 .pqc_state
4031 .calculate_crypto_frame_size(available_space, remaining_data);
4032
4033 let len = frame
4034 .data
4035 .len()
4036 .min(2usize.pow(14) - 1)
4037 .min(max_crypto_data_size)
4038 .min(optimal_size);
4039
4040 let data = frame.data.split_to(len);
4041 let truncated = frame::Crypto {
4042 offset: frame.offset,
4043 data,
4044 };
4045 trace!(
4046 "CRYPTO: off {} len {}",
4047 truncated.offset,
4048 truncated.data.len()
4049 );
4050 encode_or_close!(truncated.try_encode(buf), "CRYPTO");
4051 self.stats.frame_tx.crypto += 1;
4052 sent.retransmits.get_or_create().crypto.push_back(truncated);
4053 if !frame.data.is_empty() {
4054 frame.offset += len as u64;
4055 space.pending.crypto.push_front(frame);
4056 }
4057 }
4058
4059 if space_id == SpaceId::Data {
4060 let control_result = self.streams.write_control_frames(
4061 buf,
4062 &mut space.pending,
4063 &mut sent.retransmits,
4064 &mut self.stats.frame_tx,
4065 max_size,
4066 );
4067 encode_or_close!(control_result, "control frames");
4068 }
4069
4070 while buf.len() + 44 < max_size {
4072 let issued = match space.pending.new_cids.pop() {
4073 Some(x) => x,
4074 None => break,
4075 };
4076 trace!(
4077 sequence = issued.sequence,
4078 id = %issued.id,
4079 "NEW_CONNECTION_ID"
4080 );
4081 encode_or_close!(
4082 (frame::NewConnectionId {
4083 sequence: issued.sequence,
4084 retire_prior_to: self.local_cid_state.retire_prior_to(),
4085 id: issued.id,
4086 reset_token: issued.reset_token,
4087 })
4088 .try_encode(buf),
4089 "NEW_CONNECTION_ID"
4090 );
4091 sent.retransmits.get_or_create().new_cids.push(issued);
4092 self.stats.frame_tx.new_connection_id += 1;
4093 }
4094
4095 while buf.len() + frame::RETIRE_CONNECTION_ID_SIZE_BOUND < max_size {
4097 let seq = match space.pending.retire_cids.pop() {
4098 Some(x) => x,
4099 None => break,
4100 };
4101 trace!(sequence = seq, "RETIRE_CONNECTION_ID");
4102 encode_or_close!(
4103 frame::FrameType::RETIRE_CONNECTION_ID.try_encode(buf),
4104 "RETIRE_CONNECTION_ID"
4105 );
4106 encode_or_close!(buf.write_var(seq), "RETIRE_CONNECTION_ID seq");
4107 sent.retransmits.get_or_create().retire_cids.push(seq);
4108 self.stats.frame_tx.retire_connection_id += 1;
4109 }
4110
4111 let mut sent_datagrams = false;
4113 while buf.len() + Datagram::SIZE_BOUND < max_size && space_id == SpaceId::Data {
4114 match self.datagrams.write(buf, max_size) {
4115 true => {
4116 sent_datagrams = true;
4117 sent.non_retransmits = true;
4118 self.stats.frame_tx.datagram += 1;
4119 }
4120 false => break,
4121 }
4122 }
4123 if self.datagrams.send_blocked && sent_datagrams {
4124 self.events.push_back(Event::DatagramsUnblocked);
4125 self.datagrams.send_blocked = false;
4126 }
4127
4128 while let Some(remote_addr) = space.pending.new_tokens.pop() {
4130 debug_assert_eq!(space_id, SpaceId::Data);
4131 let ConnectionSide::Server { server_config } = &self.side else {
4132 debug_assert!(false, "NEW_TOKEN frames should not be enqueued by clients");
4134 continue;
4135 };
4136
4137 if remote_addr != self.path.remote {
4138 continue;
4143 }
4144
4145 if self.delay_new_token_until_binding && self.peer_id_for_tokens.is_none() {
4148 space.pending.new_tokens.push(remote_addr);
4150 break;
4151 }
4152
4153 let token = match crate::token_v2::encode_validation_token_with_rng(
4154 &server_config.token_key,
4155 remote_addr.ip(),
4156 server_config.time_source.now(),
4157 &mut self.rng,
4158 ) {
4159 Ok(token) => token,
4160 Err(err) => {
4161 error!(?err, "failed to encode validation token");
4162 continue;
4163 }
4164 };
4165 let new_token = NewToken {
4166 token: token.into(),
4167 };
4168
4169 if buf.len() + new_token.size() >= max_size {
4170 space.pending.new_tokens.push(remote_addr);
4171 break;
4172 }
4173
4174 encode_or_close!(new_token.try_encode(buf), "NEW_TOKEN");
4175 sent.retransmits
4176 .get_or_create()
4177 .new_tokens
4178 .push(remote_addr);
4179 self.stats.frame_tx.new_token += 1;
4180 }
4181
4182 while buf.len() + frame::AddAddress::SIZE_BOUND < max_size && space_id == SpaceId::Data {
4184 let add_address = match space.pending.add_addresses.pop() {
4185 Some(x) => x,
4186 None => break,
4187 };
4188 trace!(
4189 sequence = %add_address.sequence,
4190 address = %add_address.address,
4191 "ADD_ADDRESS"
4192 );
4193 if self.nat_traversal_frame_config.use_rfc_format {
4195 encode_or_close!(add_address.try_encode_rfc(buf), "ADD_ADDRESS (rfc)");
4196 } else {
4197 encode_or_close!(add_address.try_encode_legacy(buf), "ADD_ADDRESS (legacy)");
4198 }
4199 sent.retransmits
4200 .get_or_create()
4201 .add_addresses
4202 .push(add_address);
4203 self.stats.frame_tx.add_address += 1;
4204 }
4205
4206 while buf.len() + frame::PunchMeNow::SIZE_BOUND < max_size && space_id == SpaceId::Data {
4208 let punch_me_now = match space.pending.punch_me_now.pop() {
4209 Some(x) => x,
4210 None => break,
4211 };
4212 trace!(
4213 round = %punch_me_now.round,
4214 paired_with_sequence_number = %punch_me_now.paired_with_sequence_number,
4215 "PUNCH_ME_NOW"
4216 );
4217 if self.nat_traversal_frame_config.use_rfc_format {
4219 encode_or_close!(punch_me_now.try_encode_rfc(buf), "PUNCH_ME_NOW (rfc)");
4220 } else {
4221 encode_or_close!(punch_me_now.try_encode_legacy(buf), "PUNCH_ME_NOW (legacy)");
4222 }
4223 sent.retransmits
4224 .get_or_create()
4225 .punch_me_now
4226 .push(punch_me_now);
4227 self.stats.frame_tx.punch_me_now += 1;
4228 }
4229
4230 while buf.len() + frame::RemoveAddress::SIZE_BOUND < max_size && space_id == SpaceId::Data {
4232 let remove_address = match space.pending.remove_addresses.pop() {
4233 Some(x) => x,
4234 None => break,
4235 };
4236 trace!(
4237 sequence = %remove_address.sequence,
4238 "REMOVE_ADDRESS"
4239 );
4240 encode_or_close!(remove_address.try_encode(buf), "REMOVE_ADDRESS");
4242 sent.retransmits
4243 .get_or_create()
4244 .remove_addresses
4245 .push(remove_address);
4246 self.stats.frame_tx.remove_address += 1;
4247 }
4248
4249 while buf.len() + frame::ObservedAddress::SIZE_BOUND < max_size && space_id == SpaceId::Data
4251 {
4252 let observed_address = match space.pending.outbound_observations.pop() {
4253 Some(x) => x,
4254 None => break,
4255 };
4256 info!(
4257 address = %observed_address.address,
4258 sequence = %observed_address.sequence_number,
4259 "populate_packet: ENCODING OBSERVED_ADDRESS into packet"
4260 );
4261 encode_or_close!(observed_address.try_encode(buf), "OBSERVED_ADDRESS");
4262 sent.retransmits
4263 .get_or_create()
4264 .outbound_observations
4265 .push(observed_address);
4266 self.stats.frame_tx.observed_address += 1;
4267 }
4268
4269 if space_id == SpaceId::Data {
4271 sent.stream_frames =
4272 self.streams
4273 .write_stream_frames(buf, max_size, self.config.send_fairness);
4274 self.stats.frame_tx.stream += sent.stream_frames.len() as u64;
4275 }
4276
4277 sent
4278 }
4279
4280 fn populate_acks(
4285 now: Instant,
4286 receiving_ecn: bool,
4287 sent: &mut SentFrames,
4288 space: &mut PacketSpace,
4289 buf: &mut Vec<u8>,
4290 stats: &mut ConnectionStats,
4291 ) -> Result<(), VarIntBoundsExceeded> {
4292 debug_assert!(!space.pending_acks.ranges().is_empty());
4293
4294 debug_assert!(space.crypto.is_some(), "tried to send ACK in 0-RTT");
4296 let ecn = if receiving_ecn {
4297 Some(&space.ecn_counters)
4298 } else {
4299 None
4300 };
4301 sent.largest_acked = space.pending_acks.ranges().max();
4302
4303 let delay_micros = space.pending_acks.ack_delay(now).as_micros() as u64;
4304
4305 let ack_delay_exp = TransportParameters::default().ack_delay_exponent;
4307 let delay = delay_micros >> ack_delay_exp.into_inner();
4308
4309 trace!(
4310 "ACK {:?}, Delay = {}us",
4311 space.pending_acks.ranges(),
4312 delay_micros
4313 );
4314
4315 frame::Ack::try_encode(delay as _, space.pending_acks.ranges(), ecn, buf)?;
4316 stats.frame_tx.acks += 1;
4317 Ok(())
4318 }
4319
4320 fn close_common(&mut self) {
4321 trace!("connection closed");
4322 for &timer in &Timer::VALUES {
4323 self.timers.stop(timer);
4324 }
4325 }
4326
4327 fn set_close_timer(&mut self, now: Instant) {
4328 self.timers
4329 .set(Timer::Close, now + 3 * self.pto(self.highest_space));
4330 }
4331
4332 fn handle_peer_params(&mut self, params: TransportParameters) -> Result<(), TransportError> {
4334 if Some(self.orig_rem_cid) != params.initial_src_cid
4335 || (self.side.is_client()
4336 && (Some(self.initial_dst_cid) != params.original_dst_cid
4337 || self.retry_src_cid != params.retry_src_cid))
4338 {
4339 return Err(TransportError::TRANSPORT_PARAMETER_ERROR(
4340 "CID authentication failure",
4341 ));
4342 }
4343
4344 self.set_peer_params(params);
4345
4346 Ok(())
4347 }
4348
4349 fn set_peer_params(&mut self, params: TransportParameters) {
4350 self.streams.set_params(¶ms);
4351 self.idle_timeout =
4352 negotiate_max_idle_timeout(self.config.max_idle_timeout, Some(params.max_idle_timeout));
4353 trace!("negotiated max idle timeout {:?}", self.idle_timeout);
4354 if let Some(ref info) = params.preferred_address {
4355 self.rem_cids.insert(frame::NewConnectionId {
4356 sequence: 1,
4357 id: info.connection_id,
4358 reset_token: info.stateless_reset_token,
4359 retire_prior_to: 0,
4360 }).expect("preferred address CID is the first received, and hence is guaranteed to be legal");
4361 }
4362 self.ack_frequency.peer_max_ack_delay = get_max_ack_delay(¶ms);
4363
4364 self.negotiate_nat_traversal_capability(¶ms);
4366
4367 let local_has_nat_traversal = self.config.nat_traversal_config.is_some();
4370 let local_supports_rfc = local_has_nat_traversal;
4373 self.nat_traversal_frame_config = frame::nat_traversal_unified::NatTraversalFrameConfig {
4374 use_rfc_format: local_supports_rfc && params.supports_rfc_nat_traversal(),
4376 accept_legacy: true,
4378 };
4379
4380 self.negotiate_address_discovery(¶ms);
4382
4383 self.pqc_state.update_from_peer_params(¶ms);
4385
4386 if self.pqc_state.enabled && self.pqc_state.using_pqc {
4388 trace!("PQC enabled, adjusting MTU discovery for larger handshake packets");
4389 let current_mtu = self.path.mtud.current_mtu();
4393 if current_mtu < self.pqc_state.handshake_mtu {
4394 trace!(
4395 "Current MTU {} is less than PQC handshake MTU {}, will rely on MTU discovery",
4396 current_mtu, self.pqc_state.handshake_mtu
4397 );
4398 }
4399 }
4400
4401 self.peer_params = params;
4402 self.path.mtud.on_peer_max_udp_payload_size_received(
4403 u16::try_from(self.peer_params.max_udp_payload_size.into_inner()).unwrap_or(u16::MAX),
4404 );
4405 }
4406
4407 fn negotiate_nat_traversal_capability(&mut self, params: &TransportParameters) {
4409 let peer_nat_config = match ¶ms.nat_traversal {
4411 Some(config) => config,
4412 None => {
4413 if self.config.nat_traversal_config.is_some() {
4415 debug!(
4416 "Peer does not support NAT traversal, maintaining backward compatibility"
4417 );
4418 self.emit_nat_traversal_capability_event(false);
4419
4420 self.set_nat_traversal_compatibility_mode(false);
4422 }
4423 return;
4424 }
4425 };
4426
4427 let local_nat_config = match &self.config.nat_traversal_config {
4429 Some(config) => config,
4430 None => {
4431 debug!("NAT traversal not enabled locally, ignoring peer support");
4432 self.emit_nat_traversal_capability_event(false);
4433 self.set_nat_traversal_compatibility_mode(false);
4434 return;
4435 }
4436 };
4437
4438 info!("Both peers support NAT traversal, negotiating capabilities");
4440
4441 match self.negotiate_nat_traversal_parameters(local_nat_config, peer_nat_config) {
4443 Ok(negotiated_config) => {
4444 info!("NAT traversal capability negotiated successfully");
4445 self.emit_nat_traversal_capability_event(true);
4446
4447 self.init_nat_traversal_with_negotiated_config(&negotiated_config);
4449
4450 self.set_nat_traversal_compatibility_mode(true);
4452
4453 if matches!(
4455 negotiated_config,
4456 crate::transport_parameters::NatTraversalConfig::ClientSupport
4457 ) {
4458 self.initiate_nat_traversal_process();
4459 }
4460 }
4461 Err(e) => {
4462 warn!("NAT traversal capability negotiation failed: {}", e);
4463 self.emit_nat_traversal_capability_event(false);
4464 self.set_nat_traversal_compatibility_mode(false);
4465 }
4466 }
4467 }
4468
4469 fn emit_nat_traversal_capability_event(&mut self, negotiated: bool) {
4471 if negotiated {
4474 info!("NAT traversal capability successfully negotiated");
4475 } else {
4476 info!("NAT traversal capability not available (peer or local support missing)");
4477 }
4478
4479 }
4482
4483 fn set_nat_traversal_compatibility_mode(&mut self, enabled: bool) {
4485 if enabled {
4486 debug!("NAT traversal enabled for this connection");
4487 } else {
4489 debug!("NAT traversal disabled for this connection (backward compatibility mode)");
4490 if self.nat_traversal.is_some() {
4492 warn!("Clearing NAT traversal state due to compatibility mode");
4493 self.nat_traversal = None;
4494 }
4495 }
4496 }
4497
4498 fn negotiate_nat_traversal_parameters(
4500 &self,
4501 local_config: &crate::transport_parameters::NatTraversalConfig,
4502 peer_config: &crate::transport_parameters::NatTraversalConfig,
4503 ) -> Result<crate::transport_parameters::NatTraversalConfig, String> {
4504 match (local_config, peer_config) {
4509 (
4511 crate::transport_parameters::NatTraversalConfig::ClientSupport,
4512 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4513 concurrency_limit,
4514 },
4515 ) => Ok(
4516 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4517 concurrency_limit: *concurrency_limit,
4518 },
4519 ),
4520 (
4522 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4523 concurrency_limit,
4524 },
4525 crate::transport_parameters::NatTraversalConfig::ClientSupport,
4526 ) => Ok(
4527 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4528 concurrency_limit: *concurrency_limit,
4529 },
4530 ),
4531 (
4533 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4534 concurrency_limit: limit1,
4535 },
4536 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4537 concurrency_limit: limit2,
4538 },
4539 ) => Ok(
4540 crate::transport_parameters::NatTraversalConfig::ServerSupport {
4541 concurrency_limit: (*limit1).min(*limit2),
4542 },
4543 ),
4544 (
4546 crate::transport_parameters::NatTraversalConfig::ClientSupport,
4547 crate::transport_parameters::NatTraversalConfig::ClientSupport,
4548 ) => Err("Both endpoints claim to be NAT traversal clients".to_string()),
4549 }
4550 }
4551
4552 fn init_nat_traversal_with_negotiated_config(
4557 &mut self,
4558 _config: &crate::transport_parameters::NatTraversalConfig,
4559 ) {
4560 let max_candidates = 50; let coordination_timeout = Duration::from_secs(10); self.nat_traversal = Some(NatTraversalState::new(max_candidates, coordination_timeout));
4567
4568 trace!("NAT traversal initialized for symmetric P2P node");
4569
4570 self.prepare_address_observation();
4573 self.schedule_candidate_discovery();
4574 self.prepare_coordination_handling();
4575 }
4576
4577 fn initiate_nat_traversal_process(&mut self) {
4579 if let Some(nat_state) = &mut self.nat_traversal {
4580 match nat_state.start_candidate_discovery() {
4581 Ok(()) => {
4582 debug!("NAT traversal process initiated - candidate discovery started");
4583 self.timers.set(
4585 Timer::NatTraversal,
4586 Instant::now() + Duration::from_millis(100),
4587 );
4588 }
4589 Err(e) => {
4590 warn!("Failed to initiate NAT traversal process: {}", e);
4591 }
4592 }
4593 }
4594 }
4595
4596 fn prepare_address_observation(&mut self) {
4598 debug!("Preparing for address observation as bootstrap node");
4599 }
4602
4603 fn schedule_candidate_discovery(&mut self) {
4605 debug!("Scheduling candidate discovery for client endpoint");
4606 self.timers.set(
4608 Timer::NatTraversal,
4609 Instant::now() + Duration::from_millis(50),
4610 );
4611 }
4612
4613 fn prepare_coordination_handling(&mut self) {
4615 debug!("Preparing to handle coordination requests as server endpoint");
4616 }
4619
4620 fn handle_nat_traversal_timeout(&mut self, now: Instant) {
4622 let timeout_result = if let Some(nat_state) = &mut self.nat_traversal {
4624 nat_state.handle_timeout(now)
4625 } else {
4626 return;
4627 };
4628
4629 match timeout_result {
4631 Ok(actions) => {
4632 for action in actions {
4633 match action {
4634 nat_traversal::TimeoutAction::RetryDiscovery => {
4635 debug!("NAT traversal timeout: retrying candidate discovery");
4636 if let Some(nat_state) = &mut self.nat_traversal {
4637 if let Err(e) = nat_state.start_candidate_discovery() {
4638 warn!("Failed to retry candidate discovery: {}", e);
4639 }
4640 }
4641 }
4642 nat_traversal::TimeoutAction::RetryCoordination => {
4643 debug!("NAT traversal timeout: retrying coordination");
4644 self.timers
4646 .set(Timer::NatTraversal, now + Duration::from_secs(2));
4647 }
4648 nat_traversal::TimeoutAction::StartValidation => {
4649 debug!("NAT traversal timeout: starting path validation");
4650 self.start_nat_traversal_validation(now);
4651 }
4652 nat_traversal::TimeoutAction::Complete => {
4653 debug!("NAT traversal completed successfully");
4654 self.timers.stop(Timer::NatTraversal);
4656 }
4657 nat_traversal::TimeoutAction::Failed => {
4658 warn!("NAT traversal failed after timeout");
4659 self.handle_nat_traversal_failure();
4661 }
4662 }
4663 }
4664 }
4665 Err(e) => {
4666 warn!("NAT traversal timeout handling failed: {}", e);
4667 self.handle_nat_traversal_failure();
4668 }
4669 }
4670 }
4671
4672 fn start_nat_traversal_validation(&mut self, now: Instant) {
4674 if let Some(nat_state) = &mut self.nat_traversal {
4675 let pairs = nat_state.get_next_validation_pairs(3);
4677
4678 for pair in pairs {
4679 let challenge = self.rng.r#gen();
4681 self.path.challenge = Some(challenge);
4682 self.path.challenge_pending = true;
4683
4684 debug!(
4685 "Starting path validation for NAT traversal candidate: {}",
4686 pair.remote_addr
4687 );
4688 }
4689
4690 self.timers
4692 .set(Timer::PathValidation, now + Duration::from_secs(3));
4693 }
4694 }
4695
4696 fn handle_nat_traversal_failure(&mut self) {
4698 warn!("NAT traversal failed, considering fallback options");
4699
4700 self.nat_traversal = None;
4702 self.timers.stop(Timer::NatTraversal);
4703
4704 debug!("NAT traversal disabled for this connection due to failure");
4711 }
4712
4713 pub fn nat_traversal_supported(&self) -> bool {
4715 self.nat_traversal.is_some()
4716 && self.config.nat_traversal_config.is_some()
4717 && self.peer_params.nat_traversal.is_some()
4718 }
4719
4720 pub fn nat_traversal_config(&self) -> Option<&crate::transport_parameters::NatTraversalConfig> {
4722 self.peer_params.nat_traversal.as_ref()
4723 }
4724
4725 pub fn nat_traversal_ready(&self) -> bool {
4727 self.nat_traversal_supported() && matches!(self.state, State::Established)
4728 }
4729
4730 #[allow(dead_code)]
4735 pub(crate) fn nat_traversal_stats(&self) -> Option<nat_traversal::NatTraversalStats> {
4736 self.nat_traversal.as_ref().map(|state| state.stats.clone())
4737 }
4738
4739 #[cfg(test)]
4743 #[allow(dead_code)]
4744 pub(crate) fn force_enable_nat_traversal(&mut self) {
4745 use crate::transport_parameters::NatTraversalConfig;
4746
4747 let config = NatTraversalConfig::ServerSupport {
4749 concurrency_limit: VarInt::from_u32(5),
4750 };
4751
4752 self.peer_params.nat_traversal = Some(config.clone());
4753 self.config = Arc::new({
4754 let mut transport_config = (*self.config).clone();
4755 transport_config.nat_traversal_config = Some(config);
4756 transport_config
4757 });
4758
4759 self.nat_traversal = Some(NatTraversalState::new(8, Duration::from_secs(10)));
4761 }
4762
4763 fn handle_add_address(
4765 &mut self,
4766 add_address: &crate::frame::AddAddress,
4767 now: Instant,
4768 ) -> Result<(), TransportError> {
4769 let nat_state = self.nat_traversal.as_mut().ok_or_else(|| {
4770 TransportError::PROTOCOL_VIOLATION("AddAddress frame without NAT traversal negotiation")
4771 })?;
4772
4773 let normalized_addr = crate::shared::normalize_socket_addr(add_address.address);
4776
4777 info!(
4778 "handle_add_address: RECEIVED ADD_ADDRESS from peer addr={} (normalized={}) seq={} priority={}",
4779 add_address.address, normalized_addr, add_address.sequence, add_address.priority
4780 );
4781
4782 match nat_state.add_remote_candidate(
4783 add_address.sequence,
4784 normalized_addr,
4785 add_address.priority,
4786 now,
4787 ) {
4788 Ok(()) => {
4789 info!(
4790 "Added remote candidate: {} (seq={}, priority={})",
4791 normalized_addr, add_address.sequence, add_address.priority
4792 );
4793
4794 self.endpoint_events.push_back(
4796 crate::shared::EndpointEventInner::PeerAddressAdvertised {
4797 peer_addr: self.path.remote,
4798 advertised_addr: normalized_addr,
4799 },
4800 );
4801
4802 self.trigger_candidate_validation(normalized_addr, now)?;
4804 Ok(())
4805 }
4806 Err(NatTraversalError::TooManyCandidates) => Err(TransportError::PROTOCOL_VIOLATION(
4807 "too many NAT traversal candidates",
4808 )),
4809 Err(NatTraversalError::DuplicateAddress) => {
4810 Ok(())
4812 }
4813 Err(e) => {
4814 warn!("Failed to add remote candidate: {}", e);
4815 Ok(()) }
4817 }
4818 }
4819
4820 fn handle_punch_me_now(
4824 &mut self,
4825 punch_me_now: &crate::frame::PunchMeNow,
4826 now: Instant,
4827 ) -> Result<(), TransportError> {
4828 trace!(
4829 "Received PunchMeNow: round={}, target_seq={}, local_addr={}",
4830 punch_me_now.round, punch_me_now.paired_with_sequence_number, punch_me_now.address
4831 );
4832
4833 let nat_state = self.nat_traversal.as_mut().ok_or_else(|| {
4835 TransportError::PROTOCOL_VIOLATION("PunchMeNow frame without NAT traversal negotiation")
4836 })?;
4837
4838 let target = nat_traversal::PunchTarget {
4841 remote_addr: punch_me_now.address,
4842 remote_sequence: punch_me_now.paired_with_sequence_number,
4843 challenge: self.rng.r#gen(),
4844 };
4845
4846 if let Err(_e) =
4847 nat_state.prime_passive_coordination_target(punch_me_now.round, target, now)
4848 {
4849 debug!(
4850 "Failed to prime passive coordination for round {}",
4851 punch_me_now.round
4852 );
4853 } else {
4854 trace!(
4855 "Passive coordination primed for round {}",
4856 punch_me_now.round
4857 );
4858 }
4859
4860 Ok(())
4861 }
4862
4863 fn handle_remove_address(
4865 &mut self,
4866 remove_address: &crate::frame::RemoveAddress,
4867 ) -> Result<(), TransportError> {
4868 let nat_state = self.nat_traversal.as_mut().ok_or_else(|| {
4869 TransportError::PROTOCOL_VIOLATION(
4870 "RemoveAddress frame without NAT traversal negotiation",
4871 )
4872 })?;
4873
4874 if nat_state.remove_candidate(remove_address.sequence) {
4875 trace!(
4876 "Removed candidate with sequence {}",
4877 remove_address.sequence
4878 );
4879 } else {
4880 trace!(
4881 "Attempted to remove unknown candidate sequence {}",
4882 remove_address.sequence
4883 );
4884 }
4885
4886 Ok(())
4887 }
4888
4889 fn handle_observed_address_frame(
4891 &mut self,
4892 observed_address: &crate::frame::ObservedAddress,
4893 now: Instant,
4894 ) -> Result<(), TransportError> {
4895 tracing::info!(
4896 address = %observed_address.address,
4897 sequence = %observed_address.sequence_number,
4898 from_peer = %self.peer_id_for_tokens.map(|pid| format!("{pid}")).unwrap_or_else(|| "unknown".to_string()),
4899 "handle_observed_address_frame: RECEIVED OBSERVED_ADDRESS from peer"
4900 );
4901 let state = self.address_discovery_state.as_mut().ok_or_else(|| {
4903 TransportError::PROTOCOL_VIOLATION(
4904 "ObservedAddress frame without address discovery negotiation",
4905 )
4906 })?;
4907
4908 if !state.enabled {
4910 return Err(TransportError::PROTOCOL_VIOLATION(
4911 "ObservedAddress frame received when address discovery is disabled",
4912 ));
4913 }
4914
4915 #[cfg(feature = "trace")]
4917 {
4918 use crate::trace_observed_address_received;
4919 let peer_bytes = self
4920 .peer_id_for_tokens
4921 .as_ref()
4922 .map(|pid| pid.0)
4923 .unwrap_or([0u8; 32]);
4924 trace_observed_address_received!(
4925 &self.event_log,
4926 self.trace_context.trace_id(),
4927 observed_address.address,
4928 0u64, peer_bytes
4930 );
4931 }
4932
4933 let path_id = 0u64; if let Some(&last_seq) = state.last_received_sequence.get(&path_id) {
4941 if observed_address.sequence_number <= last_seq {
4942 trace!(
4943 "Ignoring OBSERVED_ADDRESS frame with stale sequence number {} (last was {})",
4944 observed_address.sequence_number, last_seq
4945 );
4946 return Ok(());
4947 }
4948 }
4949
4950 state
4952 .last_received_sequence
4953 .insert(path_id, observed_address.sequence_number);
4954
4955 let normalized_addr = crate::shared::normalize_socket_addr(observed_address.address);
4958
4959 state.handle_observed_address(normalized_addr, path_id, now);
4961
4962 self.path.update_observed_address(normalized_addr, now);
4964
4965 trace!(
4967 "Received ObservedAddress frame: address={} for path={}",
4968 observed_address.address, path_id
4969 );
4970
4971 Ok(())
4972 }
4973
4974 fn handle_try_connect_to(
4979 &mut self,
4980 try_connect_to: &crate::frame::TryConnectTo,
4981 now: Instant,
4982 ) -> Result<(), TransportError> {
4983 trace!(
4984 "Received TryConnectTo: request_id={}, target={}, timeout_ms={}",
4985 try_connect_to.request_id, try_connect_to.target_address, try_connect_to.timeout_ms
4986 );
4987
4988 let target = try_connect_to.target_address;
4990
4991 let allow_loopback = allow_loopback_from_env();
4993 if target.ip().is_loopback() && !allow_loopback {
4994 warn!(
4995 "Rejecting TryConnectTo request to loopback address: {}",
4996 target
4997 );
4998 let response = crate::frame::TryConnectToResponse {
5000 request_id: try_connect_to.request_id,
5001 success: false,
5002 error_code: Some(crate::frame::TryConnectError::InvalidAddress),
5003 source_address: self.path.remote,
5004 };
5005 self.spaces[SpaceId::Data]
5006 .pending
5007 .try_connect_to_responses
5008 .push(response);
5009 return Ok(());
5010 }
5011
5012 if target.ip().is_unspecified() {
5014 warn!(
5015 "Rejecting TryConnectTo request to unspecified address: {}",
5016 target
5017 );
5018 let response = crate::frame::TryConnectToResponse {
5019 request_id: try_connect_to.request_id,
5020 success: false,
5021 error_code: Some(crate::frame::TryConnectError::InvalidAddress),
5022 source_address: self.path.remote,
5023 };
5024 self.spaces[SpaceId::Data]
5025 .pending
5026 .try_connect_to_responses
5027 .push(response);
5028 return Ok(());
5029 }
5030
5031 self.endpoint_events
5034 .push_back(EndpointEventInner::TryConnectTo {
5035 request_id: try_connect_to.request_id,
5036 target_address: try_connect_to.target_address,
5037 timeout_ms: try_connect_to.timeout_ms,
5038 requester_connection: self.path.remote,
5039 requested_at: now,
5040 });
5041
5042 trace!(
5043 "Queued TryConnectTo attempt for request_id={}",
5044 try_connect_to.request_id
5045 );
5046
5047 Ok(())
5048 }
5049
5050 fn handle_try_connect_to_response(
5052 &mut self,
5053 response: &crate::frame::TryConnectToResponse,
5054 ) -> Result<(), TransportError> {
5055 trace!(
5056 "Received TryConnectToResponse: request_id={}, success={}, error={:?}, source={}",
5057 response.request_id, response.success, response.error_code, response.source_address
5058 );
5059
5060 if response.success {
5063 debug!(
5064 "TryConnectTo succeeded: target can receive connections from {}",
5065 response.source_address
5066 );
5067
5068 if let Some(nat_state) = &mut self.nat_traversal {
5070 nat_state
5071 .record_successful_callback_probe(response.request_id, response.source_address);
5072 }
5073 } else {
5074 debug!("TryConnectTo failed with error: {:?}", response.error_code);
5075
5076 if let Some(nat_state) = &mut self.nat_traversal {
5078 nat_state.record_failed_callback_probe(response.request_id, response.error_code);
5079 }
5080 }
5081
5082 Ok(())
5083 }
5084
5085 pub fn queue_add_address(&mut self, sequence: VarInt, address: SocketAddr, priority: VarInt) {
5087 let add_address = frame::AddAddress {
5089 sequence,
5090 address,
5091 priority,
5092 };
5093
5094 self.spaces[SpaceId::Data]
5095 .pending
5096 .add_addresses
5097 .push(add_address);
5098 trace!(
5099 "Queued AddAddress frame: seq={}, addr={}, priority={}",
5100 sequence, address, priority
5101 );
5102 }
5103
5104 pub fn queue_punch_me_now(
5106 &mut self,
5107 round: VarInt,
5108 paired_with_sequence_number: VarInt,
5109 address: SocketAddr,
5110 ) {
5111 self.queue_punch_me_now_with_target(round, paired_with_sequence_number, address, None);
5112 }
5113
5114 pub fn queue_punch_me_now_with_target(
5126 &mut self,
5127 round: VarInt,
5128 paired_with_sequence_number: VarInt,
5129 address: SocketAddr,
5130 target_peer_id: Option<[u8; 32]>,
5131 ) {
5132 let punch_me_now = frame::PunchMeNow {
5133 round,
5134 paired_with_sequence_number,
5135 address,
5136 target_peer_id,
5137 };
5138
5139 self.spaces[SpaceId::Data]
5140 .pending
5141 .punch_me_now
5142 .push(punch_me_now);
5143
5144 if target_peer_id.is_some() {
5145 trace!(
5146 "Queued PunchMeNow frame for relay: round={}, target_seq={}, target_peer={:?}",
5147 round,
5148 paired_with_sequence_number,
5149 target_peer_id.map(|p| hex::encode(&p[..8]))
5150 );
5151 } else {
5152 trace!(
5153 "Queued PunchMeNow frame: round={}, target={}",
5154 round, paired_with_sequence_number
5155 );
5156 }
5157 }
5158
5159 pub fn queue_remove_address(&mut self, sequence: VarInt) {
5161 let remove_address = frame::RemoveAddress { sequence };
5162
5163 self.spaces[SpaceId::Data]
5164 .pending
5165 .remove_addresses
5166 .push(remove_address);
5167 trace!("Queued RemoveAddress frame: seq={}", sequence);
5168 }
5169
5170 pub fn queue_observed_address(&mut self, address: SocketAddr) {
5172 let sequence_number = if let Some(state) = &mut self.address_discovery_state {
5174 let seq = state.next_sequence_number;
5175 state.next_sequence_number =
5176 VarInt::from_u64(state.next_sequence_number.into_inner() + 1)
5177 .expect("sequence number overflow");
5178 seq
5179 } else {
5180 VarInt::from_u32(0)
5182 };
5183
5184 let observed_address = frame::ObservedAddress {
5185 sequence_number,
5186 address,
5187 };
5188 self.spaces[SpaceId::Data]
5189 .pending
5190 .outbound_observations
5191 .push(observed_address);
5192 trace!("Queued ObservedAddress frame: addr={}", address);
5193 }
5194
5195 pub fn check_for_address_observations(&mut self, now: Instant) {
5197 let Some(state) = &mut self.address_discovery_state else {
5199 return;
5200 };
5201
5202 if !state.enabled {
5204 return;
5205 }
5206
5207 if self.peer_params.address_discovery.is_none() {
5210 return;
5211 }
5212
5213 let path_id = 0u64; let remote_address = self.path.remote;
5218
5219 if state.should_send_observation(path_id, now) {
5221 if let Some(frame) = state.queue_observed_address_frame(path_id, remote_address) {
5223 self.spaces[SpaceId::Data]
5225 .pending
5226 .outbound_observations
5227 .push(frame);
5228
5229 state.record_observation_sent(path_id);
5231
5232 #[cfg(feature = "trace")]
5234 {
5235 use crate::trace_observed_address_sent;
5236 trace_observed_address_sent!(
5238 &self.event_log,
5239 self.trace_context.trace_id(),
5240 remote_address,
5241 path_id
5242 );
5243 }
5244
5245 trace!(
5246 "Queued OBSERVED_ADDRESS frame for path {} with address {}",
5247 path_id, remote_address
5248 );
5249 }
5250 }
5251 }
5252
5253 fn trigger_candidate_validation(
5255 &mut self,
5256 candidate_address: SocketAddr,
5257 now: Instant,
5258 ) -> Result<(), TransportError> {
5259 let nat_state = self
5260 .nat_traversal
5261 .as_mut()
5262 .ok_or_else(|| TransportError::PROTOCOL_VIOLATION("NAT traversal not enabled"))?;
5263
5264 let already_validating = nat_state.active_validations.values().any(|v| {
5267 crate::shared::normalize_socket_addr(v.target_addr)
5268 == crate::shared::normalize_socket_addr(candidate_address)
5269 });
5270 if already_validating {
5271 trace!("Validation already in progress for {}", candidate_address);
5272 return Ok(());
5273 }
5274
5275 let sequence = nat_state
5277 .remote_candidates
5278 .iter()
5279 .find(|(_, c)| {
5280 crate::shared::normalize_socket_addr(c.address)
5281 == crate::shared::normalize_socket_addr(candidate_address)
5282 })
5283 .map(|(seq, _)| *seq)
5284 .unwrap_or(crate::VarInt::from_u32(0));
5285
5286 let challenge = self.rng.r#gen::<u64>();
5288
5289 let validation_state = nat_traversal::PathValidationState {
5291 challenge,
5292 sequence,
5293 target_addr: candidate_address,
5294 sent_at: now,
5295 retry_count: 0,
5296 max_retries: 3,
5297 coordination_round: None,
5298 timeout_state: nat_traversal::AdaptiveTimeoutState::new(),
5299 last_retry_at: None,
5300 };
5301
5302 nat_state
5304 .active_validations
5305 .insert(challenge, validation_state);
5306
5307 nat_state.stats.validations_succeeded += 1; trace!(
5313 "Triggered PATH_CHALLENGE validation for {} with challenge {:016x}",
5314 candidate_address, challenge
5315 );
5316
5317 Ok(())
5318 }
5319
5320 pub fn nat_traversal_state(&self) -> Option<(usize, usize)> {
5325 self.nat_traversal
5326 .as_ref()
5327 .map(|state| (state.local_candidates.len(), state.remote_candidates.len()))
5328 }
5329
5330 pub fn initiate_nat_traversal_coordination(
5332 &mut self,
5333 now: Instant,
5334 ) -> Result<(), TransportError> {
5335 let nat_state = self
5336 .nat_traversal
5337 .as_mut()
5338 .ok_or_else(|| TransportError::PROTOCOL_VIOLATION("NAT traversal not enabled"))?;
5339
5340 if nat_state.should_send_punch_request() {
5342 nat_state.generate_candidate_pairs(now);
5344
5345 let pairs = nat_state.get_next_validation_pairs(3);
5347 if pairs.is_empty() {
5348 return Err(TransportError::PROTOCOL_VIOLATION(
5349 "No candidate pairs for coordination",
5350 ));
5351 }
5352
5353 let targets: Vec<_> = pairs
5355 .into_iter()
5356 .map(|pair| nat_traversal::PunchTarget {
5357 remote_addr: pair.remote_addr,
5358 remote_sequence: pair.remote_sequence,
5359 challenge: self.rng.r#gen(),
5360 })
5361 .collect();
5362
5363 let round = nat_state
5365 .start_coordination_round(targets, now)
5366 .map_err(|_e| {
5367 TransportError::PROTOCOL_VIOLATION("Failed to start coordination round")
5368 })?;
5369
5370 let local_addr = self
5373 .local_ip
5374 .map(|ip| SocketAddr::new(ip, self.local_ip.map(|_| 0).unwrap_or(0)))
5375 .unwrap_or_else(|| {
5376 SocketAddr::new(std::net::IpAddr::V4(std::net::Ipv4Addr::UNSPECIFIED), 0)
5377 });
5378
5379 let punch_me_now = frame::PunchMeNow {
5380 round,
5381 paired_with_sequence_number: VarInt::from_u32(0), address: local_addr,
5383 target_peer_id: None, };
5385
5386 self.spaces[SpaceId::Data]
5387 .pending
5388 .punch_me_now
5389 .push(punch_me_now);
5390 nat_state.mark_punch_request_sent();
5391
5392 trace!("Initiated NAT traversal coordination round {}", round);
5393 }
5394
5395 Ok(())
5396 }
5397
5398 pub fn validate_nat_candidates(&mut self, now: Instant) {
5400 self.generate_nat_traversal_challenges(now);
5401 }
5402
5403 pub fn send_nat_address_advertisement(
5418 &mut self,
5419 address: SocketAddr,
5420 priority: u32,
5421 ) -> Result<u64, ConnectionError> {
5422 let normalized_addr = crate::shared::normalize_socket_addr(address);
5425
5426 if !is_valid_nat_advertisement_address(normalized_addr) {
5427 debug!(
5428 "Skipping NAT address advertisement for invalid candidate {}",
5429 normalized_addr
5430 );
5431 return Err(ConnectionError::TransportError(
5432 TransportError::PROTOCOL_VIOLATION("invalid NAT candidate address"),
5433 ));
5434 }
5435
5436 let nat_state = self.nat_traversal.as_mut().ok_or_else(|| {
5438 ConnectionError::TransportError(TransportError::PROTOCOL_VIOLATION(
5439 "NAT traversal not enabled on this connection",
5440 ))
5441 })?;
5442
5443 let sequence = nat_state.next_sequence;
5445 nat_state.next_sequence =
5446 VarInt::from_u64(nat_state.next_sequence.into_inner() + 1).unwrap();
5447
5448 let now = Instant::now();
5450 nat_state.local_candidates.insert(
5451 sequence,
5452 nat_traversal::AddressCandidate {
5453 address: normalized_addr,
5454 priority,
5455 source: nat_traversal::CandidateSource::Local,
5456 discovered_at: now,
5457 state: nat_traversal::CandidateState::New,
5458 attempt_count: 0,
5459 last_attempt: None,
5460 },
5461 );
5462
5463 nat_state.stats.local_candidates_sent += 1;
5465
5466 self.queue_add_address(sequence, normalized_addr, VarInt::from_u32(priority));
5468
5469 debug!(
5470 "Queued ADD_ADDRESS frame: addr={} (normalized from {}), priority={}, seq={}",
5471 normalized_addr, address, priority, sequence
5472 );
5473 Ok(sequence.into_inner())
5474 }
5475
5476 pub fn send_nat_punch_coordination(
5489 &mut self,
5490 paired_with_sequence_number: u64,
5491 address: SocketAddr,
5492 round: u32,
5493 ) -> Result<(), ConnectionError> {
5494 let _nat_state = self.nat_traversal.as_ref().ok_or_else(|| {
5496 ConnectionError::TransportError(TransportError::PROTOCOL_VIOLATION(
5497 "NAT traversal not enabled on this connection",
5498 ))
5499 })?;
5500
5501 self.queue_punch_me_now(
5503 VarInt::from_u32(round),
5504 VarInt::from_u64(paired_with_sequence_number).map_err(|_| {
5505 ConnectionError::TransportError(TransportError::PROTOCOL_VIOLATION(
5506 "Invalid target sequence number",
5507 ))
5508 })?,
5509 address,
5510 );
5511
5512 debug!(
5513 "Queued PUNCH_ME_NOW frame: paired_with_seq={}, addr={}, round={}",
5514 paired_with_sequence_number, address, round
5515 );
5516 Ok(())
5517 }
5518
5519 pub fn send_nat_punch_via_relay(
5533 &mut self,
5534 target_peer_id: [u8; 32],
5535 our_address: SocketAddr,
5536 round: u32,
5537 ) -> Result<(), ConnectionError> {
5538 let _nat_state = self.nat_traversal.as_ref().ok_or_else(|| {
5540 ConnectionError::TransportError(TransportError::PROTOCOL_VIOLATION(
5541 "NAT traversal not enabled on this connection",
5542 ))
5543 })?;
5544
5545 self.queue_punch_me_now_with_target(
5547 VarInt::from_u32(round),
5548 VarInt::from_u32(0), our_address,
5550 Some(target_peer_id),
5551 );
5552
5553 info!(
5554 "Queued PUNCH_ME_NOW for relay: target_peer={}, our_addr={}, round={}",
5555 hex::encode(&target_peer_id[..8]),
5556 our_address,
5557 round
5558 );
5559 Ok(())
5560 }
5561
5562 pub fn send_nat_address_removal(&mut self, sequence: u64) -> Result<(), ConnectionError> {
5573 let nat_state = self.nat_traversal.as_mut().ok_or_else(|| {
5575 ConnectionError::TransportError(TransportError::PROTOCOL_VIOLATION(
5576 "NAT traversal not enabled on this connection",
5577 ))
5578 })?;
5579
5580 let sequence_varint = VarInt::from_u64(sequence).map_err(|_| {
5581 ConnectionError::TransportError(TransportError::PROTOCOL_VIOLATION(
5582 "Invalid sequence number",
5583 ))
5584 })?;
5585
5586 nat_state.local_candidates.remove(&sequence_varint);
5588
5589 self.queue_remove_address(sequence_varint);
5591
5592 debug!("Queued REMOVE_ADDRESS frame: seq={}", sequence);
5593 Ok(())
5594 }
5595
5596 #[allow(dead_code)]
5605 pub(crate) fn get_nat_traversal_stats(&self) -> Option<&nat_traversal::NatTraversalStats> {
5606 self.nat_traversal.as_ref().map(|state| &state.stats)
5607 }
5608
5609 pub fn is_nat_traversal_enabled(&self) -> bool {
5611 self.nat_traversal.is_some()
5612 }
5613
5614 fn negotiate_address_discovery(&mut self, peer_params: &TransportParameters) {
5618 let now = Instant::now();
5619
5620 info!(
5621 "negotiate_address_discovery: peer_params.address_discovery = {:?}",
5622 peer_params.address_discovery
5623 );
5624
5625 match &peer_params.address_discovery {
5627 Some(peer_config) => {
5628 info!("Peer supports address discovery: {:?}", peer_config);
5630 if let Some(state) = &mut self.address_discovery_state {
5631 if state.enabled {
5632 info!(
5635 "Address discovery negotiated successfully: rate={}, all_paths={}",
5636 state.max_observation_rate, state.observe_all_paths
5637 );
5638 } else {
5639 info!("Address discovery disabled locally, ignoring peer support");
5641 }
5642 } else {
5643 self.address_discovery_state =
5645 Some(AddressDiscoveryState::new(peer_config, now));
5646 info!("Address discovery initialized from peer config");
5647 }
5648 }
5649 _ => {
5650 warn!("Peer does NOT support address discovery (transport parameter not present)");
5652 if let Some(state) = &mut self.address_discovery_state {
5653 state.enabled = false;
5654 }
5655 }
5656 }
5657
5658 if let Some(state) = &self.address_discovery_state {
5660 if state.enabled {
5661 self.path.set_observation_rate(state.max_observation_rate);
5662 }
5663 }
5664 }
5665
5666 fn decrypt_packet(
5667 &mut self,
5668 now: Instant,
5669 packet: &mut Packet,
5670 ) -> Result<Option<u64>, Option<TransportError>> {
5671 let result = packet_crypto::decrypt_packet_body(
5672 packet,
5673 &self.spaces,
5674 self.zero_rtt_crypto.as_ref(),
5675 self.key_phase,
5676 self.prev_crypto.as_ref(),
5677 self.next_crypto.as_ref(),
5678 )?;
5679
5680 let result = match result {
5681 Some(r) => r,
5682 None => return Ok(None),
5683 };
5684
5685 if result.outgoing_key_update_acked {
5686 if let Some(prev) = self.prev_crypto.as_mut() {
5687 prev.end_packet = Some((result.number, now));
5688 self.set_key_discard_timer(now, packet.header.space());
5689 }
5690 }
5691
5692 if result.incoming_key_update {
5693 trace!("key update authenticated");
5694 self.update_keys(Some((result.number, now)), true);
5695 self.set_key_discard_timer(now, packet.header.space());
5696 }
5697
5698 Ok(Some(result.number))
5699 }
5700
5701 fn update_keys(&mut self, end_packet: Option<(u64, Instant)>, remote: bool) {
5702 trace!("executing key update");
5703 let new = self
5707 .crypto
5708 .next_1rtt_keys()
5709 .expect("only called for `Data` packets");
5710 self.key_phase_size = new
5711 .local
5712 .confidentiality_limit()
5713 .saturating_sub(KEY_UPDATE_MARGIN);
5714 let old = mem::replace(
5715 &mut self.spaces[SpaceId::Data]
5716 .crypto
5717 .as_mut()
5718 .unwrap() .packet,
5720 mem::replace(self.next_crypto.as_mut().unwrap(), new),
5721 );
5722 self.spaces[SpaceId::Data].sent_with_keys = 0;
5723 self.prev_crypto = Some(PrevCrypto {
5724 crypto: old,
5725 end_packet,
5726 update_unacked: remote,
5727 });
5728 self.key_phase = !self.key_phase;
5729 }
5730
5731 fn peer_supports_ack_frequency(&self) -> bool {
5732 self.peer_params.min_ack_delay.is_some()
5733 }
5734
5735 pub(crate) fn immediate_ack(&mut self) {
5740 self.spaces[self.highest_space].immediate_ack_pending = true;
5741 }
5742
5743 #[cfg(test)]
5745 #[allow(dead_code)]
5746 pub(crate) fn decode_packet(&self, event: &ConnectionEvent) -> Option<Vec<u8>> {
5747 let (first_decode, remaining) = match &event.0 {
5748 ConnectionEventInner::Datagram(DatagramConnectionEvent {
5749 first_decode,
5750 remaining,
5751 ..
5752 }) => (first_decode, remaining),
5753 _ => return None,
5754 };
5755
5756 if remaining.is_some() {
5757 panic!("Packets should never be coalesced in tests");
5758 }
5759
5760 let decrypted_header = packet_crypto::unprotect_header(
5761 first_decode.clone(),
5762 &self.spaces,
5763 self.zero_rtt_crypto.as_ref(),
5764 self.peer_params.stateless_reset_token,
5765 )?;
5766
5767 let mut packet = decrypted_header.packet?;
5768 packet_crypto::decrypt_packet_body(
5769 &mut packet,
5770 &self.spaces,
5771 self.zero_rtt_crypto.as_ref(),
5772 self.key_phase,
5773 self.prev_crypto.as_ref(),
5774 self.next_crypto.as_ref(),
5775 )
5776 .ok()?;
5777
5778 Some(packet.payload.to_vec())
5779 }
5780
5781 #[cfg(test)]
5784 #[allow(dead_code)]
5785 pub(crate) fn bytes_in_flight(&self) -> u64 {
5786 self.path.in_flight.bytes
5787 }
5788
5789 #[cfg(test)]
5791 #[allow(dead_code)]
5792 pub(crate) fn congestion_window(&self) -> u64 {
5793 self.path
5794 .congestion
5795 .window()
5796 .saturating_sub(self.path.in_flight.bytes)
5797 }
5798
5799 #[cfg(test)]
5801 #[allow(dead_code)]
5802 pub(crate) fn is_idle(&self) -> bool {
5803 Timer::VALUES
5804 .iter()
5805 .filter(|&&t| !matches!(t, Timer::KeepAlive | Timer::PushNewCid | Timer::KeyDiscard))
5806 .filter_map(|&t| Some((t, self.timers.get(t)?)))
5807 .min_by_key(|&(_, time)| time)
5808 .is_none_or(|(timer, _)| timer == Timer::Idle)
5809 }
5810
5811 #[cfg(test)]
5813 #[allow(dead_code)]
5814 pub(crate) fn lost_packets(&self) -> u64 {
5815 self.lost_packets
5816 }
5817
5818 #[cfg(test)]
5820 #[allow(dead_code)]
5821 pub(crate) fn using_ecn(&self) -> bool {
5822 self.path.sending_ecn
5823 }
5824
5825 #[cfg(test)]
5827 #[allow(dead_code)]
5828 pub(crate) fn total_recvd(&self) -> u64 {
5829 self.path.total_recvd
5830 }
5831
5832 #[cfg(test)]
5833 #[allow(dead_code)]
5834 pub(crate) fn active_local_cid_seq(&self) -> (u64, u64) {
5835 self.local_cid_state.active_seq()
5836 }
5837
5838 #[cfg(test)]
5841 #[allow(dead_code)]
5842 pub(crate) fn rotate_local_cid(&mut self, v: u64, now: Instant) {
5843 let n = self.local_cid_state.assign_retire_seq(v);
5844 self.endpoint_events
5845 .push_back(EndpointEventInner::NeedIdentifiers(now, n));
5846 }
5847
5848 #[cfg(test)]
5850 #[allow(dead_code)]
5851 pub(crate) fn active_rem_cid_seq(&self) -> u64 {
5852 self.rem_cids.active_seq()
5853 }
5854
5855 #[cfg(test)]
5857 #[cfg(test)]
5858 #[allow(dead_code)]
5859 pub(crate) fn path_mtu(&self) -> u16 {
5860 self.path.current_mtu()
5861 }
5862
5863 fn can_send_1rtt(&self, max_size: usize) -> bool {
5867 self.streams.can_send_stream_data()
5868 || self.path.challenge_pending
5869 || self
5870 .prev_path
5871 .as_ref()
5872 .is_some_and(|(_, x)| x.challenge_pending)
5873 || !self.path_responses.is_empty()
5874 || self
5875 .datagrams
5876 .outgoing
5877 .front()
5878 .is_some_and(|x| x.size(true) <= max_size)
5879 }
5880
5881 fn remove_in_flight(&mut self, pn: u64, packet: &SentPacket) {
5883 for path in [&mut self.path]
5885 .into_iter()
5886 .chain(self.prev_path.as_mut().map(|(_, data)| data))
5887 {
5888 if path.remove_in_flight(pn, packet) {
5889 return;
5890 }
5891 }
5892 }
5893
5894 fn kill(&mut self, reason: ConnectionError) {
5896 self.close_common();
5897 self.error = Some(reason);
5898 self.state = State::Drained;
5899 self.endpoint_events.push_back(EndpointEventInner::Drained);
5900 }
5901
5902 fn generate_nat_traversal_challenges(&mut self, now: Instant) {
5904 let candidates: Vec<(VarInt, SocketAddr)> = if let Some(nat_state) = &self.nat_traversal {
5906 nat_state
5907 .get_validation_candidates()
5908 .into_iter()
5909 .take(3) .map(|(seq, candidate)| (seq, candidate.address))
5911 .collect()
5912 } else {
5913 return;
5914 };
5915
5916 if candidates.is_empty() {
5917 return;
5918 }
5919
5920 if let Some(nat_state) = &mut self.nat_traversal {
5922 for (seq, address) in candidates {
5923 let challenge: u64 = self.rng.r#gen();
5925
5926 if let Err(e) = nat_state.start_validation(seq, challenge, now) {
5928 debug!("Failed to start validation for candidate {}: {}", seq, e);
5929 continue;
5930 }
5931
5932 trace!(
5934 "Started NAT validation for {} with token {:08x}",
5935 address, challenge
5936 );
5937 }
5938 }
5939 }
5940
5941 pub fn current_mtu(&self) -> u16 {
5945 self.path.current_mtu()
5946 }
5947
5948 fn predict_1rtt_overhead(&self, pn: Option<u64>) -> usize {
5955 let pn_len = match pn {
5956 Some(pn) => PacketNumber::new(
5957 pn,
5958 self.spaces[SpaceId::Data].largest_acked_packet.unwrap_or(0),
5959 )
5960 .len(),
5961 None => 4,
5963 };
5964
5965 1 + self.rem_cids.active().len() + pn_len + self.tag_len_1rtt()
5967 }
5968
5969 fn tag_len_1rtt(&self) -> usize {
5970 let key = match self.spaces[SpaceId::Data].crypto.as_ref() {
5971 Some(crypto) => Some(&*crypto.packet.local),
5972 None => self.zero_rtt_crypto.as_ref().map(|x| &*x.packet),
5973 };
5974 key.map_or(16, |x| x.tag_len())
5978 }
5979
5980 fn on_path_validated(&mut self) {
5982 self.path.validated = true;
5983 let ConnectionSide::Server { server_config } = &self.side else {
5984 return;
5985 };
5986 let new_tokens = &mut self.spaces[SpaceId::Data as usize].pending.new_tokens;
5987 new_tokens.clear();
5988 for _ in 0..server_config.validation_token.sent {
5989 new_tokens.push(self.path.remote);
5990 }
5991 }
5992}
5993
5994impl fmt::Debug for Connection {
5995 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
5996 f.debug_struct("Connection")
5997 .field("handshake_cid", &self.handshake_cid)
5998 .finish()
5999 }
6000}
6001
6002enum ConnectionSide {
6004 Client {
6005 token: Bytes,
6007 token_store: Arc<dyn TokenStore>,
6008 server_name: String,
6009 },
6010 Server {
6011 server_config: Arc<ServerConfig>,
6012 },
6013}
6014
6015impl ConnectionSide {
6016 fn remote_may_migrate(&self) -> bool {
6017 match self {
6018 Self::Server { server_config } => server_config.migration,
6019 Self::Client { .. } => false,
6020 }
6021 }
6022
6023 fn is_client(&self) -> bool {
6024 self.side().is_client()
6025 }
6026
6027 fn is_server(&self) -> bool {
6028 self.side().is_server()
6029 }
6030
6031 fn side(&self) -> Side {
6032 match *self {
6033 Self::Client { .. } => Side::Client,
6034 Self::Server { .. } => Side::Server,
6035 }
6036 }
6037}
6038
6039impl From<SideArgs> for ConnectionSide {
6040 fn from(side: SideArgs) -> Self {
6041 match side {
6042 SideArgs::Client {
6043 token_store,
6044 server_name,
6045 } => Self::Client {
6046 token: token_store.take(&server_name).unwrap_or_default(),
6047 token_store,
6048 server_name,
6049 },
6050 SideArgs::Server {
6051 server_config,
6052 pref_addr_cid: _,
6053 path_validated: _,
6054 } => Self::Server { server_config },
6055 }
6056 }
6057}
6058
6059pub(crate) enum SideArgs {
6061 Client {
6062 token_store: Arc<dyn TokenStore>,
6063 server_name: String,
6064 },
6065 Server {
6066 server_config: Arc<ServerConfig>,
6067 pref_addr_cid: Option<ConnectionId>,
6068 path_validated: bool,
6069 },
6070}
6071
6072impl SideArgs {
6073 pub(crate) fn pref_addr_cid(&self) -> Option<ConnectionId> {
6074 match *self {
6075 Self::Client { .. } => None,
6076 Self::Server { pref_addr_cid, .. } => pref_addr_cid,
6077 }
6078 }
6079
6080 pub(crate) fn path_validated(&self) -> bool {
6081 match *self {
6082 Self::Client { .. } => true,
6083 Self::Server { path_validated, .. } => path_validated,
6084 }
6085 }
6086
6087 pub(crate) fn side(&self) -> Side {
6088 match *self {
6089 Self::Client { .. } => Side::Client,
6090 Self::Server { .. } => Side::Server,
6091 }
6092 }
6093}
6094
6095fn is_valid_nat_advertisement_address(address: SocketAddr) -> bool {
6096 if address.port() == 0 {
6097 return false;
6098 }
6099
6100 match address.ip() {
6101 IpAddr::V4(ipv4) => !ipv4.is_unspecified() && !ipv4.is_broadcast() && !ipv4.is_multicast(),
6102 IpAddr::V6(ipv6) => !ipv6.is_unspecified() && !ipv6.is_multicast(),
6103 }
6104}
6105
6106#[derive(Debug, Error, Clone, PartialEq, Eq)]
6108pub enum ConnectionError {
6109 #[error("peer doesn't implement any supported version")]
6111 VersionMismatch,
6112 #[error(transparent)]
6114 TransportError(#[from] TransportError),
6115 #[error("aborted by peer: {0}")]
6117 ConnectionClosed(frame::ConnectionClose),
6118 #[error("closed by peer: {0}")]
6120 ApplicationClosed(frame::ApplicationClose),
6121 #[error("reset by peer")]
6123 Reset,
6124 #[error("timed out")]
6130 TimedOut,
6131 #[error("closed")]
6133 LocallyClosed,
6134 #[error("CIDs exhausted")]
6138 CidsExhausted,
6139}
6140
6141impl From<Close> for ConnectionError {
6142 fn from(x: Close) -> Self {
6143 match x {
6144 Close::Connection(reason) => Self::ConnectionClosed(reason),
6145 Close::Application(reason) => Self::ApplicationClosed(reason),
6146 }
6147 }
6148}
6149
6150impl From<ConnectionError> for io::Error {
6152 fn from(x: ConnectionError) -> Self {
6153 use ConnectionError::*;
6154 let kind = match x {
6155 TimedOut => io::ErrorKind::TimedOut,
6156 Reset => io::ErrorKind::ConnectionReset,
6157 ApplicationClosed(_) | ConnectionClosed(_) => io::ErrorKind::ConnectionAborted,
6158 TransportError(_) | VersionMismatch | LocallyClosed | CidsExhausted => {
6159 io::ErrorKind::Other
6160 }
6161 };
6162 Self::new(kind, x)
6163 }
6164}
6165
6166#[derive(Clone, Debug)]
6167pub enum State {
6169 Handshake(state::Handshake),
6171 Established,
6173 Closed(state::Closed),
6175 Draining,
6177 Drained,
6179}
6180
6181impl State {
6182 fn closed<R: Into<Close>>(reason: R) -> Self {
6183 Self::Closed(state::Closed {
6184 reason: reason.into(),
6185 })
6186 }
6187
6188 fn is_handshake(&self) -> bool {
6189 matches!(*self, Self::Handshake(_))
6190 }
6191
6192 fn is_established(&self) -> bool {
6193 matches!(*self, Self::Established)
6194 }
6195
6196 fn is_closed(&self) -> bool {
6197 matches!(*self, Self::Closed(_) | Self::Draining | Self::Drained)
6198 }
6199
6200 fn is_drained(&self) -> bool {
6201 matches!(*self, Self::Drained)
6202 }
6203}
6204
6205mod state {
6206 use super::*;
6207
6208 #[derive(Clone, Debug)]
6209 pub struct Handshake {
6210 pub(super) rem_cid_set: bool,
6214 pub(super) expected_token: Bytes,
6218 pub(super) client_hello: Option<Bytes>,
6222 }
6223
6224 #[derive(Clone, Debug)]
6225 pub struct Closed {
6226 pub(super) reason: Close,
6227 }
6228}
6229
6230#[derive(Debug)]
6232pub enum Event {
6233 HandshakeDataReady,
6235 Connected,
6237 ConnectionLost {
6241 reason: ConnectionError,
6243 },
6244 Stream(StreamEvent),
6246 DatagramReceived,
6248 DatagramsUnblocked,
6250 DatagramDropped(DatagramDropStats),
6256}
6257
6258fn instant_saturating_sub(x: Instant, y: Instant) -> Duration {
6259 if x > y { x - y } else { Duration::ZERO }
6260}
6261
6262fn get_max_ack_delay(params: &TransportParameters) -> Duration {
6263 Duration::from_micros(params.max_ack_delay.0 * 1000)
6264}
6265
6266const MAX_BACKOFF_EXPONENT: u32 = 16;
6268
6269const MIN_PACKET_SPACE: usize = MAX_HANDSHAKE_OR_0RTT_HEADER_SIZE + 32;
6277
6278const MAX_HANDSHAKE_OR_0RTT_HEADER_SIZE: usize =
6284 1 + 4 + 1 + MAX_CID_SIZE + 1 + MAX_CID_SIZE + VarInt::from_u32(u16::MAX as u32).size() + 4;
6285
6286const KEY_UPDATE_MARGIN: u64 = 10_000;
6290
6291#[derive(Default)]
6292struct SentFrames {
6293 retransmits: ThinRetransmits,
6294 largest_acked: Option<u64>,
6295 stream_frames: StreamMetaVec,
6296 non_retransmits: bool,
6298 requires_padding: bool,
6299}
6300
6301impl SentFrames {
6302 fn is_ack_only(&self, streams: &StreamsState) -> bool {
6304 self.largest_acked.is_some()
6305 && !self.non_retransmits
6306 && self.stream_frames.is_empty()
6307 && self.retransmits.is_empty(streams)
6308 }
6309}
6310
6311fn negotiate_max_idle_timeout(x: Option<VarInt>, y: Option<VarInt>) -> Option<Duration> {
6319 match (x, y) {
6320 (Some(VarInt(0)) | None, Some(VarInt(0)) | None) => None,
6321 (Some(VarInt(0)) | None, Some(y)) => Some(Duration::from_millis(y.0)),
6322 (Some(x), Some(VarInt(0)) | None) => Some(Duration::from_millis(x.0)),
6323 (Some(x), Some(y)) => Some(Duration::from_millis(cmp::min(x, y).0)),
6324 }
6325}
6326
6327#[derive(Debug, Clone)]
6329pub(crate) struct PqcState {
6330 enabled: bool,
6332 #[allow(dead_code)]
6334 algorithms: Option<crate::transport_parameters::PqcAlgorithms>,
6335 handshake_mtu: u16,
6337 using_pqc: bool,
6339 packet_handler: crate::crypto::pqc::packet_handler::PqcPacketHandler,
6341}
6342
6343#[allow(dead_code)]
6344impl PqcState {
6345 fn new() -> Self {
6346 Self {
6347 enabled: false,
6348 algorithms: None,
6349 handshake_mtu: MIN_INITIAL_SIZE,
6350 using_pqc: false,
6351 packet_handler: crate::crypto::pqc::packet_handler::PqcPacketHandler::new(),
6352 }
6353 }
6354
6355 fn min_initial_size(&self) -> u16 {
6357 if self.enabled && self.using_pqc {
6358 std::cmp::max(self.handshake_mtu, 4096)
6360 } else {
6361 MIN_INITIAL_SIZE
6362 }
6363 }
6364
6365 fn update_from_peer_params(&mut self, params: &TransportParameters) {
6367 if let Some(ref algorithms) = params.pqc_algorithms {
6368 self.enabled = true;
6369 self.algorithms = Some(algorithms.clone());
6370 if algorithms.ml_kem_768 || algorithms.ml_dsa_65 {
6372 self.using_pqc = true;
6373 self.handshake_mtu = 4096; }
6375 }
6376 }
6377
6378 fn detect_pqc_from_crypto(&mut self, crypto_data: &[u8], space: SpaceId) {
6380 if !self.enabled {
6381 return;
6382 }
6383 if self.packet_handler.detect_pqc_handshake(crypto_data, space) {
6384 self.using_pqc = true;
6385 self.handshake_mtu = self.packet_handler.get_min_packet_size(space);
6387 }
6388 }
6389
6390 fn should_trigger_mtu_discovery(&mut self) -> bool {
6392 self.packet_handler.should_trigger_mtu_discovery()
6393 }
6394
6395 fn get_mtu_config(&self) -> MtuDiscoveryConfig {
6397 self.packet_handler.get_pqc_mtu_config()
6398 }
6399
6400 fn calculate_crypto_frame_size(&self, available_space: usize, remaining_data: usize) -> usize {
6402 self.packet_handler
6403 .calculate_crypto_frame_size(available_space, remaining_data)
6404 }
6405
6406 fn should_adjust_coalescing(&self, current_size: usize, space: SpaceId) -> bool {
6408 self.packet_handler
6409 .adjust_coalescing_for_pqc(current_size, space)
6410 }
6411
6412 fn on_packet_sent(&mut self, space: SpaceId, size: u16) {
6414 self.packet_handler.on_packet_sent(space, size);
6415 }
6416
6417 fn reset(&mut self) {
6419 self.enabled = false;
6420 self.algorithms = None;
6421 self.handshake_mtu = MIN_INITIAL_SIZE;
6422 self.using_pqc = false;
6423 self.packet_handler.reset();
6424 }
6425}
6426
6427impl Default for PqcState {
6428 fn default() -> Self {
6429 Self::new()
6430 }
6431}
6432
6433#[derive(Debug, Clone)]
6435pub(crate) struct AddressDiscoveryState {
6436 enabled: bool,
6438 max_observation_rate: u8,
6440 observe_all_paths: bool,
6442 sent_observations: std::collections::HashMap<u64, paths::PathAddressInfo>,
6444 received_observations: std::collections::HashMap<u64, paths::PathAddressInfo>,
6446 rate_limiter: AddressObservationRateLimiter,
6448 received_history: Vec<ObservedAddressEvent>,
6450 bootstrap_mode: bool,
6452 next_sequence_number: VarInt,
6454 last_received_sequence: std::collections::HashMap<u64, VarInt>,
6456 frames_sent: u64,
6458}
6459
6460#[derive(Debug, Clone, PartialEq, Eq)]
6462struct ObservedAddressEvent {
6463 address: SocketAddr,
6465 received_at: Instant,
6467 path_id: u64,
6469}
6470
6471#[derive(Debug, Clone)]
6473struct AddressObservationRateLimiter {
6474 tokens: f64,
6476 max_tokens: f64,
6478 rate: f64,
6480 last_update: Instant,
6482}
6483
6484#[allow(dead_code)]
6485impl AddressDiscoveryState {
6486 fn new(config: &crate::transport_parameters::AddressDiscoveryConfig, now: Instant) -> Self {
6488 use crate::transport_parameters::AddressDiscoveryConfig::*;
6489
6490 let (enabled, _can_send, _can_receive) = match config {
6492 SendOnly => (true, true, false),
6493 ReceiveOnly => (true, false, true),
6494 SendAndReceive => (true, true, true),
6495 };
6496
6497 let max_observation_rate = 10u8; let observe_all_paths = false; Self {
6503 enabled,
6504 max_observation_rate,
6505 observe_all_paths,
6506 sent_observations: std::collections::HashMap::new(),
6507 received_observations: std::collections::HashMap::new(),
6508 rate_limiter: AddressObservationRateLimiter::new(max_observation_rate, now),
6509 received_history: Vec::new(),
6510 bootstrap_mode: false,
6511 next_sequence_number: VarInt::from_u32(0),
6512 last_received_sequence: std::collections::HashMap::new(),
6513 frames_sent: 0,
6514 }
6515 }
6516
6517 fn should_send_observation(&mut self, path_id: u64, now: Instant) -> bool {
6519 if !self.should_observe_path(path_id) {
6521 return false;
6522 }
6523
6524 let needs_observation = match self.sent_observations.get(&path_id) {
6526 Some(info) => info.observed_address.is_none() || !info.notified,
6527 None => true,
6528 };
6529
6530 if !needs_observation {
6531 return false;
6532 }
6533
6534 self.rate_limiter.try_consume(1.0, now)
6536 }
6537
6538 fn record_observation_sent(&mut self, path_id: u64) {
6540 if let Some(info) = self.sent_observations.get_mut(&path_id) {
6541 info.mark_notified();
6542 }
6543 }
6544
6545 fn handle_observed_address(&mut self, address: SocketAddr, path_id: u64, now: Instant) {
6547 if !self.enabled {
6548 return;
6549 }
6550
6551 self.received_history.push(ObservedAddressEvent {
6552 address,
6553 received_at: now,
6554 path_id,
6555 });
6556
6557 let info = self
6559 .received_observations
6560 .entry(path_id)
6561 .or_insert_with(paths::PathAddressInfo::new);
6562 info.update_observed_address(address, now);
6563 }
6564
6565 pub(crate) fn get_observed_address(&self, path_id: u64) -> Option<SocketAddr> {
6567 self.received_observations
6568 .get(&path_id)
6569 .and_then(|info| info.observed_address)
6570 }
6571
6572 pub(crate) fn get_all_received_history(&self) -> Vec<SocketAddr> {
6574 self.received_observations
6575 .values()
6576 .filter_map(|info| info.observed_address)
6577 .collect()
6578 }
6579
6580 pub(crate) fn stats(&self) -> AddressDiscoveryStats {
6582 AddressDiscoveryStats {
6583 frames_sent: self.frames_sent,
6584 frames_received: self.received_history.len() as u64,
6585 addresses_discovered: self
6586 .received_observations
6587 .values()
6588 .filter(|info| info.observed_address.is_some())
6589 .count() as u64,
6590 address_changes_detected: 0, }
6592 }
6593
6594 fn has_unnotified_changes(&self) -> bool {
6600 let has_unsent = self
6602 .sent_observations
6603 .values()
6604 .any(|info| info.observed_address.is_some() && !info.notified);
6605
6606 let has_unreceived = self
6608 .received_observations
6609 .values()
6610 .any(|info| info.observed_address.is_some() && !info.notified);
6611
6612 has_unsent || has_unreceived
6613 }
6614
6615 fn queue_observed_address_frame(
6617 &mut self,
6618 path_id: u64,
6619 address: SocketAddr,
6620 ) -> Option<frame::ObservedAddress> {
6621 if !self.enabled {
6623 tracing::debug!("queue_observed_address_frame: BLOCKED - address discovery disabled");
6624 return None;
6625 }
6626
6627 if !self.observe_all_paths && path_id != 0 {
6629 tracing::debug!(
6630 "queue_observed_address_frame: BLOCKED - path {} not allowed (observe_all_paths={})",
6631 path_id,
6632 self.observe_all_paths
6633 );
6634 return None;
6635 }
6636
6637 if let Some(info) = self.sent_observations.get(&path_id) {
6639 if info.notified {
6640 tracing::trace!(
6641 "queue_observed_address_frame: BLOCKED - path {} already notified",
6642 path_id
6643 );
6644 return None;
6645 }
6646 }
6647
6648 if self.rate_limiter.tokens < 1.0 {
6650 tracing::debug!(
6651 "queue_observed_address_frame: BLOCKED - rate limited (tokens={})",
6652 self.rate_limiter.tokens
6653 );
6654 return None;
6655 }
6656
6657 tracing::info!(
6658 "queue_observed_address_frame: SENDING OBSERVED_ADDRESS to {} for path {}",
6659 address,
6660 path_id
6661 );
6662
6663 self.rate_limiter.tokens -= 1.0;
6665
6666 let info = self
6668 .sent_observations
6669 .entry(path_id)
6670 .or_insert_with(paths::PathAddressInfo::new);
6671 info.observed_address = Some(address);
6672 info.notified = true;
6673
6674 tracing::trace!(
6675 path_id = ?path_id,
6676 address = %address,
6677 "queue_observed_address_frame: queuing frame"
6678 );
6679
6680 let sequence_number = self.next_sequence_number;
6682 self.next_sequence_number = VarInt::from_u64(self.next_sequence_number.into_inner() + 1)
6683 .expect("sequence number overflow");
6684
6685 Some(frame::ObservedAddress {
6686 sequence_number,
6687 address,
6688 })
6689 }
6690
6691 fn check_for_address_observations(
6693 &mut self,
6694 _current_path: u64,
6695 peer_supports_address_discovery: bool,
6696 now: Instant,
6697 ) -> Vec<frame::ObservedAddress> {
6698 let mut frames = Vec::new();
6699
6700 if !self.enabled || !peer_supports_address_discovery {
6702 return frames;
6703 }
6704
6705 self.rate_limiter.update_tokens(now);
6707
6708 let paths_to_notify: Vec<u64> = self
6710 .sent_observations
6711 .iter()
6712 .filter_map(|(&path_id, info)| {
6713 if info.observed_address.is_some() && !info.notified {
6714 Some(path_id)
6715 } else {
6716 None
6717 }
6718 })
6719 .collect();
6720
6721 for path_id in paths_to_notify {
6723 if !self.should_observe_path(path_id) {
6725 continue;
6726 }
6727
6728 if !self.bootstrap_mode && self.rate_limiter.tokens < 1.0 {
6730 break; }
6732
6733 if let Some(info) = self.sent_observations.get_mut(&path_id) {
6735 if let Some(address) = info.observed_address {
6736 if self.bootstrap_mode {
6738 self.rate_limiter.tokens -= 0.2; } else {
6740 self.rate_limiter.tokens -= 1.0;
6741 }
6742
6743 info.notified = true;
6745
6746 let sequence_number = self.next_sequence_number;
6748 self.next_sequence_number =
6749 VarInt::from_u64(self.next_sequence_number.into_inner() + 1)
6750 .expect("sequence number overflow");
6751
6752 self.frames_sent += 1;
6753
6754 frames.push(frame::ObservedAddress {
6755 sequence_number,
6756 address,
6757 });
6758 }
6759 }
6760 }
6761
6762 frames
6763 }
6764
6765 fn update_rate_limit(&mut self, new_rate: f64) {
6767 self.max_observation_rate = new_rate as u8;
6768 self.rate_limiter.set_rate(new_rate as u8);
6769 }
6770
6771 fn from_transport_params(params: &TransportParameters) -> Option<Self> {
6773 params
6774 .address_discovery
6775 .as_ref()
6776 .map(|config| Self::new(config, Instant::now()))
6777 }
6778
6779 #[cfg(test)]
6781 fn new_with_params(enabled: bool, max_rate: f64, observe_all_paths: bool) -> Self {
6782 if !enabled {
6784 return Self {
6786 enabled: false,
6787 max_observation_rate: max_rate as u8,
6788 observe_all_paths,
6789 sent_observations: std::collections::HashMap::new(),
6790 received_observations: std::collections::HashMap::new(),
6791 rate_limiter: AddressObservationRateLimiter::new(max_rate as u8, Instant::now()),
6792 received_history: Vec::new(),
6793 bootstrap_mode: false,
6794 next_sequence_number: VarInt::from_u32(0),
6795 last_received_sequence: std::collections::HashMap::new(),
6796 frames_sent: 0,
6797 };
6798 }
6799
6800 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
6802 let mut state = Self::new(&config, Instant::now());
6803 state.max_observation_rate = max_rate as u8;
6804 state.observe_all_paths = observe_all_paths;
6805 state.rate_limiter = AddressObservationRateLimiter::new(max_rate as u8, Instant::now());
6806 state
6807 }
6808
6809 fn set_bootstrap_mode(&mut self, enabled: bool) {
6811 self.bootstrap_mode = enabled;
6812 if enabled {
6814 let bootstrap_rate = self.get_effective_rate_limit();
6815 self.rate_limiter.rate = bootstrap_rate;
6816 self.rate_limiter.max_tokens = bootstrap_rate * 2.0; self.rate_limiter.tokens = self.rate_limiter.max_tokens;
6819 }
6820 }
6821
6822 fn is_bootstrap_mode(&self) -> bool {
6824 self.bootstrap_mode
6825 }
6826
6827 fn get_effective_rate_limit(&self) -> f64 {
6829 if self.bootstrap_mode {
6830 (self.max_observation_rate as f64) * 5.0
6832 } else {
6833 self.max_observation_rate as f64
6834 }
6835 }
6836
6837 fn should_observe_path(&self, path_id: u64) -> bool {
6839 if !self.enabled {
6840 return false;
6841 }
6842
6843 if self.bootstrap_mode {
6845 return true;
6846 }
6847
6848 self.observe_all_paths || path_id == 0
6850 }
6851
6852 fn should_send_observation_immediately(&self, is_new_connection: bool) -> bool {
6854 self.bootstrap_mode && is_new_connection
6855 }
6856}
6857
6858#[allow(dead_code)]
6859impl AddressObservationRateLimiter {
6860 fn new(rate: u8, now: Instant) -> Self {
6862 let rate_f64 = rate as f64;
6863 Self {
6864 tokens: rate_f64,
6865 max_tokens: rate_f64,
6866 rate: rate_f64,
6867 last_update: now,
6868 }
6869 }
6870
6871 fn try_consume(&mut self, tokens: f64, now: Instant) -> bool {
6873 self.update_tokens(now);
6874
6875 if self.tokens >= tokens {
6876 self.tokens -= tokens;
6877 true
6878 } else {
6879 false
6880 }
6881 }
6882
6883 fn update_tokens(&mut self, now: Instant) {
6885 let elapsed = now.saturating_duration_since(self.last_update);
6886 let new_tokens = elapsed.as_secs_f64() * self.rate;
6887 self.tokens = (self.tokens + new_tokens).min(self.max_tokens);
6888 self.last_update = now;
6889 }
6890
6891 fn set_rate(&mut self, rate: u8) {
6893 let rate_f64 = rate as f64;
6894 self.rate = rate_f64;
6895 self.max_tokens = rate_f64;
6896 if self.tokens > self.max_tokens {
6898 self.tokens = self.max_tokens;
6899 }
6900 }
6901}
6902
6903#[doc(hidden)]
6905pub fn address_discovery_burst_admissions_for_test(attempts: usize) -> usize {
6906 let now = Instant::now();
6907 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
6908 let mut state = AddressDiscoveryState::new(&config, now);
6909 state.observe_all_paths = true;
6910
6911 (0..attempts)
6912 .filter(|attempt| {
6913 let octet = ((*attempt % 200) + 1) as u8;
6914 let port = 40000u16 + (*attempt % 1000) as u16;
6915 let address = SocketAddr::from(([93, 184, 216, octet], port));
6916 state
6917 .queue_observed_address_frame(*attempt as u64, address)
6918 .is_some()
6919 })
6920 .count()
6921}
6922
6923impl Connection {
6924 pub(crate) fn supports_ack_receive_v2(&self) -> bool {
6925 self.peer_params.ack_receive_v2
6926 }
6927}
6928
6929#[cfg(test)]
6930mod tests {
6931 use super::*;
6932 use crate::transport_parameters::AddressDiscoveryConfig;
6933 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
6934
6935 #[test]
6936 fn nat_advertisement_address_validation_rejects_unspecified_and_zero_port() {
6937 assert!(!is_valid_nat_advertisement_address(SocketAddr::new(
6938 IpAddr::V6(Ipv6Addr::UNSPECIFIED),
6939 5000,
6940 )));
6941 assert!(!is_valid_nat_advertisement_address(SocketAddr::new(
6942 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 10)),
6943 0,
6944 )));
6945 assert!(is_valid_nat_advertisement_address(SocketAddr::new(
6946 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 10)),
6947 5000,
6948 )));
6949 }
6950
6951 #[test]
6952 fn address_discovery_state_new() {
6953 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
6954 let now = Instant::now();
6955 let state = AddressDiscoveryState::new(&config, now);
6956
6957 assert!(state.enabled);
6958 assert_eq!(state.max_observation_rate, 10);
6959 assert!(!state.observe_all_paths);
6960 assert!(state.sent_observations.is_empty());
6961 assert!(state.received_observations.is_empty());
6962 assert!(state.received_history.is_empty());
6963 assert_eq!(state.rate_limiter.tokens, 10.0);
6964 }
6965
6966 #[test]
6967 fn address_discovery_state_disabled() {
6968 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
6969 let now = Instant::now();
6970 let mut state = AddressDiscoveryState::new(&config, now);
6971
6972 state.enabled = false;
6974
6975 assert!(!state.should_send_observation(0, now));
6977 }
6978
6979 #[test]
6980 fn address_discovery_state_should_send_observation() {
6981 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
6982 let now = Instant::now();
6983 let mut state = AddressDiscoveryState::new(&config, now);
6984
6985 assert!(state.should_send_observation(0, now));
6987
6988 let mut path_info = paths::PathAddressInfo::new();
6990 path_info.update_observed_address(
6991 SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080),
6992 now,
6993 );
6994 path_info.mark_notified();
6995 state.sent_observations.insert(0, path_info);
6996
6997 assert!(!state.should_send_observation(0, now));
6999
7000 assert!(!state.should_send_observation(1, now));
7002 }
7003
7004 #[test]
7005 fn address_discovery_state_rate_limiting() {
7006 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7007 let now = Instant::now();
7008 let mut state = AddressDiscoveryState::new(&config, now);
7009
7010 state.observe_all_paths = true;
7012
7013 assert!(state.should_send_observation(0, now));
7015
7016 state.rate_limiter.try_consume(9.0, now); assert!(!state.should_send_observation(0, now));
7021
7022 let later = now + Duration::from_secs(1);
7024 state.rate_limiter.update_tokens(later);
7025 assert!(state.should_send_observation(0, later));
7026 }
7027
7028 #[test]
7029 fn address_discovery_state_handle_observed_address() {
7030 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7031 let now = Instant::now();
7032 let mut state = AddressDiscoveryState::new(&config, now);
7033
7034 let addr1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 443);
7035 let addr2 = SocketAddr::new(
7036 IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
7037 8080,
7038 );
7039
7040 state.handle_observed_address(addr1, 0, now);
7042 assert_eq!(state.received_history.len(), 1);
7043 assert_eq!(state.received_history[0].address, addr1);
7044 assert_eq!(state.received_history[0].path_id, 0);
7045
7046 let later = now + Duration::from_millis(100);
7048 state.handle_observed_address(addr2, 1, later);
7049 assert_eq!(state.received_history.len(), 2);
7050 assert_eq!(state.received_history[1].address, addr2);
7051 assert_eq!(state.received_history[1].path_id, 1);
7052 }
7053
7054 #[test]
7055 fn address_discovery_state_get_observed_address() {
7056 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7057 let now = Instant::now();
7058 let mut state = AddressDiscoveryState::new(&config, now);
7059
7060 assert_eq!(state.get_observed_address(0), None);
7062
7063 let mut path_info = paths::PathAddressInfo::new();
7065 let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 80);
7066 path_info.update_observed_address(addr, now);
7067 state.received_observations.insert(0, path_info);
7068
7069 assert_eq!(state.get_observed_address(0), Some(addr));
7071 assert_eq!(state.get_observed_address(1), None);
7072 }
7073
7074 #[test]
7075 fn address_discovery_state_unnotified_changes() {
7076 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7077 let now = Instant::now();
7078 let mut state = AddressDiscoveryState::new(&config, now);
7079
7080 assert!(!state.has_unnotified_changes());
7082
7083 let mut path_info = paths::PathAddressInfo::new();
7085 path_info.update_observed_address(
7086 SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080),
7087 now,
7088 );
7089 state.sent_observations.insert(0, path_info);
7090
7091 assert!(state.has_unnotified_changes());
7093
7094 state.record_observation_sent(0);
7096 assert!(!state.has_unnotified_changes());
7097 }
7098
7099 #[test]
7100 fn address_observation_rate_limiter_token_bucket() {
7101 let now = Instant::now();
7102 let mut limiter = AddressObservationRateLimiter::new(5, now); assert_eq!(limiter.tokens, 5.0);
7106 assert_eq!(limiter.max_tokens, 5.0);
7107 assert_eq!(limiter.rate, 5.0);
7108
7109 assert!(limiter.try_consume(3.0, now));
7111 assert_eq!(limiter.tokens, 2.0);
7112
7113 assert!(!limiter.try_consume(3.0, now));
7115 assert_eq!(limiter.tokens, 2.0);
7116
7117 let later = now + Duration::from_secs(1);
7119 limiter.update_tokens(later);
7120 assert_eq!(limiter.tokens, 5.0); let half_sec = now + Duration::from_millis(500);
7124 let mut limiter2 = AddressObservationRateLimiter::new(5, now);
7125 limiter2.try_consume(3.0, now);
7126 limiter2.update_tokens(half_sec);
7127 assert_eq!(limiter2.tokens, 4.5); }
7129
7130 #[test]
7132 fn connection_initializes_address_discovery_state_default() {
7133 let config = crate::transport_parameters::AddressDiscoveryConfig::default();
7136 let state = AddressDiscoveryState::new(&config, Instant::now());
7137 assert!(state.enabled); assert_eq!(state.max_observation_rate, 10); assert!(!state.observe_all_paths);
7140 }
7141
7142 #[test]
7143 fn connection_initializes_with_address_discovery_enabled() {
7144 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7146 let state = AddressDiscoveryState::new(&config, Instant::now());
7147 assert!(state.enabled);
7148 assert_eq!(state.max_observation_rate, 10);
7149 assert!(!state.observe_all_paths);
7150 }
7151
7152 #[test]
7153 fn connection_address_discovery_enabled_by_default() {
7154 let config = crate::transport_parameters::AddressDiscoveryConfig::default();
7156 let state = AddressDiscoveryState::new(&config, Instant::now());
7157 assert!(state.enabled); }
7159
7160 #[test]
7161 fn negotiate_max_idle_timeout_commutative() {
7162 let test_params = [
7163 (None, None, None),
7164 (None, Some(VarInt(0)), None),
7165 (None, Some(VarInt(2)), Some(Duration::from_millis(2))),
7166 (Some(VarInt(0)), Some(VarInt(0)), None),
7167 (
7168 Some(VarInt(2)),
7169 Some(VarInt(0)),
7170 Some(Duration::from_millis(2)),
7171 ),
7172 (
7173 Some(VarInt(1)),
7174 Some(VarInt(4)),
7175 Some(Duration::from_millis(1)),
7176 ),
7177 ];
7178
7179 for (left, right, result) in test_params {
7180 assert_eq!(negotiate_max_idle_timeout(left, right), result);
7181 assert_eq!(negotiate_max_idle_timeout(right, left), result);
7182 }
7183 }
7184
7185 #[test]
7186 fn path_creation_initializes_address_discovery() {
7187 let config = TransportConfig::default();
7188 let remote = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7189 let now = Instant::now();
7190
7191 let path = paths::PathData::new(remote, false, None, now, &config);
7193
7194 assert!(path.address_info.observed_address.is_none());
7196 assert!(path.address_info.last_observed.is_none());
7197 assert_eq!(path.address_info.observation_count, 0);
7198 assert!(!path.address_info.notified);
7199
7200 assert_eq!(path.observation_rate_limiter.rate, 10.0);
7202 assert_eq!(path.observation_rate_limiter.max_tokens, 10.0);
7203 assert_eq!(path.observation_rate_limiter.tokens, 10.0);
7204 }
7205
7206 #[test]
7207 fn path_migration_resets_address_discovery() {
7208 let config = TransportConfig::default();
7209 let remote1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7210 let remote2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 443);
7211 let now = Instant::now();
7212
7213 let mut path1 = paths::PathData::new(remote1, false, None, now, &config);
7215 path1.update_observed_address(remote1, now);
7216 path1.mark_address_notified();
7217 path1.consume_observation_token(now);
7218 path1.set_observation_rate(20);
7219
7220 let path2 = paths::PathData::from_previous(remote2, &path1, now);
7222
7223 assert!(path2.address_info.observed_address.is_none());
7225 assert!(path2.address_info.last_observed.is_none());
7226 assert_eq!(path2.address_info.observation_count, 0);
7227 assert!(!path2.address_info.notified);
7228
7229 assert_eq!(path2.observation_rate_limiter.rate, 20.0);
7231 assert_eq!(path2.observation_rate_limiter.tokens, 20.0);
7232 }
7233
7234 #[test]
7235 fn connection_path_updates_observation_rate() {
7236 let config = TransportConfig::default();
7237 let remote = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)), 42);
7238 let now = Instant::now();
7239
7240 let mut path = paths::PathData::new(remote, false, None, now, &config);
7241
7242 assert_eq!(path.observation_rate_limiter.rate, 10.0);
7244
7245 path.set_observation_rate(25);
7247 assert_eq!(path.observation_rate_limiter.rate, 25.0);
7248 assert_eq!(path.observation_rate_limiter.max_tokens, 25.0);
7249
7250 path.observation_rate_limiter.tokens = 30.0; path.set_observation_rate(20);
7253 assert_eq!(path.observation_rate_limiter.tokens, 20.0); }
7255
7256 #[test]
7257 fn path_validation_preserves_discovery_state() {
7258 let config = TransportConfig::default();
7259 let remote = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7260 let now = Instant::now();
7261
7262 let mut path = paths::PathData::new(remote, false, None, now, &config);
7263
7264 let observed = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(1, 2, 3, 4)), 5678);
7266 path.update_observed_address(observed, now);
7267 path.set_observation_rate(15);
7268
7269 path.validated = true;
7271
7272 assert_eq!(path.address_info.observed_address, Some(observed));
7274 assert_eq!(path.observation_rate_limiter.rate, 15.0);
7275 }
7276
7277 #[test]
7278 fn address_discovery_state_initialization() {
7279 let state = AddressDiscoveryState::new_with_params(true, 30.0, true);
7281
7282 assert!(state.enabled);
7283 assert_eq!(state.max_observation_rate, 30);
7284 assert!(state.observe_all_paths);
7285 assert!(state.sent_observations.is_empty());
7286 assert!(state.received_observations.is_empty());
7287 assert!(state.received_history.is_empty());
7288 }
7289
7290 #[test]
7292 fn handle_observed_address_frame_basic() {
7293 let config = AddressDiscoveryConfig::SendAndReceive;
7294 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7295 let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7296 let now = Instant::now();
7297 let path_id = 0;
7298
7299 state.handle_observed_address(addr, path_id, now);
7301
7302 assert_eq!(state.received_history.len(), 1);
7304 assert_eq!(state.received_history[0].address, addr);
7305 assert_eq!(state.received_history[0].path_id, path_id);
7306 assert_eq!(state.received_history[0].received_at, now);
7307
7308 assert!(state.received_observations.contains_key(&path_id));
7310 let path_info = &state.received_observations[&path_id];
7311 assert_eq!(path_info.observed_address, Some(addr));
7312 assert_eq!(path_info.last_observed, Some(now));
7313 assert_eq!(path_info.observation_count, 1);
7314 }
7315
7316 #[test]
7317 fn handle_observed_address_frame_multiple_observations() {
7318 let config = AddressDiscoveryConfig::SendAndReceive;
7319 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7320 let addr1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7321 let addr2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1)), 443);
7322 let now = Instant::now();
7323 let path_id = 0;
7324
7325 state.handle_observed_address(addr1, path_id, now);
7327 state.handle_observed_address(addr1, path_id, now + Duration::from_secs(1));
7328 state.handle_observed_address(addr2, path_id, now + Duration::from_secs(2));
7329
7330 assert_eq!(state.received_history.len(), 3);
7332
7333 let path_info = &state.received_observations[&path_id];
7335 assert_eq!(path_info.observed_address, Some(addr2));
7336 assert_eq!(path_info.observation_count, 1); }
7338
7339 #[test]
7340 fn handle_observed_address_frame_disabled() {
7341 let config = AddressDiscoveryConfig::SendAndReceive;
7342 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7343 state.enabled = false; let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7345 let now = Instant::now();
7346
7347 state.handle_observed_address(addr, 0, now);
7349
7350 assert!(state.received_history.is_empty());
7352 assert!(state.sent_observations.is_empty());
7353 assert!(state.received_observations.is_empty());
7354 }
7355
7356 #[test]
7357 fn should_send_observation_basic() {
7358 let config = AddressDiscoveryConfig::SendAndReceive;
7359 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7360 state.max_observation_rate = 10;
7361 let now = Instant::now();
7362 let path_id = 0;
7363
7364 assert!(state.should_send_observation(path_id, now));
7366
7367 state.record_observation_sent(path_id);
7369
7370 assert!(state.should_send_observation(path_id, now));
7372 }
7373
7374 #[test]
7375 fn should_send_observation_rate_limiting() {
7376 let config = AddressDiscoveryConfig::SendAndReceive;
7377 let now = Instant::now();
7378 let mut state = AddressDiscoveryState::new(&config, now);
7379 state.max_observation_rate = 2; state.update_rate_limit(2.0);
7381 let path_id = 0;
7382
7383 assert!(state.should_send_observation(path_id, now));
7385 state.record_observation_sent(path_id);
7386 assert!(state.should_send_observation(path_id, now));
7387 state.record_observation_sent(path_id);
7388
7389 assert!(!state.should_send_observation(path_id, now));
7391
7392 let later = now + Duration::from_secs(1);
7394 assert!(state.should_send_observation(path_id, later));
7395 }
7396
7397 #[test]
7398 fn should_send_observation_disabled() {
7399 let config = AddressDiscoveryConfig::SendAndReceive;
7400 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7401 state.enabled = false;
7402
7403 assert!(!state.should_send_observation(0, Instant::now()));
7405 }
7406
7407 #[test]
7408 fn should_send_observation_per_path() {
7409 let config = AddressDiscoveryConfig::SendAndReceive;
7410 let now = Instant::now();
7411 let mut state = AddressDiscoveryState::new(&config, now);
7412 state.max_observation_rate = 2; state.observe_all_paths = true;
7414 state.update_rate_limit(2.0);
7415
7416 assert!(state.should_send_observation(0, now));
7418 state.record_observation_sent(0);
7419
7420 assert!(state.should_send_observation(1, now));
7422 state.record_observation_sent(1);
7423
7424 assert!(!state.should_send_observation(0, now));
7426 assert!(!state.should_send_observation(1, now));
7427
7428 let later = now + Duration::from_secs(1);
7430 assert!(state.should_send_observation(0, later));
7431 }
7432
7433 #[test]
7434 fn has_unnotified_changes_test() {
7435 let config = AddressDiscoveryConfig::SendAndReceive;
7436 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7437 let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7438 let now = Instant::now();
7439
7440 assert!(!state.has_unnotified_changes());
7442
7443 state.handle_observed_address(addr, 0, now);
7445 assert!(state.has_unnotified_changes());
7446
7447 state.received_observations.get_mut(&0).unwrap().notified = true;
7449 assert!(!state.has_unnotified_changes());
7450 }
7451
7452 #[test]
7453 fn get_observed_address_test() {
7454 let config = AddressDiscoveryConfig::SendAndReceive;
7455 let mut state = AddressDiscoveryState::new(&config, Instant::now());
7456 let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7457 let now = Instant::now();
7458 let path_id = 0;
7459
7460 assert_eq!(state.get_observed_address(path_id), None);
7462
7463 state.handle_observed_address(addr, path_id, now);
7465 assert_eq!(state.get_observed_address(path_id), Some(addr));
7466
7467 assert_eq!(state.get_observed_address(999), None);
7469 }
7470
7471 #[test]
7473 fn rate_limiter_token_bucket_basic() {
7474 let now = Instant::now();
7475 let mut limiter = AddressObservationRateLimiter::new(10, now); assert!(limiter.try_consume(5.0, now));
7479 assert!(limiter.try_consume(5.0, now));
7480
7481 assert!(!limiter.try_consume(1.0, now));
7483 }
7484
7485 #[test]
7486 fn rate_limiter_token_replenishment() {
7487 let now = Instant::now();
7488 let mut limiter = AddressObservationRateLimiter::new(10, now); assert!(limiter.try_consume(10.0, now));
7492 assert!(!limiter.try_consume(0.1, now)); let later = now + Duration::from_secs(1);
7496 assert!(limiter.try_consume(10.0, later)); assert!(!limiter.try_consume(0.1, later)); let later = later + Duration::from_millis(500);
7501 assert!(limiter.try_consume(5.0, later)); assert!(!limiter.try_consume(0.1, later)); }
7504
7505 #[test]
7506 fn rate_limiter_max_tokens_cap() {
7507 let now = Instant::now();
7508 let mut limiter = AddressObservationRateLimiter::new(10, now);
7509
7510 let later = now + Duration::from_secs(2);
7512 assert!(limiter.try_consume(10.0, later));
7514 assert!(!limiter.try_consume(10.1, later)); let later2 = later + Duration::from_secs(1);
7518 assert!(limiter.try_consume(3.0, later2));
7519
7520 let much_later = later2 + Duration::from_secs(2);
7522 assert!(limiter.try_consume(10.0, much_later)); assert!(!limiter.try_consume(0.1, much_later)); }
7525
7526 #[test]
7527 fn rate_limiter_fractional_consumption() {
7528 let now = Instant::now();
7529 let mut limiter = AddressObservationRateLimiter::new(10, now);
7530
7531 assert!(limiter.try_consume(0.5, now));
7533 assert!(limiter.try_consume(2.3, now));
7534 assert!(limiter.try_consume(7.2, now)); assert!(!limiter.try_consume(0.1, now)); let later = now + Duration::from_millis(100); assert!(limiter.try_consume(1.0, later));
7540 assert!(!limiter.try_consume(0.1, later));
7541 }
7542
7543 #[test]
7544 fn rate_limiter_zero_rate() {
7545 let now = Instant::now();
7546 let mut limiter = AddressObservationRateLimiter::new(0, now); assert!(!limiter.try_consume(1.0, now));
7550 assert!(!limiter.try_consume(0.1, now));
7551 assert!(!limiter.try_consume(0.001, now));
7552
7553 let later = now + Duration::from_secs(10);
7555 assert!(!limiter.try_consume(0.001, later));
7556 }
7557
7558 #[test]
7559 fn rate_limiter_high_rate() {
7560 let now = Instant::now();
7561 let mut limiter = AddressObservationRateLimiter::new(63, now); assert!(limiter.try_consume(60.0, now));
7565 assert!(limiter.try_consume(3.0, now));
7566 assert!(!limiter.try_consume(0.1, now)); let later = now + Duration::from_secs(1);
7570 assert!(limiter.try_consume(63.0, later)); assert!(!limiter.try_consume(0.1, later)); }
7573
7574 #[test]
7575 fn rate_limiter_time_precision() {
7576 let now = Instant::now();
7577 let mut limiter = AddressObservationRateLimiter::new(100, now); assert!(limiter.try_consume(100.0, now));
7581 assert!(!limiter.try_consume(0.1, now));
7582
7583 let later = now + Duration::from_millis(10);
7585 assert!(limiter.try_consume(0.8, later)); assert!(!limiter.try_consume(0.5, later)); let much_later = later + Duration::from_millis(100); assert!(limiter.try_consume(5.0, much_later)); limiter.tokens = 0.0; let final_time = much_later + Duration::from_millis(1);
7597 limiter.update_tokens(final_time); assert!(limiter.tokens >= 0.09 && limiter.tokens <= 0.11);
7602 }
7603
7604 #[test]
7605 fn per_path_rate_limiting_independent() {
7606 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7607 let now = Instant::now();
7608 let mut state = AddressDiscoveryState::new(&config, now);
7609
7610 state.observe_all_paths = true;
7612
7613 state.update_rate_limit(5.0);
7615
7616 state
7618 .sent_observations
7619 .insert(0, paths::PathAddressInfo::new());
7620 state
7621 .sent_observations
7622 .insert(1, paths::PathAddressInfo::new());
7623 state
7624 .sent_observations
7625 .insert(2, paths::PathAddressInfo::new());
7626
7627 state
7629 .sent_observations
7630 .get_mut(&0)
7631 .unwrap()
7632 .observed_address = Some(SocketAddr::new(
7633 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)),
7634 8080,
7635 ));
7636 state
7637 .sent_observations
7638 .get_mut(&1)
7639 .unwrap()
7640 .observed_address = Some(SocketAddr::new(
7641 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2)),
7642 8081,
7643 ));
7644 state
7645 .sent_observations
7646 .get_mut(&2)
7647 .unwrap()
7648 .observed_address = Some(SocketAddr::new(
7649 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 3)),
7650 8082,
7651 ));
7652
7653 for _ in 0..3 {
7655 assert!(state.should_send_observation(0, now));
7656 state.record_observation_sent(0);
7657 state.sent_observations.get_mut(&0).unwrap().notified = false;
7659 }
7660
7661 for _ in 0..2 {
7663 assert!(state.should_send_observation(1, now));
7664 state.record_observation_sent(1);
7665 state.sent_observations.get_mut(&1).unwrap().notified = false;
7667 }
7668
7669 assert!(!state.should_send_observation(2, now));
7671
7672 let later = now + Duration::from_secs(1);
7674
7675 assert!(state.should_send_observation(0, later));
7677 assert!(state.should_send_observation(1, later));
7678 assert!(state.should_send_observation(2, later));
7679 }
7680
7681 #[test]
7682 fn per_path_rate_limiting_with_path_specific_limits() {
7683 let now = Instant::now();
7684 let remote1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7685 let remote2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2)), 8081);
7686 let config = TransportConfig::default();
7687
7688 let mut path1 = paths::PathData::new(remote1, false, None, now, &config);
7690 let mut path2 = paths::PathData::new(remote2, false, None, now, &config);
7691
7692 path1.observation_rate_limiter = paths::PathObservationRateLimiter::new(10, now); path2.observation_rate_limiter = paths::PathObservationRateLimiter::new(5, now); for _ in 0..10 {
7698 assert!(path1.observation_rate_limiter.can_send(now));
7699 path1.observation_rate_limiter.consume_token(now);
7700 }
7701 assert!(!path1.observation_rate_limiter.can_send(now));
7702
7703 for _ in 0..5 {
7705 assert!(path2.observation_rate_limiter.can_send(now));
7706 path2.observation_rate_limiter.consume_token(now);
7707 }
7708 assert!(!path2.observation_rate_limiter.can_send(now));
7709 }
7710
7711 #[test]
7712 fn per_path_rate_limiting_address_change_detection() {
7713 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7714 let now = Instant::now();
7715 let mut state = AddressDiscoveryState::new(&config, now);
7716
7717 let path_id = 0;
7719 let addr1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7720 let addr2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2)), 8080);
7721
7722 assert!(state.should_send_observation(path_id, now));
7724
7725 let frame = state.queue_observed_address_frame(path_id, addr1);
7727 assert!(frame.is_some());
7728
7729 assert!(!state.should_send_observation(path_id, now));
7731
7732 if let Some(info) = state.sent_observations.get_mut(&path_id) {
7734 info.notified = false;
7735 info.observed_address = Some(addr2);
7736 }
7737
7738 assert!(state.should_send_observation(path_id, now));
7740 }
7741
7742 #[test]
7743 fn per_path_rate_limiting_migration() {
7744 let now = Instant::now();
7745 let remote1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080);
7746 let remote2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2)), 8081);
7747 let config = TransportConfig::default();
7748
7749 let mut path = paths::PathData::new(remote1, false, None, now, &config);
7751 path.observation_rate_limiter = paths::PathObservationRateLimiter::new(10, now);
7752
7753 for _ in 0..5 {
7755 assert!(path.observation_rate_limiter.can_send(now));
7756 path.observation_rate_limiter.consume_token(now);
7757 }
7758
7759 let mut new_path = paths::PathData::new(remote2, false, None, now, &config);
7761
7762 new_path.observation_rate_limiter = paths::PathObservationRateLimiter::new(10, now);
7765
7766 for _ in 0..10 {
7768 assert!(new_path.observation_rate_limiter.can_send(now));
7769 new_path.observation_rate_limiter.consume_token(now);
7770 }
7771 assert!(!new_path.observation_rate_limiter.can_send(now));
7772 }
7773
7774 #[test]
7775 fn per_path_rate_limiting_disabled_paths() {
7776 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7777 let now = Instant::now();
7778 let mut state = AddressDiscoveryState::new(&config, now);
7779
7780 assert!(state.should_send_observation(0, now));
7782
7783 assert!(!state.should_send_observation(1, now));
7785 assert!(!state.should_send_observation(2, now));
7786
7787 let later = now + Duration::from_secs(1);
7789 assert!(!state.should_send_observation(1, later));
7790 }
7791
7792 #[test]
7793 fn respecting_negotiated_max_observation_rate_basic() {
7794 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7795 let now = Instant::now();
7796 let mut state = AddressDiscoveryState::new(&config, now);
7797
7798 state.max_observation_rate = 10; state.rate_limiter = AddressObservationRateLimiter::new(10, now);
7801
7802 for _ in 0..10 {
7804 assert!(state.should_send_observation(0, now));
7805 }
7806 assert!(!state.should_send_observation(0, now));
7808 }
7809
7810 #[test]
7811 fn respecting_negotiated_max_observation_rate_zero() {
7812 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7813 let now = Instant::now();
7814 let mut state = AddressDiscoveryState::new(&config, now);
7815
7816 state.max_observation_rate = 0;
7818 state.rate_limiter = AddressObservationRateLimiter::new(0, now);
7819
7820 assert!(!state.should_send_observation(0, now));
7822 assert!(!state.should_send_observation(1, now));
7823
7824 let later = now + Duration::from_secs(10);
7826 assert!(!state.should_send_observation(0, later));
7827 }
7828
7829 #[test]
7830 fn respecting_negotiated_max_observation_rate_higher() {
7831 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7832 let now = Instant::now();
7833 let mut state = AddressDiscoveryState::new(&config, now);
7834
7835 state
7837 .sent_observations
7838 .insert(0, paths::PathAddressInfo::new());
7839 state
7840 .sent_observations
7841 .get_mut(&0)
7842 .unwrap()
7843 .observed_address = Some(SocketAddr::new(
7844 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)),
7845 8080,
7846 ));
7847
7848 state.update_rate_limit(5.0);
7850
7851 state.max_observation_rate = 20; for _ in 0..5 {
7856 assert!(state.should_send_observation(0, now));
7857 state.record_observation_sent(0);
7858 state.sent_observations.get_mut(&0).unwrap().notified = false;
7860 }
7861 assert!(!state.should_send_observation(0, now));
7863 }
7864
7865 #[test]
7866 fn respecting_negotiated_max_observation_rate_dynamic_update() {
7867 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7868 let now = Instant::now();
7869 let mut state = AddressDiscoveryState::new(&config, now);
7870
7871 state
7873 .sent_observations
7874 .insert(0, paths::PathAddressInfo::new());
7875 state
7876 .sent_observations
7877 .get_mut(&0)
7878 .unwrap()
7879 .observed_address = Some(SocketAddr::new(
7880 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)),
7881 8080,
7882 ));
7883
7884 for _ in 0..5 {
7886 assert!(state.should_send_observation(0, now));
7887 state.record_observation_sent(0);
7888 state.sent_observations.get_mut(&0).unwrap().notified = false;
7890 }
7891
7892 state.max_observation_rate = 3;
7896 state.rate_limiter.set_rate(3);
7897
7898 for _ in 0..3 {
7901 assert!(state.should_send_observation(0, now));
7902 state.record_observation_sent(0);
7903 state.sent_observations.get_mut(&0).unwrap().notified = false;
7905 }
7906
7907 assert!(!state.should_send_observation(0, now));
7909
7910 let later = now + Duration::from_secs(1);
7912 for _ in 0..3 {
7913 assert!(state.should_send_observation(0, later));
7914 state.record_observation_sent(0);
7915 state.sent_observations.get_mut(&0).unwrap().notified = false;
7917 }
7918
7919 assert!(!state.should_send_observation(0, later));
7921 }
7922
7923 #[test]
7924 fn respecting_negotiated_max_observation_rate_with_paths() {
7925 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7926 let now = Instant::now();
7927 let mut state = AddressDiscoveryState::new(&config, now);
7928
7929 state.observe_all_paths = true;
7931
7932 for i in 0..3 {
7934 state
7935 .sent_observations
7936 .insert(i, paths::PathAddressInfo::new());
7937 state
7938 .sent_observations
7939 .get_mut(&i)
7940 .unwrap()
7941 .observed_address = Some(SocketAddr::new(
7942 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100 + i as u8)),
7943 5000,
7944 ));
7945 }
7946
7947 for _ in 0..3 {
7950 for i in 0..3 {
7952 if state.should_send_observation(i, now) {
7953 state.record_observation_sent(i);
7954 state.sent_observations.get_mut(&i).unwrap().notified = false;
7956 }
7957 }
7958 }
7959
7960 assert!(state.should_send_observation(0, now));
7963 state.record_observation_sent(0);
7964
7965 assert!(!state.should_send_observation(0, now));
7967 assert!(!state.should_send_observation(1, now));
7968 assert!(!state.should_send_observation(2, now));
7969 }
7970
7971 #[test]
7972 fn queue_observed_address_frame_basic() {
7973 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7974 let now = Instant::now();
7975 let mut state = AddressDiscoveryState::new(&config, now);
7976
7977 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
7979 let frame = state.queue_observed_address_frame(0, address);
7980
7981 assert!(frame.is_some());
7983 let frame = frame.unwrap();
7984 assert_eq!(frame.address, address);
7985
7986 assert!(state.sent_observations.contains_key(&0));
7988 assert!(state.sent_observations.get(&0).unwrap().notified);
7989 }
7990
7991 #[test]
7992 fn queue_observed_address_frame_rate_limited() {
7993 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
7994 let now = Instant::now();
7995 let mut state = AddressDiscoveryState::new(&config, now);
7996
7997 state.observe_all_paths = true;
7999
8000 let mut addresses = Vec::new();
8002 for i in 0..10 {
8003 let addr = SocketAddr::new(
8004 IpAddr::V4(Ipv4Addr::new(192, 168, 1, i as u8)),
8005 5000 + i as u16,
8006 );
8007 addresses.push(addr);
8008 assert!(
8009 state.queue_observed_address_frame(i as u64, addr).is_some(),
8010 "Frame {} should be allowed",
8011 i + 1
8012 );
8013 }
8014
8015 let addr11 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 11)), 5011);
8017 assert!(
8018 state.queue_observed_address_frame(10, addr11).is_none(),
8019 "11th frame should be rate limited"
8020 );
8021 }
8022
8023 #[test]
8024 fn queue_observed_address_frame_disabled() {
8025 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8026 let now = Instant::now();
8027 let mut state = AddressDiscoveryState::new(&config, now);
8028
8029 state.enabled = false;
8031
8032 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8033
8034 assert!(state.queue_observed_address_frame(0, address).is_none());
8036 }
8037
8038 #[test]
8039 fn queue_observed_address_frame_already_notified() {
8040 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8041 let now = Instant::now();
8042 let mut state = AddressDiscoveryState::new(&config, now);
8043
8044 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8045
8046 assert!(state.queue_observed_address_frame(0, address).is_some());
8048
8049 assert!(state.queue_observed_address_frame(0, address).is_none());
8051
8052 let new_address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 101)), 5001);
8054 assert!(state.queue_observed_address_frame(0, new_address).is_none());
8055 }
8056
8057 #[test]
8058 fn queue_observed_address_frame_primary_path_only() {
8059 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8060 let now = Instant::now();
8061 let mut state = AddressDiscoveryState::new(&config, now);
8062
8063 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8064
8065 assert!(state.queue_observed_address_frame(0, address).is_some());
8067
8068 assert!(state.queue_observed_address_frame(1, address).is_none());
8070 assert!(state.queue_observed_address_frame(2, address).is_none());
8071 }
8072
8073 #[test]
8074 fn queue_observed_address_frame_updates_path_info() {
8075 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8076 let now = Instant::now();
8077 let mut state = AddressDiscoveryState::new(&config, now);
8078
8079 let address = SocketAddr::new(
8080 IpAddr::V6(Ipv6Addr::new(0x2001, 0xdb8, 0, 0, 0, 0, 0, 1)),
8081 5000,
8082 );
8083
8084 let frame = state.queue_observed_address_frame(0, address);
8086 assert!(frame.is_some());
8087
8088 let path_info = state.sent_observations.get(&0).unwrap();
8090 assert_eq!(path_info.observed_address, Some(address));
8091 assert!(path_info.notified);
8092
8093 assert_eq!(state.received_history.len(), 0);
8096 }
8097
8098 #[test]
8099 fn retransmits_includes_outbound_observations() {
8100 use crate::connection::spaces::Retransmits;
8101
8102 let mut retransmits = Retransmits::default();
8104
8105 assert!(retransmits.outbound_observations.is_empty());
8107
8108 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8110 let frame = frame::ObservedAddress {
8111 sequence_number: VarInt::from_u32(1),
8112 address,
8113 };
8114 retransmits.outbound_observations.push(frame);
8115
8116 assert_eq!(retransmits.outbound_observations.len(), 1);
8118 assert_eq!(retransmits.outbound_observations[0].address, address);
8119 }
8120
8121 #[test]
8122 fn check_for_address_observations_no_peer_support() {
8123 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8124 let now = Instant::now();
8125 let mut state = AddressDiscoveryState::new(&config, now);
8126
8127 state
8129 .sent_observations
8130 .insert(0, paths::PathAddressInfo::new());
8131 state
8132 .sent_observations
8133 .get_mut(&0)
8134 .unwrap()
8135 .observed_address = Some(SocketAddr::new(
8136 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)),
8137 5000,
8138 ));
8139
8140 let frames = state.check_for_address_observations(0, false, now);
8142
8143 assert!(frames.is_empty());
8145 }
8146
8147 #[test]
8148 fn check_for_address_observations_with_peer_support() {
8149 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8150 let now = Instant::now();
8151 let mut state = AddressDiscoveryState::new(&config, now);
8152
8153 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8155 state
8156 .sent_observations
8157 .insert(0, paths::PathAddressInfo::new());
8158 state
8159 .sent_observations
8160 .get_mut(&0)
8161 .unwrap()
8162 .observed_address = Some(address);
8163
8164 let frames = state.check_for_address_observations(0, true, now);
8166
8167 assert_eq!(frames.len(), 1);
8169 assert_eq!(frames[0].address, address);
8170
8171 assert!(state.sent_observations.get(&0).unwrap().notified);
8173 }
8174
8175 #[test]
8176 fn check_for_address_observations_rate_limited() {
8177 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8178 let now = Instant::now();
8179 let mut state = AddressDiscoveryState::new(&config, now);
8180
8181 let address = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8183 state
8184 .sent_observations
8185 .insert(0, paths::PathAddressInfo::new());
8186 state
8187 .sent_observations
8188 .get_mut(&0)
8189 .unwrap()
8190 .observed_address = Some(address);
8191
8192 for _ in 0..10 {
8194 let frames = state.check_for_address_observations(0, true, now);
8195 if frames.is_empty() {
8196 break;
8197 }
8198 state.sent_observations.get_mut(&0).unwrap().notified = false;
8200 }
8201
8202 assert_eq!(state.rate_limiter.tokens, 0.0);
8204
8205 state.sent_observations.get_mut(&0).unwrap().notified = false;
8207
8208 let frames2 = state.check_for_address_observations(0, true, now);
8210 assert_eq!(frames2.len(), 0);
8211
8212 state.sent_observations.get_mut(&0).unwrap().notified = false;
8214
8215 let later = now + Duration::from_millis(200); let frames3 = state.check_for_address_observations(0, true, later);
8218 assert_eq!(frames3.len(), 1);
8219 }
8220
8221 #[test]
8222 fn check_for_address_observations_multiple_paths() {
8223 let config = crate::transport_parameters::AddressDiscoveryConfig::SendAndReceive;
8224 let now = Instant::now();
8225 let mut state = AddressDiscoveryState::new(&config, now);
8226
8227 state.observe_all_paths = true;
8229
8230 let addr1 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 100)), 5000);
8232 let addr2 = SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 101)), 5001);
8233
8234 state
8235 .sent_observations
8236 .insert(0, paths::PathAddressInfo::new());
8237 state
8238 .sent_observations
8239 .get_mut(&0)
8240 .unwrap()
8241 .observed_address = Some(addr1);
8242
8243 state
8244 .sent_observations
8245 .insert(1, paths::PathAddressInfo::new());
8246 state
8247 .sent_observations
8248 .get_mut(&1)
8249 .unwrap()
8250 .observed_address = Some(addr2);
8251
8252 let frames = state.check_for_address_observations(0, true, now);
8254
8255 assert_eq!(frames.len(), 2);
8257
8258 let addresses: Vec<_> = frames.iter().map(|f| f.address).collect();
8260 assert!(addresses.contains(&addr1));
8261 assert!(addresses.contains(&addr2));
8262
8263 assert!(state.sent_observations.get(&0).unwrap().notified);
8265 assert!(state.sent_observations.get(&1).unwrap().notified);
8266 }
8267
8268 #[test]
8270 fn test_rate_limiter_configuration() {
8271 let state = AddressDiscoveryState::new_with_params(true, 10.0, false);
8273 assert_eq!(state.rate_limiter.rate, 10.0);
8274 assert_eq!(state.rate_limiter.max_tokens, 10.0);
8275 assert_eq!(state.rate_limiter.tokens, 10.0);
8276
8277 let state = AddressDiscoveryState::new_with_params(true, 63.0, false);
8278 assert_eq!(state.rate_limiter.rate, 63.0);
8279 assert_eq!(state.rate_limiter.max_tokens, 63.0);
8280 }
8281
8282 #[test]
8283 fn test_rate_limiter_update_configuration() {
8284 let mut state = AddressDiscoveryState::new_with_params(true, 5.0, false);
8285
8286 assert_eq!(state.rate_limiter.rate, 5.0);
8288
8289 state.update_rate_limit(10.0);
8291 assert_eq!(state.rate_limiter.rate, 10.0);
8292 assert_eq!(state.rate_limiter.max_tokens, 10.0);
8293
8294 state.rate_limiter.tokens = 15.0;
8296 state.update_rate_limit(8.0);
8297 assert_eq!(state.rate_limiter.tokens, 8.0);
8298 }
8299
8300 #[test]
8301 fn test_rate_limiter_from_transport_params() {
8302 let mut params = TransportParameters::default();
8303 params.address_discovery = Some(AddressDiscoveryConfig::SendAndReceive);
8304
8305 let state = AddressDiscoveryState::from_transport_params(¶ms);
8306 assert!(state.is_some());
8307 let state = state.unwrap();
8308 assert_eq!(state.rate_limiter.rate, 10.0); assert!(!state.observe_all_paths); }
8311
8312 #[test]
8313 fn test_rate_limiter_zero_rate() {
8314 let state = AddressDiscoveryState::new_with_params(true, 0.0, false);
8315 assert_eq!(state.rate_limiter.rate, 0.0);
8316 assert_eq!(state.rate_limiter.tokens, 0.0);
8317
8318 let address = "192.168.1.1:443".parse().unwrap();
8320 let mut state = AddressDiscoveryState::new_with_params(true, 0.0, false);
8321 let frame = state.queue_observed_address_frame(0, address);
8322 assert!(frame.is_none());
8323 }
8324
8325 #[test]
8326 fn test_rate_limiter_configuration_edge_cases() {
8327 let state = AddressDiscoveryState::new_with_params(true, 63.0, false);
8329 assert_eq!(state.rate_limiter.rate, 63.0);
8330
8331 let state = AddressDiscoveryState::new_with_params(true, 100.0, false);
8333 assert_eq!(state.rate_limiter.rate, 100.0);
8335
8336 let state = AddressDiscoveryState::new_with_params(true, 2.5, false);
8338 assert_eq!(state.rate_limiter.rate, 2.0);
8340 }
8341
8342 #[test]
8343 fn test_rate_limiter_runtime_update() {
8344 let mut state = AddressDiscoveryState::new_with_params(true, 10.0, false);
8345 let now = Instant::now();
8346
8347 state.rate_limiter.tokens = 5.0;
8349
8350 state.update_rate_limit(3.0);
8352
8353 assert_eq!(state.rate_limiter.tokens, 3.0);
8355 assert_eq!(state.rate_limiter.rate, 3.0);
8356 assert_eq!(state.rate_limiter.max_tokens, 3.0);
8357
8358 let later = now + Duration::from_secs(1);
8360 state.rate_limiter.update_tokens(later);
8361
8362 assert_eq!(state.rate_limiter.tokens, 3.0);
8364 }
8365
8366 #[test]
8368 fn test_address_discovery_state_initialization_default() {
8369 let now = Instant::now();
8371 let default_config = crate::transport_parameters::AddressDiscoveryConfig::default();
8372
8373 let address_discovery_state = Some(AddressDiscoveryState::new(&default_config, now));
8376
8377 assert!(address_discovery_state.is_some());
8378 let state = address_discovery_state.unwrap();
8379
8380 assert!(state.enabled); assert_eq!(state.max_observation_rate, 10); assert!(!state.observe_all_paths);
8384 }
8385
8386 #[test]
8387 fn test_address_discovery_state_initialization_on_handshake() {
8388 let now = Instant::now();
8390
8391 let mut address_discovery_state = Some(AddressDiscoveryState::new(
8393 &crate::transport_parameters::AddressDiscoveryConfig::default(),
8394 now,
8395 ));
8396
8397 let peer_params = TransportParameters {
8399 address_discovery: Some(AddressDiscoveryConfig::SendAndReceive),
8400 ..TransportParameters::default()
8401 };
8402
8403 if let Some(peer_config) = &peer_params.address_discovery {
8405 address_discovery_state = Some(AddressDiscoveryState::new(peer_config, now));
8407 }
8408
8409 assert!(address_discovery_state.is_some());
8411 let state = address_discovery_state.unwrap();
8412 assert!(state.enabled);
8413 assert_eq!(state.max_observation_rate, 10); assert!(!state.observe_all_paths); }
8417
8418 #[test]
8419 fn test_address_discovery_negotiation_disabled_peer() {
8420 let now = Instant::now();
8422
8423 let our_config = AddressDiscoveryConfig::SendAndReceive;
8425 let mut address_discovery_state = Some(AddressDiscoveryState::new(&our_config, now));
8426
8427 let peer_params = TransportParameters {
8429 address_discovery: None,
8430 ..TransportParameters::default()
8431 };
8432
8433 if peer_params.address_discovery.is_none() {
8435 if let Some(state) = &mut address_discovery_state {
8436 state.enabled = false;
8437 }
8438 }
8439
8440 let state = address_discovery_state.unwrap();
8442 assert!(!state.enabled); }
8444
8445 #[test]
8446 fn test_address_discovery_negotiation_rate_limiting() {
8447 let now = Instant::now();
8449
8450 let our_config = AddressDiscoveryConfig::SendAndReceive;
8452 let mut address_discovery_state = Some(AddressDiscoveryState::new(&our_config, now));
8453
8454 if let Some(state) = &mut address_discovery_state {
8456 state.max_observation_rate = 30;
8457 state.update_rate_limit(30.0);
8458 }
8459
8460 let peer_params = TransportParameters {
8462 address_discovery: Some(AddressDiscoveryConfig::SendAndReceive),
8463 ..TransportParameters::default()
8464 };
8465
8466 if let (Some(state), Some(_peer_config)) =
8469 (&mut address_discovery_state, &peer_params.address_discovery)
8470 {
8471 let peer_rate = 15u8;
8474 let negotiated_rate = state.max_observation_rate.min(peer_rate);
8475 state.update_rate_limit(negotiated_rate as f64);
8476 }
8477
8478 let state = address_discovery_state.unwrap();
8480 assert_eq!(state.rate_limiter.rate, 15.0); }
8482
8483 #[test]
8484 fn test_address_discovery_path_initialization() {
8485 let now = Instant::now();
8487 let config = AddressDiscoveryConfig::SendAndReceive;
8488 let mut state = AddressDiscoveryState::new(&config, now);
8489
8490 assert!(state.sent_observations.is_empty());
8492 assert!(state.received_observations.is_empty());
8493
8494 let should_send = state.should_send_observation(0, now);
8496 assert!(should_send); }
8501
8502 #[test]
8503 fn test_address_discovery_multiple_path_initialization() {
8504 let now = Instant::now();
8506 let config = AddressDiscoveryConfig::SendAndReceive;
8507 let mut state = AddressDiscoveryState::new(&config, now);
8508
8509 assert!(state.should_send_observation(0, now)); assert!(!state.should_send_observation(1, now)); assert!(!state.should_send_observation(2, now)); state.observe_all_paths = true;
8516 assert!(state.should_send_observation(1, now)); assert!(state.should_send_observation(2, now)); let config_primary_only = AddressDiscoveryConfig::SendAndReceive;
8521 let mut state_primary = AddressDiscoveryState::new(&config_primary_only, now);
8522
8523 assert!(state_primary.should_send_observation(0, now)); assert!(!state_primary.should_send_observation(1, now)); }
8526
8527 #[test]
8528 fn test_handle_observed_address_frame_valid() {
8529 let now = Instant::now();
8531 let config = AddressDiscoveryConfig::SendAndReceive;
8532 let mut state = AddressDiscoveryState::new(&config, now);
8533
8534 let observed_addr = SocketAddr::from(([192, 168, 1, 100], 5000));
8536 state.handle_observed_address(observed_addr, 0, now);
8537
8538 assert_eq!(state.received_history.len(), 1);
8540 assert_eq!(state.received_history[0].address, observed_addr);
8541 assert_eq!(state.received_history[0].path_id, 0);
8542 assert_eq!(state.received_history[0].received_at, now);
8543
8544 let path_info = state.received_observations.get(&0).unwrap();
8546 assert_eq!(path_info.observed_address, Some(observed_addr));
8547 assert_eq!(path_info.last_observed, Some(now));
8548 assert_eq!(path_info.observation_count, 1);
8549 }
8550
8551 #[test]
8552 fn test_handle_multiple_received_history() {
8553 let now = Instant::now();
8555 let config = AddressDiscoveryConfig::SendAndReceive;
8556 let mut state = AddressDiscoveryState::new(&config, now);
8557
8558 let addr1 = SocketAddr::from(([192, 168, 1, 100], 5000));
8560 let addr2 = SocketAddr::from(([10, 0, 0, 50], 6000));
8561 let addr3 = SocketAddr::from(([192, 168, 1, 100], 7000)); state.handle_observed_address(addr1, 0, now);
8564 state.handle_observed_address(addr2, 1, now);
8565 state.handle_observed_address(addr3, 0, now + Duration::from_millis(100));
8566
8567 assert_eq!(state.received_history.len(), 3);
8569
8570 let path0_info = state.received_observations.get(&0).unwrap();
8572 assert_eq!(path0_info.observed_address, Some(addr3));
8573 assert_eq!(path0_info.observation_count, 1); let path1_info = state.received_observations.get(&1).unwrap();
8577 assert_eq!(path1_info.observed_address, Some(addr2));
8578 assert_eq!(path1_info.observation_count, 1);
8579 }
8580
8581 #[test]
8582 fn test_get_observed_address() {
8583 let now = Instant::now();
8585 let config = AddressDiscoveryConfig::SendAndReceive;
8586 let mut state = AddressDiscoveryState::new(&config, now);
8587
8588 assert_eq!(state.get_observed_address(0), None);
8590
8591 let addr = SocketAddr::from(([192, 168, 1, 100], 5000));
8593 state.handle_observed_address(addr, 0, now);
8594
8595 assert_eq!(state.get_observed_address(0), Some(addr));
8597
8598 assert_eq!(state.get_observed_address(999), None);
8600 }
8601
8602 #[test]
8603 fn test_has_unnotified_changes() {
8604 let now = Instant::now();
8606 let config = AddressDiscoveryConfig::SendAndReceive;
8607 let mut state = AddressDiscoveryState::new(&config, now);
8608
8609 assert!(!state.has_unnotified_changes());
8611
8612 let addr = SocketAddr::from(([192, 168, 1, 100], 5000));
8614 state.handle_observed_address(addr, 0, now);
8615 assert!(state.has_unnotified_changes());
8616
8617 if let Some(path_info) = state.received_observations.get_mut(&0) {
8619 path_info.notified = true;
8620 }
8621 assert!(!state.has_unnotified_changes());
8622
8623 let addr2 = SocketAddr::from(([192, 168, 1, 100], 6000));
8625 state.handle_observed_address(addr2, 0, now + Duration::from_secs(1));
8626 assert!(state.has_unnotified_changes());
8627 }
8628
8629 #[test]
8630 fn test_address_discovery_disabled() {
8631 let now = Instant::now();
8633 let config = AddressDiscoveryConfig::SendAndReceive;
8634 let mut state = AddressDiscoveryState::new(&config, now);
8635
8636 state.enabled = false;
8638
8639 let addr = SocketAddr::from(([192, 168, 1, 100], 5000));
8641 state.handle_observed_address(addr, 0, now);
8642
8643 assert_eq!(state.received_history.len(), 0);
8645
8646 assert!(!state.should_send_observation(0, now));
8648 }
8649
8650 #[test]
8651 fn test_rate_limiting_basic() {
8652 let now = Instant::now();
8654 let config = AddressDiscoveryConfig::SendAndReceive;
8655 let mut state = AddressDiscoveryState::new(&config, now);
8656
8657 state.observe_all_paths = true;
8659 state.rate_limiter.set_rate(2); assert!(state.should_send_observation(0, now));
8663 state.record_observation_sent(0);
8665
8666 assert!(state.should_send_observation(1, now));
8668 state.record_observation_sent(1);
8669
8670 assert!(!state.should_send_observation(2, now));
8672
8673 let later = now + Duration::from_millis(500);
8675 assert!(state.should_send_observation(3, later));
8676 state.record_observation_sent(3);
8677
8678 assert!(!state.should_send_observation(4, later));
8680
8681 let _one_sec_later = now + Duration::from_secs(1);
8685 let two_sec_later = now + Duration::from_secs(2);
8689 assert!(state.should_send_observation(5, two_sec_later));
8690 state.record_observation_sent(5);
8691
8692 assert!(state.should_send_observation(6, two_sec_later));
8703 state.record_observation_sent(6);
8704
8705 assert!(
8707 !state.should_send_observation(7, two_sec_later),
8708 "Expected no tokens available"
8709 );
8710 }
8711
8712 #[test]
8713 fn test_rate_limiting_per_path() {
8714 let now = Instant::now();
8716 let config = AddressDiscoveryConfig::SendAndReceive;
8717 let mut state = AddressDiscoveryState::new(&config, now);
8718
8719 state
8721 .sent_observations
8722 .insert(0, paths::PathAddressInfo::new());
8723 state
8724 .sent_observations
8725 .get_mut(&0)
8726 .unwrap()
8727 .observed_address = Some(SocketAddr::new(
8728 IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)),
8729 8080,
8730 ));
8731
8732 for _ in 0..10 {
8734 assert!(state.should_send_observation(0, now));
8735 state.record_observation_sent(0);
8736 state.sent_observations.get_mut(&0).unwrap().notified = false;
8738 }
8739
8740 assert!(!state.should_send_observation(0, now));
8742
8743 let later = now + Duration::from_millis(100);
8745 assert!(state.should_send_observation(0, later));
8746 state.record_observation_sent(0);
8747
8748 state.sent_observations.get_mut(&0).unwrap().notified = false;
8750
8751 assert!(!state.should_send_observation(0, later));
8753 }
8754
8755 #[test]
8756 fn test_rate_limiting_zero_rate() {
8757 let now = Instant::now();
8759 let config = AddressDiscoveryConfig::SendAndReceive;
8760 let mut state = AddressDiscoveryState::new(&config, now);
8761
8762 state.rate_limiter.set_rate(0);
8764 state.rate_limiter.tokens = 0.0;
8765 state.rate_limiter.max_tokens = 0.0;
8766
8767 assert!(!state.should_send_observation(0, now));
8769 assert!(!state.should_send_observation(0, now + Duration::from_secs(10)));
8770 assert!(!state.should_send_observation(0, now + Duration::from_secs(100)));
8771 }
8772
8773 #[test]
8774 fn test_rate_limiting_update() {
8775 let now = Instant::now();
8777 let config = AddressDiscoveryConfig::SendAndReceive;
8778 let mut state = AddressDiscoveryState::new(&config, now);
8779
8780 state.observe_all_paths = true;
8782
8783 for i in 0..12 {
8785 state
8786 .sent_observations
8787 .insert(i, paths::PathAddressInfo::new());
8788 state
8789 .sent_observations
8790 .get_mut(&i)
8791 .unwrap()
8792 .observed_address = Some(SocketAddr::new(
8793 IpAddr::V4(Ipv4Addr::new(192, 168, 1, (i + 1) as u8)),
8794 8080,
8795 ));
8796 }
8797
8798 for i in 0..10 {
8801 assert!(state.should_send_observation(i, now));
8802 state.record_observation_sent(i);
8803 }
8804 assert!(!state.should_send_observation(10, now));
8806
8807 state.update_rate_limit(20.0);
8809
8810 let later = now + Duration::from_millis(50);
8813 assert!(state.should_send_observation(10, later));
8814 state.record_observation_sent(10);
8815
8816 let later2 = now + Duration::from_millis(100);
8818 assert!(state.should_send_observation(11, later2));
8819 }
8820
8821 #[test]
8822 fn test_rate_limiting_burst() {
8823 let now = Instant::now();
8825 let config = AddressDiscoveryConfig::SendAndReceive;
8826 let mut state = AddressDiscoveryState::new(&config, now);
8827
8828 for _ in 0..10 {
8830 assert!(state.should_send_observation(0, now));
8831 state.record_observation_sent(0);
8832 }
8833
8834 assert!(!state.should_send_observation(0, now));
8836
8837 let later = now + Duration::from_millis(100);
8839 assert!(state.should_send_observation(0, later));
8840 state.record_observation_sent(0);
8841 assert!(!state.should_send_observation(0, later));
8842 }
8843
8844 #[test]
8845 fn test_connection_rate_limiting_with_check_observations() {
8846 let now = Instant::now();
8848 let config = AddressDiscoveryConfig::SendAndReceive;
8849 let mut state = AddressDiscoveryState::new(&config, now);
8850
8851 let mut path_info = paths::PathAddressInfo::new();
8853 path_info.update_observed_address(
8854 SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1)), 8080),
8855 now,
8856 );
8857 state.sent_observations.insert(0, path_info);
8858
8859 let frame1 =
8861 state.queue_observed_address_frame(0, SocketAddr::from(([192, 168, 1, 1], 8080)));
8862 assert!(frame1.is_some());
8863 state.record_observation_sent(0);
8864
8865 if let Some(info) = state.sent_observations.get_mut(&0) {
8867 info.notified = false;
8868 }
8869
8870 for _ in 1..10 {
8872 if let Some(info) = state.sent_observations.get_mut(&0) {
8874 info.notified = false;
8875 }
8876 let frame =
8877 state.queue_observed_address_frame(0, SocketAddr::from(([192, 168, 1, 1], 8080)));
8878 assert!(frame.is_some());
8879 state.record_observation_sent(0);
8880 }
8881
8882 if let Some(info) = state.sent_observations.get_mut(&0) {
8884 info.notified = false;
8885 }
8886 let frame3 =
8887 state.queue_observed_address_frame(0, SocketAddr::from(([192, 168, 1, 1], 8080)));
8888 assert!(frame3.is_none()); let later = now + Duration::from_millis(100);
8892 state.rate_limiter.update_tokens(later); if let Some(info) = state.sent_observations.get_mut(&0) {
8896 info.notified = false;
8897 }
8898
8899 let frame4 =
8900 state.queue_observed_address_frame(0, SocketAddr::from(([192, 168, 1, 1], 8080)));
8901 assert!(frame4.is_some()); }
8903
8904 #[test]
8905 fn test_queue_observed_address_frame() {
8906 let now = Instant::now();
8908 let config = AddressDiscoveryConfig::SendAndReceive;
8909 let mut state = AddressDiscoveryState::new(&config, now);
8910
8911 let addr = SocketAddr::from(([192, 168, 1, 100], 5000));
8912
8913 let frame = state.queue_observed_address_frame(0, addr);
8915 assert!(frame.is_some());
8916 assert_eq!(frame.unwrap().address, addr);
8917
8918 state.record_observation_sent(0);
8920
8921 for i in 0..9 {
8923 if let Some(info) = state.sent_observations.get_mut(&0) {
8925 info.notified = false;
8926 }
8927
8928 let frame = state.queue_observed_address_frame(0, addr);
8929 assert!(frame.is_some(), "Frame {} should be allowed", i + 2);
8930 state.record_observation_sent(0);
8931 }
8932
8933 if let Some(info) = state.sent_observations.get_mut(&0) {
8935 info.notified = false;
8936 }
8937
8938 let frame = state.queue_observed_address_frame(0, addr);
8940 assert!(frame.is_none(), "11th frame should be rate limited");
8941 }
8942
8943 #[test]
8944 fn test_multi_path_basic() {
8945 let now = Instant::now();
8947 let config = AddressDiscoveryConfig::SendAndReceive;
8948 let mut state = AddressDiscoveryState::new(&config, now);
8949
8950 let addr1 = SocketAddr::from(([192, 168, 1, 1], 5000));
8951 let addr2 = SocketAddr::from(([10, 0, 0, 1], 6000));
8952 let addr3 = SocketAddr::from(([172, 16, 0, 1], 7000));
8953
8954 state.handle_observed_address(addr1, 0, now);
8956 state.handle_observed_address(addr2, 1, now);
8957 state.handle_observed_address(addr3, 2, now);
8958
8959 assert_eq!(state.get_observed_address(0), Some(addr1));
8961 assert_eq!(state.get_observed_address(1), Some(addr2));
8962 assert_eq!(state.get_observed_address(2), Some(addr3));
8963
8964 assert!(state.has_unnotified_changes());
8966
8967 assert_eq!(state.received_history.len(), 3);
8969 }
8970
8971 #[test]
8972 fn test_multi_path_observe_primary_only() {
8973 let now = Instant::now();
8975 let config = AddressDiscoveryConfig::SendAndReceive;
8976 let mut state = AddressDiscoveryState::new(&config, now);
8977
8978 assert!(state.should_send_observation(0, now));
8980 state.record_observation_sent(0);
8981
8982 assert!(!state.should_send_observation(1, now));
8984 assert!(!state.should_send_observation(2, now));
8985
8986 let addr = SocketAddr::from(([192, 168, 1, 1], 5000));
8988 assert!(state.queue_observed_address_frame(0, addr).is_some());
8989 assert!(state.queue_observed_address_frame(1, addr).is_none());
8990 assert!(state.queue_observed_address_frame(2, addr).is_none());
8991 }
8992
8993 #[test]
8994 fn test_multi_path_rate_limiting() {
8995 let now = Instant::now();
8997 let config = AddressDiscoveryConfig::SendAndReceive;
8998 let mut state = AddressDiscoveryState::new(&config, now);
8999
9000 state.observe_all_paths = true;
9002
9003 for i in 0..21 {
9005 state
9006 .sent_observations
9007 .insert(i, paths::PathAddressInfo::new());
9008 state
9009 .sent_observations
9010 .get_mut(&i)
9011 .unwrap()
9012 .observed_address = Some(SocketAddr::new(
9013 IpAddr::V4(Ipv4Addr::new(192, 168, 1, (i + 1) as u8)),
9014 8080,
9015 ));
9016 }
9017
9018 for i in 0..10 {
9020 assert!(state.should_send_observation(i, now));
9021 state.record_observation_sent(i);
9022 }
9023
9024 assert!(!state.should_send_observation(10, now));
9026
9027 state.sent_observations.get_mut(&0).unwrap().notified = false;
9029 assert!(!state.should_send_observation(0, now)); let later = now + Duration::from_secs(1);
9033 for i in 10..20 {
9034 assert!(state.should_send_observation(i, later));
9035 state.record_observation_sent(i);
9036 }
9037 assert!(!state.should_send_observation(20, later));
9039 }
9040
9041 #[test]
9042 fn test_multi_path_address_changes() {
9043 let now = Instant::now();
9045 let config = AddressDiscoveryConfig::SendAndReceive;
9046 let mut state = AddressDiscoveryState::new(&config, now);
9047
9048 let addr1a = SocketAddr::from(([192, 168, 1, 1], 5000));
9049 let addr1b = SocketAddr::from(([192, 168, 1, 2], 5000));
9050 let addr2a = SocketAddr::from(([10, 0, 0, 1], 6000));
9051 let addr2b = SocketAddr::from(([10, 0, 0, 2], 6000));
9052
9053 state.handle_observed_address(addr1a, 0, now);
9055 state.handle_observed_address(addr2a, 1, now);
9056
9057 if let Some(info) = state.received_observations.get_mut(&0) {
9059 info.notified = true;
9060 }
9061 if let Some(info) = state.received_observations.get_mut(&1) {
9062 info.notified = true;
9063 }
9064 assert!(!state.has_unnotified_changes());
9065
9066 state.handle_observed_address(addr1b, 0, now + Duration::from_secs(1));
9068 assert!(state.has_unnotified_changes());
9069
9070 assert_eq!(state.get_observed_address(0), Some(addr1b));
9072 assert_eq!(state.get_observed_address(1), Some(addr2a));
9073
9074 if let Some(info) = state.received_observations.get_mut(&0) {
9076 info.notified = true;
9077 }
9078 assert!(!state.has_unnotified_changes());
9079
9080 state.handle_observed_address(addr2b, 1, now + Duration::from_secs(2));
9082 assert!(state.has_unnotified_changes());
9083 }
9084
9085 #[test]
9086 fn test_multi_path_migration() {
9087 let now = Instant::now();
9089 let config = AddressDiscoveryConfig::SendAndReceive;
9090 let mut state = AddressDiscoveryState::new(&config, now);
9091
9092 let addr_old = SocketAddr::from(([192, 168, 1, 1], 5000));
9093 let addr_new = SocketAddr::from(([10, 0, 0, 1], 6000));
9094
9095 state.handle_observed_address(addr_old, 0, now);
9097 assert_eq!(state.get_observed_address(0), Some(addr_old));
9098
9099 state.handle_observed_address(addr_new, 1, now + Duration::from_secs(1));
9101
9102 assert_eq!(state.get_observed_address(0), Some(addr_old));
9104 assert_eq!(state.get_observed_address(1), Some(addr_new));
9105
9106 assert_eq!(state.received_observations.len(), 2);
9109 }
9110
9111 #[test]
9112 fn test_check_for_address_observations_multi_path() {
9113 let now = Instant::now();
9115 let config = AddressDiscoveryConfig::SendAndReceive;
9116 let mut state = AddressDiscoveryState::new(&config, now);
9117
9118 state.observe_all_paths = true;
9120
9121 let addr1 = SocketAddr::from(([192, 168, 1, 1], 5000));
9123 let addr2 = SocketAddr::from(([10, 0, 0, 1], 6000));
9124 let addr3 = SocketAddr::from(([172, 16, 0, 1], 7000));
9125
9126 state
9128 .sent_observations
9129 .insert(0, paths::PathAddressInfo::new());
9130 state
9131 .sent_observations
9132 .get_mut(&0)
9133 .unwrap()
9134 .observed_address = Some(addr1);
9135 state
9136 .sent_observations
9137 .insert(1, paths::PathAddressInfo::new());
9138 state
9139 .sent_observations
9140 .get_mut(&1)
9141 .unwrap()
9142 .observed_address = Some(addr2);
9143 state
9144 .sent_observations
9145 .insert(2, paths::PathAddressInfo::new());
9146 state
9147 .sent_observations
9148 .get_mut(&2)
9149 .unwrap()
9150 .observed_address = Some(addr3);
9151
9152 let frames = state.check_for_address_observations(0, true, now);
9154
9155 assert_eq!(frames.len(), 3);
9157
9158 let frame_addrs: Vec<_> = frames.iter().map(|f| f.address).collect();
9160 assert!(frame_addrs.contains(&addr1), "addr1 should be in frames");
9161 assert!(frame_addrs.contains(&addr2), "addr2 should be in frames");
9162 assert!(frame_addrs.contains(&addr3), "addr3 should be in frames");
9163
9164 assert!(!state.has_unnotified_changes());
9166 }
9167
9168 #[test]
9169 fn test_multi_path_with_peer_not_supporting() {
9170 let now = Instant::now();
9172 let config = AddressDiscoveryConfig::SendAndReceive;
9173 let mut state = AddressDiscoveryState::new(&config, now);
9174
9175 state.handle_observed_address(SocketAddr::from(([192, 168, 1, 1], 5000)), 0, now);
9177 state.handle_observed_address(SocketAddr::from(([10, 0, 0, 1], 6000)), 1, now);
9178
9179 let frames = state.check_for_address_observations(0, false, now);
9181 assert_eq!(frames.len(), 0);
9182
9183 assert!(state.has_unnotified_changes());
9185 }
9186
9187 #[test]
9189 fn test_bootstrap_node_aggressive_observation_mode() {
9190 let config = AddressDiscoveryConfig::SendAndReceive;
9192 let now = Instant::now();
9193 let mut state = AddressDiscoveryState::new(&config, now);
9194
9195 assert!(!state.is_bootstrap_mode());
9197
9198 state.set_bootstrap_mode(true);
9200 assert!(state.is_bootstrap_mode());
9201
9202 assert!(state.should_observe_path(0)); assert!(state.should_observe_path(1)); assert!(state.should_observe_path(2));
9206
9207 let bootstrap_rate = state.get_effective_rate_limit();
9209 assert!(bootstrap_rate > 10.0); }
9211
9212 #[test]
9213 fn test_bootstrap_node_immediate_observation() {
9214 let config = AddressDiscoveryConfig::SendAndReceive;
9216 let now = Instant::now();
9217 let mut state = AddressDiscoveryState::new(&config, now);
9218 state.set_bootstrap_mode(true);
9219
9220 let addr = SocketAddr::from(([192, 168, 1, 100], 5000));
9222 state.handle_observed_address(addr, 0, now);
9223
9224 assert!(state.should_send_observation_immediately(true));
9226
9227 assert!(state.should_send_observation(0, now));
9229
9230 let frame = state.queue_observed_address_frame(0, addr);
9232 assert!(frame.is_some());
9233 }
9234
9235 #[test]
9236 fn test_bootstrap_node_multiple_path_observations() {
9237 let config = AddressDiscoveryConfig::SendAndReceive;
9239 let now = Instant::now();
9240 let mut state = AddressDiscoveryState::new(&config, now);
9241 state.set_bootstrap_mode(true);
9242
9243 let addrs = vec![
9245 (0u64, SocketAddr::from(([192, 168, 1, 1], 5000))),
9246 (1u64, SocketAddr::from(([10, 0, 0, 1], 6000))),
9247 (2u64, SocketAddr::from(([172, 16, 0, 1], 7000))),
9248 ];
9249
9250 for (path_id, addr) in &addrs {
9251 state
9252 .sent_observations
9253 .insert(*path_id, paths::PathAddressInfo::new());
9254 state
9255 .sent_observations
9256 .get_mut(path_id)
9257 .unwrap()
9258 .observed_address = Some(*addr);
9259 }
9260
9261 let frames = state.check_for_address_observations(0, true, now);
9263 assert_eq!(frames.len(), 3);
9264
9265 for (_, addr) in &addrs {
9267 assert!(frames.iter().any(|f| f.address == *addr));
9268 }
9269 }
9270
9271 #[test]
9272 fn test_bootstrap_node_rate_limit_override() {
9273 let config = AddressDiscoveryConfig::SendAndReceive;
9275 let now = Instant::now();
9276 let mut state = AddressDiscoveryState::new(&config, now);
9277 state.set_bootstrap_mode(true);
9278
9279 let addr = SocketAddr::from(([192, 168, 1, 1], 5000));
9281
9282 for i in 0..10 {
9284 state.handle_observed_address(addr, i, now);
9285 let can_send = state.should_send_observation(i, now);
9286 assert!(can_send, "Bootstrap node should send observation {i}");
9287 state.record_observation_sent(i);
9288 }
9289 }
9290
9291 #[test]
9292 fn test_bootstrap_node_configuration() {
9293 let config = AddressDiscoveryConfig::SendAndReceive;
9295 let mut state = AddressDiscoveryState::new(&config, Instant::now());
9296
9297 state.set_bootstrap_mode(true);
9299
9300 assert!(state.bootstrap_mode);
9302 assert!(state.enabled);
9303
9304 let effective_rate = state.get_effective_rate_limit();
9306 assert!(effective_rate > state.max_observation_rate as f64);
9307 }
9308
9309 #[test]
9310 fn test_bootstrap_node_persistent_observation() {
9311 let config = AddressDiscoveryConfig::SendAndReceive;
9313 let mut now = Instant::now();
9314 let mut state = AddressDiscoveryState::new(&config, now);
9315 state.set_bootstrap_mode(true);
9316
9317 let addr1 = SocketAddr::from(([192, 168, 1, 1], 5000));
9318 let addr2 = SocketAddr::from(([192, 168, 1, 2], 5000));
9319
9320 state.handle_observed_address(addr1, 0, now);
9322 assert!(state.should_send_observation(0, now));
9323 state.record_observation_sent(0);
9324
9325 now += Duration::from_secs(60);
9327 state.handle_observed_address(addr2, 0, now);
9328
9329 assert!(state.should_send_observation(0, now));
9331 }
9332
9333 #[test]
9334 fn test_bootstrap_node_multi_peer_support() {
9335 let config = AddressDiscoveryConfig::SendAndReceive;
9338 let now = Instant::now();
9339 let mut state = AddressDiscoveryState::new(&config, now);
9340 state.set_bootstrap_mode(true);
9341
9342 let peer_addresses: Vec<(u64, SocketAddr)> = vec![
9344 (0, SocketAddr::from(([192, 168, 1, 1], 5000))), (1, SocketAddr::from(([10, 0, 0, 1], 6000))), (2, SocketAddr::from(([172, 16, 0, 1], 7000))), (3, SocketAddr::from(([192, 168, 2, 1], 8000))), ];
9349
9350 for (path_id, addr) in &peer_addresses {
9352 state
9353 .sent_observations
9354 .insert(*path_id, paths::PathAddressInfo::new());
9355 state
9356 .sent_observations
9357 .get_mut(path_id)
9358 .unwrap()
9359 .observed_address = Some(*addr);
9360 }
9361
9362 let frames = state.check_for_address_observations(0, true, now);
9364 assert_eq!(frames.len(), peer_addresses.len());
9365
9366 for (_, addr) in &peer_addresses {
9368 assert!(frames.iter().any(|f| f.address == *addr));
9369 }
9370 }
9371
9372 mod address_discovery_tests {
9374 include!("address_discovery_tests.rs");
9375 }
9376}