mirror of
https://github.com/n0-computer/noq.git
synced 2026-09-23 11:45:25 +00:00
Abandon a path (#114)
Abandons a path, issues new MAX_PATH_ID and CIDs, closes connection if last path is abandoned. Removes accepted reset tokens when a path is abandoned.
This commit is contained in:
Floris Bruynooghe
committed by
GitHub
GitHub
parent
0dc50edf68
commit
7ed66d6a89
@@ -580,27 +580,59 @@ impl Connection {
|
||||
|
||||
/// Closes a path by sending a PATH_ABANDON frame
|
||||
///
|
||||
/// This will not allow closing the last path.
|
||||
/// This will not allow closing the last path. It does allow closing paths which have
|
||||
/// not yet been opened, as e.g. is the case when receiving a PATH_ABANDON from the peer
|
||||
/// for a path that was never opened locally.
|
||||
pub fn close_path(
|
||||
&mut self,
|
||||
now: Instant,
|
||||
path_id: PathId,
|
||||
_error_code: VarInt,
|
||||
error_code: VarInt,
|
||||
) -> Result<(), ClosePathError> {
|
||||
// TODO(flub): We are allowed to close paths that have not yet been opened to use up
|
||||
// already issued path IDs.
|
||||
let _path = self
|
||||
if self.abandoned_paths.contains(&path_id) || Some(path_id) > self.max_path_id() {
|
||||
return Err(ClosePathError::ClosedPath);
|
||||
}
|
||||
if self
|
||||
.paths
|
||||
.get_mut(&path_id)
|
||||
.ok_or(ClosePathError::ClosedPath)?;
|
||||
if self.paths.len() < 2 {
|
||||
.keys()
|
||||
.filter(|&id| !self.abandoned_paths.contains(id))
|
||||
.count()
|
||||
< 2
|
||||
{
|
||||
return Err(ClosePathError::LastOpenPath);
|
||||
}
|
||||
// - send PATH_ABANDON
|
||||
// - retire received CIDs for this path - this means no more sending anything using
|
||||
// it.
|
||||
// - Now set a timer for 3 * PTO to remove the rest of the state? and set the reset
|
||||
// token.
|
||||
todo!()
|
||||
|
||||
// Send PATH_ABANDON
|
||||
self.spaces[SpaceId::Data]
|
||||
.pending
|
||||
.path_abandon
|
||||
.insert(path_id, error_code.into());
|
||||
|
||||
// Consider remotely issued CIDs as retired.
|
||||
// Technically we don't have to do this just yet. We only need to do this *after*
|
||||
// the ABANDON_PATH frame is sent, allowing us to still send it on the
|
||||
// to-be-abandoned path. However it is recommended to send it on another path, and
|
||||
// we do not allow abandoning the last path anyway.
|
||||
// We don't fully retire these CIDs. We remove them so we can no longer send using
|
||||
// them, but the reset tokens are still registered with the endpoint. They will be
|
||||
// removed when the connection is cleaned up, which is right because we might still
|
||||
// receive stateless resets.
|
||||
self.rem_cids.remove(&path_id);
|
||||
self.endpoint_events
|
||||
.push_back(EndpointEventInner::RetireResetToken(path_id));
|
||||
|
||||
self.abandoned_paths.insert(path_id);
|
||||
|
||||
self.set_max_path_id(now, self.local_max_path_id.saturating_add(1u8));
|
||||
|
||||
// The peer MUST respond with a corresponding PATH_ABANDON frame. If not, this timer
|
||||
// expires.
|
||||
self.timers.set(
|
||||
Timer::PathNotAbandoned(path_id),
|
||||
now + self.pto_max_path(SpaceId::Data),
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Gets the [`PathData`] for a known [`PathId`].
|
||||
@@ -843,19 +875,24 @@ impl Connection {
|
||||
// check if there is at least one active CID to use for sending
|
||||
let Some(remote_cid) = self.rem_cids.get(&path_id).map(CidQueue::active) else {
|
||||
let err = PathError::RemoteCidsExhausted;
|
||||
debug!(?err, %path_id, "no active CID for path");
|
||||
self.events.push_back(Event::Path(PathEvent::LocallyClosed {
|
||||
id: path_id,
|
||||
error: err,
|
||||
}));
|
||||
// Locally we should have refused to open this path, the remote should have
|
||||
// given us CIDs for this path before opening it.
|
||||
self.close_path(path_id, TransportErrorCode::NO_CID_AVAILABLE.into())
|
||||
.ok();
|
||||
self.spaces[SpaceId::Data]
|
||||
.pending
|
||||
.path_cids_blocked
|
||||
.push(path_id);
|
||||
if !self.abandoned_paths.contains(&path_id) {
|
||||
debug!(?err, %path_id, "no active CID for path");
|
||||
self.events.push_back(Event::Path(PathEvent::LocallyClosed {
|
||||
id: path_id,
|
||||
error: err,
|
||||
}));
|
||||
// Locally we should have refused to open this path, the remote should
|
||||
// have given us CIDs for this path before opening it. So we can always
|
||||
// abandon this here.
|
||||
self.close_path(now, path_id, TransportErrorCode::NO_CID_AVAILABLE.into())
|
||||
.ok();
|
||||
self.spaces[SpaceId::Data]
|
||||
.pending
|
||||
.path_cids_blocked
|
||||
.push(path_id);
|
||||
} else {
|
||||
trace!(?path_id, "remote CIDs retired for abandoned path");
|
||||
}
|
||||
|
||||
match self.paths.keys().find(|&&next| next > path_id) {
|
||||
Some(next_path_id) => {
|
||||
@@ -864,7 +901,7 @@ impl Connection {
|
||||
?space_id,
|
||||
?path_id,
|
||||
?next_path_id,
|
||||
"nothing to send on path"
|
||||
"no CIDs to send on path"
|
||||
);
|
||||
path_id = *next_path_id;
|
||||
space_id = SpaceId::Data;
|
||||
@@ -884,7 +921,7 @@ impl Connection {
|
||||
trace!(
|
||||
?space_id,
|
||||
?path_id,
|
||||
"nothing to send on path, no more paths"
|
||||
"no CIDs to send on path, no more paths"
|
||||
);
|
||||
break;
|
||||
}
|
||||
@@ -1119,8 +1156,9 @@ impl Connection {
|
||||
}
|
||||
}
|
||||
builder.finish_and_track(now, self, path_id, sent_frames, pad_datagram);
|
||||
if space_id == self.highest_space && path_id == *self.paths.keys().max().unwrap() {
|
||||
// Don't send another close packet
|
||||
if space_id == self.highest_space {
|
||||
// Don't send another close packet. Even with multipath we only send
|
||||
// CONNECTION_CLOSE on a single path since we expect our paths to work.
|
||||
self.close = false;
|
||||
// `CONNECTION_CLOSE` is the final packet
|
||||
break;
|
||||
@@ -1654,7 +1692,7 @@ impl Connection {
|
||||
Timer::PathIdle(path_id) => {
|
||||
// TODO(flub): TransportErrorCode::NO_ERROR but where's the API to get
|
||||
// that into a VarInt?
|
||||
self.close_path(path_id, TransportErrorCode::NO_ERROR.into())
|
||||
self.close_path(now, path_id, TransportErrorCode::NO_ERROR.into())
|
||||
.ok();
|
||||
}
|
||||
Timer::KeepAlive => {
|
||||
@@ -1719,6 +1757,28 @@ impl Connection {
|
||||
.pending_acks
|
||||
.on_max_ack_delay_timeout()
|
||||
}
|
||||
Timer::PathAbandoned(path_id) => {
|
||||
// The path was abandoned and 3*PTO has expired since. Clean up all
|
||||
// remaining state and install stateless reset token.
|
||||
if let Some(loc_cid_state) = self.local_cid_state.remove(&path_id) {
|
||||
let (min_seq, max_seq) = loc_cid_state.active_seq();
|
||||
for seq in min_seq..=max_seq {
|
||||
self.endpoint_events
|
||||
.push_back(EndpointEventInner::RetireConnectionId(
|
||||
now, path_id, seq, false,
|
||||
));
|
||||
}
|
||||
}
|
||||
self.paths.remove(&path_id);
|
||||
self.spaces[SpaceId::Data].number_spaces.remove(&path_id);
|
||||
}
|
||||
Timer::PathNotAbandoned(path_id) => {
|
||||
// The peer failed to respond with a PATH_ABANDON when we sent such a
|
||||
// frame.
|
||||
warn!(?path_id, "missing PATH_ABANDON from peer");
|
||||
// TODO(flub): What should the error code be?
|
||||
self.close(now, 0u8.into(), "peer ignored PATH_ABANDON frame".into());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -3348,7 +3408,7 @@ impl Connection {
|
||||
if let Some(token) = params.stateless_reset_token {
|
||||
let remote = self.path_data(path_id).remote;
|
||||
self.endpoint_events
|
||||
.push_back(EndpointEventInner::ResetToken(remote, token));
|
||||
.push_back(EndpointEventInner::ResetToken(path_id, remote, token));
|
||||
}
|
||||
self.handle_peer_params(params, loc_cid, rem_cid)?;
|
||||
self.issue_first_cids(now);
|
||||
@@ -3705,6 +3765,9 @@ impl Connection {
|
||||
retire_prior_to = frame.retire_prior_to,
|
||||
);
|
||||
let path_id = frame.path_id.unwrap_or_default();
|
||||
// TODO(flub): We should only accept CIDs if path_id < self.max_path_id()
|
||||
// because otherwise someone could attack us by sending us lots of
|
||||
// CIDs.
|
||||
let rem_cids = self
|
||||
.rem_cids
|
||||
.entry(path_id)
|
||||
@@ -3740,7 +3803,7 @@ impl Connection {
|
||||
));
|
||||
}
|
||||
pending_retired.extend(retired.map(|seq| (path_id, seq)));
|
||||
self.set_reset_token(remote, reset_token);
|
||||
self.set_reset_token(path_id, remote, reset_token);
|
||||
}
|
||||
Err(InsertError::ExceedsLimit) => {
|
||||
return Err(TransportError::CONNECTION_ID_LIMIT_ERROR(""));
|
||||
@@ -3861,8 +3924,31 @@ impl Connection {
|
||||
migration_observed_addr = Some(observed)
|
||||
}
|
||||
}
|
||||
Frame::PathAbandon(_) => {
|
||||
// TODO(@divma): jump ship?
|
||||
Frame::PathAbandon(frame::PathAbandon {
|
||||
path_id,
|
||||
error_code,
|
||||
}) => {
|
||||
// TODO(flub): don't really know which error code to use here.
|
||||
match self.close_path(now, path_id, error_code.into()) {
|
||||
Ok(()) => {
|
||||
trace!(?path_id, "peer abandoned path");
|
||||
}
|
||||
Err(ClosePathError::LastOpenPath) => {
|
||||
trace!("peer abandoned last path, closing connection");
|
||||
// TODO(flub): which error code?
|
||||
self.close(
|
||||
now,
|
||||
0u8.into(),
|
||||
Bytes::from_static(b"last path abandoned by peer"),
|
||||
);
|
||||
}
|
||||
Err(ClosePathError::ClosedPath) => {
|
||||
trace!(?path_id, "peer abandoned already closed path");
|
||||
}
|
||||
}
|
||||
let delay = self.pto(SpaceId::Data, path_id) * 3;
|
||||
self.timers.set(Timer::PathAbandoned(path_id), now + delay);
|
||||
self.timers.stop(Timer::PathNotAbandoned(path_id));
|
||||
}
|
||||
Frame::PathAvailable(info) => {
|
||||
if self.is_multipath_negotiated() {
|
||||
@@ -4062,11 +4148,11 @@ impl Connection {
|
||||
|
||||
/// Switch to a previously unused remote connection ID, if possible
|
||||
fn update_rem_cid(&mut self, path_id: PathId) {
|
||||
let (reset_token, retired) =
|
||||
match self.rem_cids.get_mut(&path_id).and_then(|cids| cids.next()) {
|
||||
Some(x) => x,
|
||||
None => return,
|
||||
};
|
||||
let Some((reset_token, retired)) =
|
||||
self.rem_cids.get_mut(&path_id).and_then(|cids| cids.next())
|
||||
else {
|
||||
return;
|
||||
};
|
||||
|
||||
// Retire the current remote CID and any CIDs we had to skip.
|
||||
self.spaces[SpaceId::Data]
|
||||
@@ -4074,21 +4160,29 @@ impl Connection {
|
||||
.retire_cids
|
||||
.extend(retired.map(|seq| (path_id, seq)));
|
||||
let remote = self.path_data(path_id).remote;
|
||||
self.set_reset_token(remote, reset_token);
|
||||
self.set_reset_token(path_id, remote, reset_token);
|
||||
}
|
||||
|
||||
/// Sends this reset token to the endpoint.
|
||||
/// Sends this reset token to the endpoint
|
||||
///
|
||||
/// The endpoint needs to have reset-tokens for past connections so that it can still
|
||||
/// use those for stateless resets when the connection state is dropped. See RFC 9000
|
||||
/// section 10.3. Stateless Reset.
|
||||
/// The endpoint needs to know the reset tokens issued by the peer, so that if the peer
|
||||
/// sends a reset token it knows to route it to this connection. See RFC 9000 section
|
||||
/// 10.3. Stateless Reset.
|
||||
///
|
||||
/// Reset tokens are different for each path, the endpoint identifies paths by peer
|
||||
/// socket address however, not by path ID.
|
||||
fn set_reset_token(&mut self, remote: SocketAddr, reset_token: ResetToken) {
|
||||
fn set_reset_token(&mut self, path_id: PathId, remote: SocketAddr, reset_token: ResetToken) {
|
||||
self.endpoint_events
|
||||
.push_back(EndpointEventInner::ResetToken(remote, reset_token));
|
||||
self.peer_params.stateless_reset_token = Some(reset_token);
|
||||
.push_back(EndpointEventInner::ResetToken(path_id, remote, reset_token));
|
||||
|
||||
// During the handshake the server sends a reset token in the transport
|
||||
// parameters. When we are the client and we receive the reset token during the
|
||||
// handshake we want this to affect our peer transport parameters.
|
||||
// TODO(flub): Pretty sure this is pointless, the entire params is overwritten
|
||||
// shortly after this was called. And then the params don't have this anymore.
|
||||
if path_id == PathId::ZERO {
|
||||
self.peer_params.stateless_reset_token = Some(reset_token);
|
||||
}
|
||||
}
|
||||
|
||||
/// Issue an initial set of connection IDs to the peer upon connection
|
||||
@@ -4387,7 +4481,7 @@ impl Connection {
|
||||
&& space_id == SpaceId::Data
|
||||
&& frame::PathAbandon::SIZE_BOUND <= buf.remaining_mut()
|
||||
{
|
||||
let Some((path_id, error_code)) = space.pending.path_abandon.pop() else {
|
||||
let Some((path_id, error_code)) = space.pending.path_abandon.pop_first() else {
|
||||
break;
|
||||
};
|
||||
frame::PathAbandon {
|
||||
@@ -4396,6 +4490,12 @@ impl Connection {
|
||||
}
|
||||
.encode(buf);
|
||||
self.stats.frame_tx.path_abandon += 1;
|
||||
trace!(?path_id, "PATH_ABANDON");
|
||||
sent.retransmits
|
||||
.get_or_create()
|
||||
.path_abandon
|
||||
.entry(path_id)
|
||||
.or_insert(error_code);
|
||||
}
|
||||
|
||||
// PATH_AVAILABLE & PATH_BACKUP
|
||||
@@ -4678,11 +4778,11 @@ impl Connection {
|
||||
let delay = delay_micros >> ack_delay_exp.into_inner();
|
||||
|
||||
if send_path_acks {
|
||||
trace!("PATH_ACK {:?}, Delay = {}us", ranges, delay_micros);
|
||||
trace!("PATH_ACK {path_id:?} {ranges:?}, Delay = {delay_micros}us");
|
||||
frame::PathAck::encode(path_id, delay as _, ranges, ecn, buf);
|
||||
stats.frame_tx.path_acks += 1;
|
||||
} else {
|
||||
trace!("ACK {:?}, Delay = {}us", ranges, delay_micros);
|
||||
trace!("ACK {ranges:?}, Delay = {delay_micros}us");
|
||||
frame::Ack::encode(delay as _, ranges, ecn, buf);
|
||||
stats.frame_tx.acks += 1;
|
||||
}
|
||||
@@ -4737,11 +4837,10 @@ impl Connection {
|
||||
self.idle_timeout =
|
||||
negotiate_max_idle_timeout(self.config.max_idle_timeout, Some(params.max_idle_timeout));
|
||||
trace!("negotiated max idle timeout {:?}", self.idle_timeout);
|
||||
let path_id = PathId(0);
|
||||
|
||||
if let Some(ref info) = params.preferred_address {
|
||||
// During the handshake PathId(0) exists.
|
||||
self.rem_cids.get_mut(&path_id).expect("not yet abandoned").insert(frame::NewConnectionId {
|
||||
self.rem_cids.get_mut(&PathId::ZERO).expect("not yet abandoned").insert(frame::NewConnectionId {
|
||||
path_id: None,
|
||||
sequence: 1,
|
||||
id: info.connection_id,
|
||||
@@ -4751,8 +4850,8 @@ impl Connection {
|
||||
.expect(
|
||||
"preferred address CID is the first received, and hence is guaranteed to be legal",
|
||||
);
|
||||
let remote = self.path_data(path_id).remote;
|
||||
self.set_reset_token(remote, info.stateless_reset_token);
|
||||
let remote = self.path_data(PathId::ZERO).remote;
|
||||
self.set_reset_token(PathId::ZERO, remote, info.stateless_reset_token);
|
||||
}
|
||||
self.ack_frequency.peer_max_ack_delay = get_max_ack_delay(¶ms);
|
||||
|
||||
@@ -4769,7 +4868,7 @@ impl Connection {
|
||||
self.peer_params = params;
|
||||
let peer_max_udp_payload_size =
|
||||
u16::try_from(self.peer_params.max_udp_payload_size.into_inner()).unwrap_or(u16::MAX);
|
||||
self.path_data_mut(path_id)
|
||||
self.path_data_mut(PathId::ZERO)
|
||||
.mtud
|
||||
.on_peer_max_udp_payload_size_received(peer_max_udp_payload_size);
|
||||
}
|
||||
|
||||
@@ -38,6 +38,7 @@ impl PathId {
|
||||
/// The 0 path id.
|
||||
pub const ZERO: Self = Self(0);
|
||||
|
||||
/// The number of bytes this [`PathId`] uses when encoded as a [`VarInt`]
|
||||
pub(crate) fn size(&self) -> usize {
|
||||
VarInt(self.0 as u64).size()
|
||||
}
|
||||
|
||||
@@ -543,7 +543,7 @@ pub struct Retransmits {
|
||||
/// that; consider what such a change would mean for implementing `BitOrAssign` on Self.
|
||||
pub(super) new_tokens: Vec<SocketAddr>,
|
||||
/// Paths which need to be abandoned
|
||||
pub(super) path_abandon: Vec<(PathId, TransportErrorCode)>,
|
||||
pub(super) path_abandon: BTreeMap<PathId, TransportErrorCode>,
|
||||
/// If a [`frame::PathAvailable`] and [`frame::PathBackup`] need to be sent for a path
|
||||
pub(super) path_status: BTreeSet<PathId>,
|
||||
/// If a PATH_CIDS_BLOCKED frame needs to be sent for a path
|
||||
@@ -575,7 +575,7 @@ impl Retransmits {
|
||||
}
|
||||
|
||||
impl ::std::ops::BitOrAssign for Retransmits {
|
||||
fn bitor_assign(&mut self, rhs: Self) {
|
||||
fn bitor_assign(&mut self, mut rhs: Self) {
|
||||
// We reduce in-stream head-of-line blocking by queueing retransmits before other data for
|
||||
// STREAM and CRYPTO frames.
|
||||
self.max_data |= rhs.max_data;
|
||||
@@ -594,7 +594,7 @@ impl ::std::ops::BitOrAssign for Retransmits {
|
||||
self.handshake_done |= rhs.handshake_done;
|
||||
self.observed_addr |= rhs.observed_addr;
|
||||
self.new_tokens.extend_from_slice(&rhs.new_tokens);
|
||||
self.path_abandon.extend_from_slice(&rhs.path_abandon);
|
||||
self.path_abandon.append(&mut rhs.path_abandon);
|
||||
self.max_path_id |= rhs.max_path_id;
|
||||
self.paths_blocked |= rhs.paths_blocked;
|
||||
}
|
||||
|
||||
@@ -30,6 +30,10 @@ pub(crate) enum Timer {
|
||||
PushNewCid,
|
||||
/// When to send an immediate ACK if there are unacked ack-eliciting packets of the peer
|
||||
MaxAckDelay(PathId),
|
||||
/// When to clean up state for an abandoned path
|
||||
PathAbandoned(PathId),
|
||||
/// When the peer fails to confirm abandoning the path
|
||||
PathNotAbandoned(PathId),
|
||||
}
|
||||
|
||||
/// Keeps track of the nearest timeout for each `Timer`
|
||||
|
||||
+25
-10
@@ -105,21 +105,29 @@ impl Endpoint {
|
||||
NeedIdentifiers(path_id, now, n) => {
|
||||
return Some(self.send_new_identifiers(path_id, now, ch, n));
|
||||
}
|
||||
ResetToken(remote, token) => {
|
||||
if let Some(old) = self.connections[ch].reset_token.replace((remote, token)) {
|
||||
ResetToken(path_id, remote, token) => {
|
||||
if let Some(old) = self.connections[ch]
|
||||
.reset_token
|
||||
.insert(path_id, (remote, token))
|
||||
{
|
||||
self.index.connection_reset_tokens.remove(old.0, old.1);
|
||||
}
|
||||
if self.index.connection_reset_tokens.insert(remote, token, ch) {
|
||||
warn!("duplicate reset token");
|
||||
}
|
||||
}
|
||||
RetireResetToken(path_id) => {
|
||||
if let Some(old) = self.connections[ch].reset_token.remove(&path_id) {
|
||||
self.index.connection_reset_tokens.remove(old.0, old.1);
|
||||
}
|
||||
}
|
||||
RetireConnectionId(now, path_id, seq, allow_more_cids) => {
|
||||
if let Some(cid) = self.connections[ch]
|
||||
.loc_cids
|
||||
.get_mut(&path_id)
|
||||
.and_then(|pcid| pcid.cids.remove(&seq))
|
||||
{
|
||||
trace!(?path_id, "peer retired CID {}: {}", seq, cid);
|
||||
trace!(?path_id, "local CID retired {}: {}", seq, cid);
|
||||
self.index.retire(cid);
|
||||
if allow_more_cids {
|
||||
return Some(self.send_new_identifiers(path_id, now, ch, 1));
|
||||
@@ -266,6 +274,7 @@ impl Endpoint {
|
||||
}
|
||||
}
|
||||
|
||||
/// Builds a stateless reset packet to respond with
|
||||
fn stateless_reset(
|
||||
&mut self,
|
||||
now: Instant,
|
||||
@@ -854,7 +863,7 @@ impl Endpoint {
|
||||
loc_cids: FxHashMap::from_iter([(PathId(0), path_cids)]),
|
||||
addresses,
|
||||
side,
|
||||
reset_token: None,
|
||||
reset_token: Default::default(),
|
||||
});
|
||||
debug_assert_eq!(id, ch.0, "connection handle allocation out of sync");
|
||||
|
||||
@@ -1085,8 +1094,8 @@ impl ConnectionIndex {
|
||||
self.incoming_connection_remotes.remove(&conn.addresses);
|
||||
self.outgoing_connection_remotes
|
||||
.remove(&conn.addresses.remote);
|
||||
if let Some((remote, token)) = conn.reset_token {
|
||||
self.connection_reset_tokens.remove(remote, token);
|
||||
for (remote, token) in conn.reset_token.values() {
|
||||
self.connection_reset_tokens.remove(*remote, *token);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1122,7 +1131,7 @@ impl ConnectionIndex {
|
||||
self.connection_reset_tokens
|
||||
.get(addresses.remote, &data[data.len() - RESET_TOKEN_SIZE..])
|
||||
.cloned()
|
||||
.map(|ch| RouteDatagramTo::Connection(ch, PathId(0)))
|
||||
.map(|ch| RouteDatagramTo::Connection(ch, PathId::ZERO))
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1137,9 +1146,15 @@ pub(crate) struct ConnectionMeta {
|
||||
/// bother keeping it up to date.
|
||||
addresses: FourTuple,
|
||||
side: Side,
|
||||
/// Reset token provided by the peer for the CID we're currently sending to, and the address
|
||||
/// being sent to
|
||||
reset_token: Option<(SocketAddr, ResetToken)>,
|
||||
/// Reset tokens provided by the peer for CIDs we're currently sending to
|
||||
///
|
||||
/// Since each reset token is for a CID, it is also for a fixed remote address which is
|
||||
/// also stored. This allows us to look up which reset tokens we might expect from a
|
||||
/// given remote address, see [`ResetTokenTable`].
|
||||
///
|
||||
/// Each path has its own active CID. We use the [`PathId`] as a unique index, allowing
|
||||
/// us to retire the reset token when a path is abandoned.
|
||||
reset_token: FxHashMap<PathId, (SocketAddr, ResetToken)>,
|
||||
}
|
||||
|
||||
/// Local connection IDs for a single path
|
||||
|
||||
@@ -53,10 +53,18 @@ impl EndpointEvent {
|
||||
pub(crate) enum EndpointEventInner {
|
||||
/// The connection has been drained
|
||||
Drained,
|
||||
/// The reset token and/or address eligible for generating resets has been updated
|
||||
ResetToken(SocketAddr, ResetToken),
|
||||
/// The connection has a new active reset token
|
||||
///
|
||||
/// Whenever the connection switches to a new remote CID issued by the peer, it also
|
||||
/// switches the matching reset token that can be used to abort this connection. This
|
||||
/// event provides a new reset token for the active remote CID.
|
||||
ResetToken(PathId, SocketAddr, ResetToken),
|
||||
/// Retire the reset token for a path, without replacing it with a new one
|
||||
RetireResetToken(PathId),
|
||||
/// The connection needs connection identifiers
|
||||
NeedIdentifiers(PathId, Instant, u64),
|
||||
/// Retire a locally issued CID
|
||||
///
|
||||
/// Stop routing connection ID for this sequence number to the connection
|
||||
/// When `bool == true`, a new connection ID will be issued to peer
|
||||
RetireConnectionId(Instant, PathId, u64, bool),
|
||||
|
||||
@@ -147,7 +147,7 @@ fn path_close_last_path() {
|
||||
|
||||
let client_conn = pair.client_conn_mut(client_ch);
|
||||
let err = client_conn
|
||||
.close_path(PathId::ZERO, 0u8.into())
|
||||
.close_path(Instant::now(), PathId::ZERO, 0u8.into())
|
||||
.err()
|
||||
.unwrap();
|
||||
assert!(matches!(err, ClosePathError::LastOpenPath));
|
||||
@@ -378,3 +378,66 @@ fn open_path() {
|
||||
Event::Path(crate::PathEvent::Opened { id }) if id == path_id
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn close_path() {
|
||||
let _guard = subscribe();
|
||||
let (mut pair, client_ch, _server_ch) = multipath_pair();
|
||||
|
||||
let server_addr = pair.server.addr;
|
||||
let path_id = pair
|
||||
.client_conn_mut(client_ch)
|
||||
.open_path(server_addr, PathStatus::Available, Instant::now())
|
||||
.unwrap();
|
||||
pair.drive();
|
||||
assert_ne!(path_id, PathId::ZERO);
|
||||
|
||||
let stats0 = pair.client_conn_mut(client_ch).stats();
|
||||
assert_eq!(stats0.frame_tx.path_abandon, 0);
|
||||
assert_eq!(stats0.frame_rx.path_abandon, 0);
|
||||
assert_eq!(stats0.frame_tx.max_path_id, 0);
|
||||
assert_eq!(stats0.frame_rx.max_path_id, 0);
|
||||
|
||||
info!("closing path 0");
|
||||
pair.client_conn_mut(client_ch)
|
||||
.close_path(Instant::now(), PathId::ZERO, 0u8.into())
|
||||
.unwrap();
|
||||
pair.drive();
|
||||
|
||||
let stats1 = pair.client_conn_mut(client_ch).stats();
|
||||
assert_eq!(stats1.frame_tx.path_abandon, 1);
|
||||
assert_eq!(stats1.frame_rx.path_abandon, 1);
|
||||
assert_eq!(stats1.frame_tx.max_path_id, 1);
|
||||
assert_eq!(stats1.frame_rx.max_path_id, 1);
|
||||
assert!(stats1.frame_tx.path_new_connection_id > stats0.frame_tx.path_new_connection_id);
|
||||
assert!(stats1.frame_rx.path_new_connection_id > stats0.frame_rx.path_new_connection_id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn close_last_path() {
|
||||
let _guard = subscribe();
|
||||
let (mut pair, client_ch, server_ch) = multipath_pair();
|
||||
|
||||
let server_addr = pair.server.addr;
|
||||
let path_id = pair
|
||||
.client_conn_mut(client_ch)
|
||||
.open_path(server_addr, PathStatus::Available, Instant::now())
|
||||
.unwrap();
|
||||
pair.drive();
|
||||
assert_ne!(path_id, PathId::ZERO);
|
||||
|
||||
info!("client closes path 0");
|
||||
pair.client_conn_mut(client_ch)
|
||||
.close_path(Instant::now(), PathId::ZERO, 0u8.into())
|
||||
.unwrap();
|
||||
|
||||
info!("server closes path 1");
|
||||
pair.server_conn_mut(server_ch)
|
||||
.close_path(Instant::now(), PathId(1), 0u8.into())
|
||||
.unwrap();
|
||||
|
||||
pair.drive();
|
||||
|
||||
assert!(pair.server_conn_mut(server_ch).is_closed());
|
||||
assert!(pair.client_conn_mut(client_ch).is_closed());
|
||||
}
|
||||
|
||||
+3
-1
@@ -93,7 +93,9 @@ impl Path {
|
||||
let (on_path_close_send, on_path_close_recv) = oneshot::channel();
|
||||
{
|
||||
let mut state = self.conn.state.lock("close_path");
|
||||
state.inner.close_path(self.id, error_code)?;
|
||||
state
|
||||
.inner
|
||||
.close_path(crate::Instant::now(), self.id, error_code)?;
|
||||
state.close_path.insert(self.id, on_path_close_send);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user