From 7ed66d6a89cc6d6ae192b79ca6f20fa033e034a8 Mon Sep 17 00:00:00 2001 From: Floris Bruynooghe Date: Tue, 15 Jul 2025 18:05:04 +0200 Subject: [PATCH] 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. --- quinn-proto/src/connection/mod.rs | 213 ++++++++++++++++++++------- quinn-proto/src/connection/paths.rs | 1 + quinn-proto/src/connection/spaces.rs | 6 +- quinn-proto/src/connection/timer.rs | 4 + quinn-proto/src/endpoint.rs | 35 +++-- quinn-proto/src/shared.rs | 12 +- quinn-proto/src/tests/multipath.rs | 65 +++++++- quinn/src/path.rs | 4 +- 8 files changed, 266 insertions(+), 74 deletions(-) diff --git a/quinn-proto/src/connection/mod.rs b/quinn-proto/src/connection/mod.rs index d931ff3de..83f4a12cd 100644 --- a/quinn-proto/src/connection/mod.rs +++ b/quinn-proto/src/connection/mod.rs @@ -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); } diff --git a/quinn-proto/src/connection/paths.rs b/quinn-proto/src/connection/paths.rs index a67714b12..c3a150584 100644 --- a/quinn-proto/src/connection/paths.rs +++ b/quinn-proto/src/connection/paths.rs @@ -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() } diff --git a/quinn-proto/src/connection/spaces.rs b/quinn-proto/src/connection/spaces.rs index 3eed5107f..a35a1ade9 100644 --- a/quinn-proto/src/connection/spaces.rs +++ b/quinn-proto/src/connection/spaces.rs @@ -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, /// Paths which need to be abandoned - pub(super) path_abandon: Vec<(PathId, TransportErrorCode)>, + pub(super) path_abandon: BTreeMap, /// If a [`frame::PathAvailable`] and [`frame::PathBackup`] need to be sent for a path pub(super) path_status: BTreeSet, /// 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; } diff --git a/quinn-proto/src/connection/timer.rs b/quinn-proto/src/connection/timer.rs index 0370f17fc..f27ec3fa9 100644 --- a/quinn-proto/src/connection/timer.rs +++ b/quinn-proto/src/connection/timer.rs @@ -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` diff --git a/quinn-proto/src/endpoint.rs b/quinn-proto/src/endpoint.rs index fe65cf5db..e417e2bc6 100644 --- a/quinn-proto/src/endpoint.rs +++ b/quinn-proto/src/endpoint.rs @@ -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, } /// Local connection IDs for a single path diff --git a/quinn-proto/src/shared.rs b/quinn-proto/src/shared.rs index c2382a3dd..3f93d3ef8 100644 --- a/quinn-proto/src/shared.rs +++ b/quinn-proto/src/shared.rs @@ -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), diff --git a/quinn-proto/src/tests/multipath.rs b/quinn-proto/src/tests/multipath.rs index d9afafd36..2ebd99016 100644 --- a/quinn-proto/src/tests/multipath.rs +++ b/quinn-proto/src/tests/multipath.rs @@ -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()); +} diff --git a/quinn/src/path.rs b/quinn/src/path.rs index 5a3e62694..168282c9a 100644 --- a/quinn/src/path.rs +++ b/quinn/src/path.rs @@ -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); }