diff --git a/noq-proto/src/connection/mod.rs b/noq-proto/src/connection/mod.rs index b12e7c2ab6..9635c214a5 100644 --- a/noq-proto/src/connection/mod.rs +++ b/noq-proto/src/connection/mod.rs @@ -74,15 +74,14 @@ use packet_crypto::CryptoState; pub(crate) use packet_crypto::EncryptionLevel; mod paths; -pub use paths::{ - ClosedPath, PathAbandonReason, PathEvent, PathId, PathStatus, RttEstimator, SetPathStatusError, -}; +pub use paths::{ClosedPath, PathAbandonReason, PathEvent, PathId, RttEstimator, SetPathStatusError}; use paths::{PathData, PathState}; pub(crate) mod qlog; pub(crate) mod send_buffer; pub(crate) mod spaces; +pub use spaces::PathStatus; #[cfg(fuzzing)] pub use spaces::Retransmits; #[cfg(not(fuzzing))] @@ -584,8 +583,9 @@ impl Connection { return Err(PathError::RemoteCidsExhausted); } - let path = self.create_path(path_id, network_path, now, None); - path.status.local_update(initial_status); + self.create_network_path(path_id, network_path, now, None); + let pns = self.spaces[SpaceKind::Data].for_path(path_id); + pns.status.local_update(initial_status); Ok(path_id) } @@ -794,8 +794,10 @@ impl Connection { /// Gets the local [`PathStatus`] for a known [`PathId`] pub fn path_status(&self, path_id: PathId) -> Result { - self.path(path_id) - .map(PathData::local_status) + self.spaces[SpaceKind::Data] + .number_spaces + .get(&path_id) + .map(|pns| pns.local_status()) .ok_or(ClosedPath { _private: () }) } @@ -817,28 +819,32 @@ impl Connection { if !self.is_multipath_negotiated() { return Err(SetPathStatusError::MultipathNotNegotiated); } - let path = self - .path_mut(path_id) + let pns = self.spaces[SpaceKind::Data] + .number_spaces + .get_mut(&path_id) .ok_or(SetPathStatusError::ClosedPath)?; - let prev = match path.status.local_update(status) { + let prev = match pns.status.local_update(status) { Some(prev) => { - self.spaces[SpaceId::Data] + self.spaces[SpaceKind::Data] .pending .path_status .insert(path_id); prev } - None => path.local_status(), + None => pns.local_status(), }; Ok(prev) } - /// Returns the remote path status + /// Returns the remote path status. // TODO(flub): Probably should also be some kind of path event? Not even sure if I like // this as an API, but for now it allows me to write a test easily. - // TODO(flub): Technically this should be a Result>? + // TODO(flub): Technically this should be a Result>? pub fn remote_path_status(&self, path_id: PathId) -> Option { - self.path(path_id).and_then(|path| path.remote_status()) + self.spaces[SpaceKind::Data] + .number_spaces + .get(&path_id) + .and_then(|pns| pns.remote_status()) } /// Sets the max_idle_timeout for a specific path. @@ -934,7 +940,7 @@ impl Connection { /// Creates the [`PathData`] for a new [`PathId`]. /// /// Called for incoming packets as well as when opening a new path locally. - fn create_path( + fn create_network_path( &mut self, path_id: PathId, network_path: FourTuple, @@ -1154,10 +1160,12 @@ impl Connection { fn scheduling_info(&self, path_id: PathId) -> PathSchedulingInfo { // Such a space is preferred for SpaceKind::Data frames. let have_validated_status_available_space = self.paths.iter().any(|(path_id, path)| { + // pns can never be None here, that would be a logical error. + let pns = self.spaces[SpaceKind::Data].number_spaces.get(path_id); self.remote_cids.contains_key(path_id) && !self.abandoned_paths.contains(path_id) && path.data.validated - && path.data.local_status() == PathStatus::Available + && pns.map(|pns| pns.local_status()).unwrap_or_default() == PathStatus::Available }); // Such a space is able to send SpaceKind::Data frames. @@ -1172,7 +1180,10 @@ impl Connection { let is_abandoned = self.abandoned_paths.contains(&path_id); let path_data = self.path_data(path_id); let validated = path_data.validated; - let status = path_data.local_status(); + + // pns can never be None here, that would be a logical error. + let pns = self.spaces[SpaceKind::Data].number_spaces.get(&path_id); + let status = pns.map(|pns| pns.local_status()).unwrap_or_default(); // This is the core packet scheduling, whether this space ID may send // SpaceKind::Data frames. @@ -4354,7 +4365,7 @@ impl Connection { if self.side().is_server() && !self.abandoned_paths.contains(&path_id) { // Only the client is allowed to open paths - self.create_path(path_id, network_path, now, pn); + self.create_network_path(path_id, network_path, now, pn); } if self.paths.contains_key(&path_id) { self.on_packet_authenticated( @@ -5840,12 +5851,14 @@ impl Connection { let is_client = self.side().is_client(); let immediate_ack_allowed = self.peer_supports_ack_frequency(); - for (path_id, path) in self.paths.iter_mut() { + for path_id in self.spaces[SpaceKind::Data].number_spaces.keys() { if self.abandoned_paths.contains(path_id) { continue; } open_paths += 1; + let path = self.paths.get_mut(path_id).expect("PathData missing"); + // Read the network path BEFORE clearing local_ip, so the hint can // check which interface the path was using. let network_path = path.data.network_path; @@ -5872,7 +5885,7 @@ impl Connection { if attempt_to_recover { recoverable_paths.push((*path_id, remote)); } else { - non_recoverable_paths.push((*path_id, remote, path.data.local_status())) + non_recoverable_paths.push((*path_id, remote)); } } @@ -5886,12 +5899,16 @@ impl Connection { // We prefer closing paths first unless we identify this is the last open path. let open_first = open_paths == non_recoverable_paths.len(); - for (path_id, remote, status) in non_recoverable_paths.into_iter() { + for (path_id, remote) in non_recoverable_paths.into_iter() { let network_path = FourTuple { remote, local_ip: None, /* allow the local ip to be discovered */ }; - + let status = self.spaces[SpaceKind::Data] + .number_spaces + .get(&path_id) + .map(|pns| pns.local_status()) + .expect("spaces iterated above"); if open_first && let Err(e) = self.open_path(network_path, status, now) { if self.side().is_client() { debug!(%e, "Failed to open new path for network change"); @@ -6342,13 +6359,13 @@ impl Connection { let Some(path_id) = space.pending.path_status.pop_first() else { break; }; - let Some(path) = self.paths.get(&path_id).map(|path_state| &path_state.data) else { + let Some(pns) = space.number_spaces.get(&path_id) else { trace!(%path_id, "discarding queued path status for unknown path"); continue; }; - let seq = path.status.seq(); - match path.local_status() { + let seq = pns.status.seq(); + match pns.local_status() { PathStatus::Available => { let frame = frame::PathStatusAvailable { path_id, @@ -7033,18 +7050,18 @@ impl Connection { /// Handle new path status information: PATH_STATUS_AVAILABLE, PATH_STATUS_BACKUP fn on_path_status(&mut self, path_id: PathId, status: PathStatus, status_seq_no: VarInt) { - if let Some(path) = self.paths.get_mut(&path_id) { - path.data.status.remote_update(status, status_seq_no); + if let Some(pns) = self.spaces[SpaceKind::Data].number_spaces.get_mut(&path_id) { + pns.status.remote_update(status, status_seq_no); + self.events.push_back( + PathEvent::RemoteStatus { + id: path_id, + status, + } + .into(), + ); } else { debug!("PATH_STATUS_AVAILABLE received unknown path {:?}", path_id); } - self.events.push_back( - PathEvent::RemoteStatus { - id: path_id, - status, - } - .into(), - ); } /// Returns the maximum [`PathId`] to be used for sending in this connection. diff --git a/noq-proto/src/connection/paths.rs b/noq-proto/src/connection/paths.rs index efe71812f6..00cc7aa483 100644 --- a/noq-proto/src/connection/paths.rs +++ b/noq-proto/src/connection/paths.rs @@ -2,10 +2,10 @@ use std::{cmp, net::SocketAddr}; use identity_hash::IntMap; use thiserror::Error; -use tracing::{debug, trace}; +use tracing::trace; use super::{ - PathStats, SpaceKind, + PathStats, PathStatus, SpaceKind, mtud::MtuDiscovery, pacing::Pacer, spaces::{PacketNumberSpace, SentPacket}, @@ -224,8 +224,6 @@ pub(super) struct PathData { /// Observed address frame with the largest sequence number received from the peer on this /// path. pub(super) last_observed_addr_report: Option, - /// The QUIC-MULTIPATH path status - pub(super) status: PathStatusState, /// Number of the first packet sent on this path /// /// With RFC9000 ยง9 style migration (i.e. not multipath) the PathId does not change and @@ -337,7 +335,6 @@ impl PathData { in_flight: InFlight::new(), pending: PathRetransmits::default(), last_observed_addr_report: None, - status: Default::default(), first_packet: None, pto_count: 0, idle_timeout: config.default_path_max_idle_timeout, @@ -385,7 +382,6 @@ impl PathData { in_flight: InFlight::new(), pending: PathRetransmits::default(), last_observed_addr_report: None, - status: prev.status.clone(), first_packet: None, pto_count: 0, idle_timeout: prev.idle_timeout, @@ -646,14 +642,6 @@ impl PathData { } } - pub(crate) fn remote_status(&self) -> Option { - self.status.remote_status.map(|(_seq, status)| status) - } - - pub(crate) fn local_status(&self) -> PathStatus { - self.status.local_status - } - /// Tag uniquely identifying a path in a connection. /// /// When a migration happens on the same [`PathId`] we still detect a change in the @@ -978,69 +966,6 @@ impl InFlight { } } -/// State for QUIC-MULTIPATH PATH_STATUS_AVAILABLE and PATH_STATUS_BACKUP frames -#[derive(Debug, Clone, Default)] -pub(super) struct PathStatusState { - /// The local status - local_status: PathStatus, - /// Local sequence number, for both PATH_STATUS_AVAILABLE and PATH_STATUS_BACKUP - /// - /// This is the number of the *next* path status frame to be sent. - local_seq: VarInt, - /// The status set by the remote - remote_status: Option<(VarInt, PathStatus)>, -} - -impl PathStatusState { - /// To be called on received PATH_STATUS_AVAILABLE/PATH_STATUS_BACKUP frames - pub(super) fn remote_update(&mut self, status: PathStatus, seq: VarInt) { - if self.remote_status.is_some_and(|(curr, _)| curr >= seq) { - return trace!(%seq, "ignoring path status update"); - } - - let prev = self.remote_status.replace((seq, status)).map(|(_, s)| s); - if prev != Some(status) { - debug!(?status, ?seq, "remote changed path status"); - } - } - - /// Updates the local status - /// - /// If the local status changed, the previous value is returned - pub(super) fn local_update(&mut self, status: PathStatus) -> Option { - if self.local_status == status { - return None; - } - - self.local_seq = self.local_seq.saturating_add(1u8); - Some(std::mem::replace(&mut self.local_status, status)) - } - - pub(crate) fn seq(&self) -> VarInt { - self.local_seq - } -} - -/// The QUIC-MULTIPATH path status -/// -/// See section "3.3 Path Status Management": -/// -#[cfg_attr(test, derive(test_strategy::Arbitrary))] -#[derive(Debug, Copy, Clone, Default, PartialEq, Eq)] -pub enum PathStatus { - /// Paths marked with as available will be used when scheduling packets - /// - /// If multiple paths are available, packets will be scheduled on whichever has - /// capacity. - #[default] - Available, - /// Paths marked as backup will only be used if there are no available paths - /// - /// If the max_idle_timeout is specified the path will be kept alive so that it does not - /// expire. - Backup, -} - /// Application events about paths #[derive(Debug, Clone, PartialEq, Eq)] #[non_exhaustive] diff --git a/noq-proto/src/connection/spaces.rs b/noq-proto/src/connection/spaces.rs index cbed253a2e..f74429912f 100644 --- a/noq-proto/src/connection/spaces.rs +++ b/noq-proto/src/connection/spaces.rs @@ -8,7 +8,7 @@ use std::{ use rand::{CryptoRng, RngExt}; use rustc_hash::{FxHashMap, FxHashSet}; use sorted_index_buffer::SortedIndexBuffer; -use tracing::trace; +use tracing::{debug, trace}; use super::{PathId, paths::PathResponses, paths::PathRetransmits}; use crate::{ @@ -238,6 +238,11 @@ pub(super) struct PacketNumberSpace { /// reported. This is not required by the spec, and in the future might be changed for /// simply requiring a first ack'd packet. pub(super) open_status: OpenStatus, + /// The QUIC-MULTIPATH path status. + /// + /// This field is unused for the Initial and Handshake spaces, and when multipath is not + /// negotiated. + pub(super) status: PathStatusState, /// Highest received packet number, if any pub(super) largest_received_packet_number: Option, @@ -306,6 +311,7 @@ impl PacketNumberSpace { }; Self { open_status: OpenStatus::default(), + status: PathStatusState::default(), largest_received_packet_number: None, next_packet_number: 0, largest_acked_packet_pn: None, @@ -336,6 +342,7 @@ impl PacketNumberSpace { }; Self { open_status: OpenStatus::default(), + status: PathStatusState::default(), largest_received_packet_number: None, next_packet_number: 0, largest_acked_packet_pn: None, @@ -358,6 +365,14 @@ impl PacketNumberSpace { } } + pub(crate) fn remote_status(&self) -> Option { + self.status.remote_status.map(|(_seq, status)| status) + } + + pub(crate) fn local_status(&self) -> PathStatus { + self.status.local_status + } + /// Get the next outgoing packet number in this space /// /// In the Data space, the connection's [`PacketNumberFilter`] must be used rather than calling @@ -502,6 +517,69 @@ pub(super) enum OpenStatus { Informed, } +/// State for QUIC-MULTIPATH PATH_STATUS_AVAILABLE and PATH_STATUS_BACKUP frames +#[derive(Debug, Clone, Default)] +pub(super) struct PathStatusState { + /// The local status + local_status: PathStatus, + /// Local sequence number, for both PATH_STATUS_AVAILABLE and PATH_STATUS_BACKUP + /// + /// This is the number of the *next* path status frame to be sent. + local_seq: VarInt, + /// The status set by the remote + remote_status: Option<(VarInt, PathStatus)>, +} + +impl PathStatusState { + /// To be called on received PATH_STATUS_AVAILABLE/PATH_STATUS_BACKUP frames + pub(super) fn remote_update(&mut self, status: PathStatus, seq: VarInt) { + if self.remote_status.is_some_and(|(curr, _)| curr >= seq) { + return trace!(%seq, "ignoring path status update"); + } + + let prev = self.remote_status.replace((seq, status)).map(|(_, s)| s); + if prev != Some(status) { + debug!(?status, ?seq, "remote changed path status"); + } + } + + /// Updates the local status + /// + /// If the local status changed, the previous value is returned + pub(super) fn local_update(&mut self, status: PathStatus) -> Option { + if self.local_status == status { + return None; + } + + self.local_seq = self.local_seq.saturating_add(1u8); + Some(std::mem::replace(&mut self.local_status, status)) + } + + pub(crate) fn seq(&self) -> VarInt { + self.local_seq + } +} + +/// The QUIC-MULTIPATH path status +/// +/// See section "3.3 Path Status Management": +/// +#[cfg_attr(test, derive(test_strategy::Arbitrary))] +#[derive(Debug, Copy, Clone, Default, PartialEq, Eq)] +pub enum PathStatus { + /// Paths marked with as available will be used when scheduling packets + /// + /// If multiple paths are available, packets will be scheduled on whichever has + /// capacity. + #[default] + Available, + /// Paths marked as backup will only be used if there are no available paths + /// + /// If the max_idle_timeout is specified the path will be kept alive so that it does not + /// expire. + Backup, +} + /// Represents one or more packets subject to retransmission #[derive(Debug, Clone)] pub(super) struct SentPacket {