diff --git a/.github/workflows/publish-crates.yml b/.github/workflows/publish-crates.yml index d4db01f3..bec84ba4 100644 --- a/.github/workflows/publish-crates.yml +++ b/.github/workflows/publish-crates.yml @@ -16,7 +16,7 @@ jobs: release: runs-on: ubuntu-latest permissions: - contents: read + contents: write id-token: write steps: @@ -33,7 +33,7 @@ jobs: uses: rust-lang/crates-io-auth-action@v1 id: auth - - name: Release workspace crates + - name: Release publishable workspace crates shell: bash env: CARGO_REGISTRY_TOKEN: ${{ steps.auth.outputs.token }} @@ -62,16 +62,6 @@ jobs: dquic ) - if [[ "$mode" == "dry-run" ]]; then - publish_args=(cargo publish --dry-run --locked) - for package in "${packages[@]}"; do - publish_args+=(-p "$package") - done - echo "dry-run publish packages: ${packages[*]}" - "${publish_args[@]}" - exit 0 - fi - cargo metadata --format-version 1 > "$RUNNER_TEMP/workspace-metadata.json" package_versions="$( PACKAGES="$(printf '%s\n' "${packages[@]}")" RUNNER_TEMP="$RUNNER_TEMP" python3 - <<'PY' @@ -90,7 +80,7 @@ jobs: PY )" - missing_packages=() + packages_to_publish=() while IFS=$'\t' read -r crate_name crate_version; do [ -n "$crate_name" ] || continue crate_state="$( @@ -100,37 +90,100 @@ jobs: import urllib.request name, version = sys.argv[1], sys.argv[2] - url = f"https://crates.io/api/v1/crates/{name}/{version}" - request = urllib.request.Request(url, headers={"User-Agent": "genmeta dquic publish workflow"}) + headers = {"User-Agent": "genmeta dquic publish workflow"} + version_url = f"https://crates.io/api/v1/crates/{name}/{version}" + version_request = urllib.request.Request(version_url, headers=headers) try: - with urllib.request.urlopen(request, timeout=20) as response: + with urllib.request.urlopen(version_request, timeout=20) as response: if response.status == 200: - print("published") + print("published_version") else: raise SystemExit(f"unexpected crates.io status for {name} {version}: {response.status}") except urllib.error.HTTPError as error: if error.code == 404: - print("missing") + crate_url = f"https://crates.io/api/v1/crates/{name}" + crate_request = urllib.request.Request(crate_url, headers=headers) + try: + with urllib.request.urlopen(crate_request, timeout=20) as response: + if response.status == 200: + print("missing_version") + else: + raise SystemExit(f"unexpected crates.io crate status for {name}: {response.status}") + except urllib.error.HTTPError as crate_error: + if crate_error.code == 404: + print("missing_crate") + else: + raise else: raise PY )" - if [[ "$crate_state" == "published" ]]; then + if [[ "$crate_state" == "published_version" ]]; then echo "skip $crate_name $crate_version (already on crates.io)" - else + elif [[ "$crate_state" == "missing_version" ]]; then echo "publish $crate_name $crate_version" - missing_packages+=("$crate_name") + packages_to_publish+=("$crate_name") + else + echo "skip $crate_name $crate_version (crate not yet initialized on crates.io)" fi done <<< "$package_versions" - if [[ "${#missing_packages[@]}" -eq 0 ]]; then - echo "all selected packages are already published" + if [[ "${#packages_to_publish[@]}" -eq 0 ]]; then + echo "no already-initialized crates need a new crates.io release" exit 0 fi - publish_args=(cargo publish --locked) - for package in "${missing_packages[@]}"; do + if [[ "$mode" == "dry-run" ]]; then + publish_args=(cargo publish --dry-run --locked) + echo "dry-run publish packages: ${packages_to_publish[*]}" + else + publish_args=(cargo publish --locked) + echo "publish packages: ${packages_to_publish[*]}" + fi + + for package in "${packages_to_publish[@]}"; do publish_args+=(-p "$package") done "${publish_args[@]}" + + - name: Create GitHub Release + if: github.ref_type == 'tag' && startsWith(github.ref_name, 'v') + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + tag_ref="refs/tags/$GITHUB_REF_NAME" + git fetch --force origin "$tag_ref:$tag_ref" + tag_object="$(git rev-parse "$tag_ref^{tag}" 2>/dev/null || true)" + target_commit="$(git rev-parse "$tag_ref^{commit}")" + notes_file="$RUNNER_TEMP/release-notes.md" + git for-each-ref "$tag_ref" --format='%(contents)' > "$notes_file" + if [ ! -s "$notes_file" ]; then + git log -1 --format=%B "$target_commit" > "$notes_file" + fi + { + printf '\n## Authentication and provenance\n\n' + printf -- '- Tag: `%s`\n' "$GITHUB_REF_NAME" + if [ -n "$tag_object" ]; then + printf -- '- Annotated tag object: `%s`\n' "$tag_object" + else + printf -- '- Tag object: lightweight tag\n' + fi + printf -- '- Target commit: `%s`\n' "$target_commit" + printf -- '- Workflow run: %s/%s/actions/runs/%s\n' \ + "$GITHUB_SERVER_URL" "$GITHUB_REPOSITORY" "$GITHUB_RUN_ID" + printf -- '- Workflow attempt: `%s`\n' "$GITHUB_RUN_ATTEMPT" + printf -- '- Published by: GitHub Actions `%s` workflow\n' "$GITHUB_WORKFLOW" + } >> "$notes_file" + + if gh release view "$GITHUB_REF_NAME" --repo "$GITHUB_REPOSITORY" >/dev/null 2>&1; then + echo "github release $GITHUB_REF_NAME already exists" + exit 0 + fi + + gh release create "$GITHUB_REF_NAME" \ + --repo "$GITHUB_REPOSITORY" \ + --verify-tag \ + --title "$GITHUB_REF_NAME" \ + --notes-file "$notes_file" diff --git a/Cargo.toml b/Cargo.toml index b8450feb..b28be0aa 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -106,7 +106,7 @@ qresolve = { path = "./qresolve", version = "0.5.1" } qrecovery = { path = "./qrecovery", version = "0.5.1" } qtraversal = { path = "./qtraversal", version = "0.5.1" } qcongestion = { path = "./qcongestion", version = "0.5.1" } -qconnection = { path = "./qconnection", version = "0.5.1" } +qconnection = { path = "./qconnection", version = "0.5.2" } dquic = { path = "./dquic", version = "0.5.1" } h3-shim = { path = "./h3-shim", version = "0.5.1" } diff --git a/dquic/tests/echo.rs b/dquic/tests/echo.rs index f3df64b8..2c46373e 100644 --- a/dquic/tests/echo.rs +++ b/dquic/tests/echo.rs @@ -130,6 +130,30 @@ fn shutdown() -> Result<(), BoxError> { }) } +#[test] +fn application_close_then_drop_does_not_strand_connection() -> Result<(), BoxError> { + run(async { + let router = Arc::new(QuicRouter::default()); + let (listeners, server_task) = + launch_echo_server(router.clone(), server_parameters()).await?; + let _server_task = AbortOnDropHandle::new(tokio::spawn(server_task)); + + let server_addr = get_server_addr(&listeners); + let client = launch_test_client(router, client_parameters()); + let connection = client + .connected_to_with_source("localhost", [(Source::System, server_addr.into())]) + .await?; + + send_and_verify_echo(&connection, TEST_DATA).await?; + connection.close("client done", 0)?; + drop(connection); + + tokio::time::sleep(Duration::from_millis(100)).await; + listeners.shutdown(); + Ok(()) + }) +} + #[test] fn idle_timeout() -> Result<(), BoxError> { run(async { diff --git a/qconnection/Cargo.toml b/qconnection/Cargo.toml index 242678f6..c4bfc1aa 100644 --- a/qconnection/Cargo.toml +++ b/qconnection/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "qconnection" -version = "0.5.1" +version = "0.5.2" edition.workspace = true description = "Encapsulation of QUIC connections, a part of dquic" readme.workspace = true diff --git a/qconnection/src/builder.rs b/qconnection/src/builder.rs index 7bd76df4..ebe26ee7 100644 --- a/qconnection/src/builder.rs +++ b/qconnection/src/builder.rs @@ -551,7 +551,7 @@ impl PendingConnection { interfaces: self.interfaces, locations: self.locations, rcvd_pkt_q: self.rcvd_pkt_q, - conn_state, + conn_state: conn_state.clone(), idle_config: ArcIdleConfig::new(max_idle_timeout, self.defer_idle_timeout), paths: ArcPathContexts::new(self.tx_wakers.clone(), event_broker.clone()), send_lock: self.send_lock, @@ -575,8 +575,10 @@ impl PendingConnection { spawn_tls_handshake(&components, self.tx_wakers.clone()); spawn_deliver_and_parse(&components); - let connection = Arc::new(Connection { + let connection = Arc::new_cyclic(|weak_self| Connection { state: Ok(components).into(), + conn_state: conn_state.clone(), + weak_self: weak_self.clone(), qlog_span, tracing_span, }); @@ -739,17 +741,34 @@ fn spawn_drive_connection( ) { tokio::spawn( async move { + let mut retained_connection: Option> = None; while let Some(event) = events.recv().await { - let Some(connection) = weak_connection.upgrade() else { + let Some(connection) = retained_connection + .as_ref() + .cloned() + .or_else(|| weak_connection.upgrade()) + else { break; }; + match event { Event::Handshaked => {} - Event::Failed(quic_error) => _ = connection.enter_closing(quic_error), - Event::ApplicationClose(_app_error) => {} - Event::Closed(ccf) => _ = connection.enter_draining(ccf), + Event::Failed(quic_error) => { + retained_connection.get_or_insert_with(|| connection.clone()); + _ = connection.enter_closing(quic_error); + } + Event::ApplicationClose(_app_error) => { + retained_connection.get_or_insert_with(|| connection.clone()); + } + Event::Closed(ccf) => { + retained_connection.get_or_insert_with(|| connection.clone()); + _ = connection.enter_draining(ccf); + } Event::StatelessReset => {} - Event::Terminated => {} + Event::Terminated => { + retained_connection.take(); + break; + } } } } diff --git a/qconnection/src/events.rs b/qconnection/src/events.rs index a9385a89..2afb402f 100644 --- a/qconnection/src/events.rs +++ b/qconnection/src/events.rs @@ -5,7 +5,6 @@ use qbase::{ error::{AppError, QuicError}, frame::ConnectionCloseFrame, }; -use qevent::quic::connectivity::BaseConnectionStates; use tokio::sync::mpsc; use crate::state::ArcConnState; @@ -54,25 +53,12 @@ impl EmitEvent for ArcEventBroker { return; } } - Event::Failed(error) => { - if self.conn_state.enter_closing(error).is_none() { - return; - } - } - Event::ApplicationClose(error) => { - if self.conn_state.enter_closing(error).is_none() { - return; - } - } - Event::Closed(ccf) => { - if self.conn_state.enter_draining(ccf).is_none() { + Event::Failed(_) | Event::ApplicationClose(_) | Event::Closed(_) => {} + Event::Terminated => { + if self.conn_state.enter_closed().is_none() { return; } } - Event::Terminated => { - let terminated_state = BaseConnectionStates::Closed; - self.conn_state.update(terminated_state.into()); - } Event::StatelessReset => todo!("unsupported"), }; tracing::debug!(target: "quic", new_state = ?event, "connection state changed"); @@ -88,9 +74,15 @@ impl EmitEvent for mpsc::UnboundedSender { #[cfg(test)] mod tests { + use qbase::{ + error::{ErrorFrameType, ErrorKind, QuicError}, + frame::ConnectionCloseFrame, + varint::VarInt, + }; use tokio::sync::mpsc; use super::*; + use crate::state; #[test] fn test_emit_event() { @@ -98,4 +90,47 @@ mod tests { tx.emit(Event::Handshaked); assert_eq!(rx.try_recv().unwrap(), Event::Handshaked); } + + #[test] + fn failed_event_is_forwarded_without_entering_closing() { + let conn_state = ArcConnState::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + let broker = ArcEventBroker::new(conn_state.clone(), tx); + let error = QuicError::with_default_fty(ErrorKind::NoViablePath, "no path"); + + broker.emit(Event::Failed(error.clone())); + + assert_eq!(rx.try_recv().unwrap(), Event::Failed(error)); + assert_ne!(conn_state.current(), Some(state::CLOSING)); + } + + #[test] + fn closed_event_is_forwarded_without_entering_draining() { + let conn_state = ArcConnState::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + let broker = ArcEventBroker::new(conn_state.clone(), tx); + let ccf = ConnectionCloseFrame::new_quic( + ErrorKind::NoViablePath, + ErrorFrameType::Ext(VarInt::from_u32(0)), + "", + ); + + broker.emit(Event::Closed(ccf.clone())); + + assert_eq!(rx.try_recv().unwrap(), Event::Closed(ccf)); + assert_ne!(conn_state.current(), Some(state::DRAINING)); + } + + #[test] + fn application_close_event_is_forwarded_without_entering_closing() { + let conn_state = ArcConnState::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + let broker = ArcEventBroker::new(conn_state.clone(), tx); + let error = qbase::error::AppError::new(VarInt::from_u32(0), ""); + + broker.emit(Event::ApplicationClose(error.clone())); + + assert_eq!(rx.try_recv().unwrap(), Event::ApplicationClose(error)); + assert_ne!(conn_state.current(), Some(state::CLOSING)); + } } diff --git a/qconnection/src/lib.rs b/qconnection/src/lib.rs index ce79868f..241c49e1 100644 --- a/qconnection/src/lib.rs +++ b/qconnection/src/lib.rs @@ -49,7 +49,7 @@ use std::{ future::Future, io, net::SocketAddr, - sync::{Arc, RwLock, atomic::AtomicBool}, + sync::{Arc, RwLock, Weak, atomic::AtomicBool}, }; pub use ::{qbase, qdatagram, qevent, qinterface, qrecovery, qtraversal}; @@ -351,8 +351,14 @@ impl Components { { let pto_duration = self.paths.max_pto_duration().unwrap_or_default(); let event_broker = self.event_broker.clone(); + let rcvd_pkt_q = self.rcvd_pkt_q.clone(); + let paths = self.paths.clone(); + let local_cids = self.cid_registry.local.clone(); async move { tokio::time::sleep(pto_duration).await; + rcvd_pkt_q.close_all(); + paths.clear(); + local_cids.clear(); event_broker.emit(Event::Terminated); } } @@ -370,14 +376,11 @@ impl Components { .in_current_span(), ); } - // No need to send packets, just clear the paths. - false => { - // TODO: check the remote of close spaces - self.paths.clear(); - } + // The send lock denies close packets for silent refusal paths. + false => self.paths.clear(), } - Termination::closing(error, self.cid_registry.local, self.rcvd_pkt_q) + Termination::closing(error, self.cid_registry.local, self.rcvd_pkt_q, self.paths) } pub fn enter_draining(self, ccf: ConnectionCloseFrame) -> Termination { @@ -396,8 +399,10 @@ impl Components { { let pto_duration = self.paths.max_pto_duration().unwrap_or_default(); let event_broker = self.event_broker.clone(); + let local_cids = self.cid_registry.local.clone(); async move { tokio::time::sleep(pto_duration).await; + local_cids.clear(); event_broker.emit(Event::Terminated); } } @@ -405,43 +410,75 @@ impl Components { .in_current_span(), ); - match self.send_lock.is_permitted() { - // If permitted, we can send ccf packets. - true => { - let terminator = Arc::new(Terminator::new(ccf, &self)); - tokio::spawn( - async move { self.spaces.send_ccf_packets(terminator.as_ref()).await } - .instrument_in_current() - .in_current_span(), - ); - } - // No need to send packets, just clear the paths. - false => { - self.paths.clear(); - } - } - // No need to receive packets, just close all queues. self.rcvd_pkt_q.close_all(); + self.paths.clear(); Termination::draining(error, self.cid_registry.local) } } pub struct Connection { state: RwLock>, + conn_state: ArcConnState, + weak_self: Weak, qlog_span: qevent::telemetry::Span, tracing_span: tracing::Span, } impl Connection { + fn keep_alive_until_closed(&self) { + let Some(connection) = self.weak_self.upgrade() else { + return; + }; + let span = connection.tracing_span.clone(); + tokio::spawn(tracing::Instrument::instrument( + async move { + connection.conn_state.closed().await; + }, + span, + )); + } + + fn close_by_application(&self, error: AppError, keep_alive: bool) -> Result<(), Error> { + let _span = (self.qlog_span.enter(), self.tracing_span.enter()); + let event_broker = { + let mut conn = self.state.write().unwrap(); + match conn.as_ref() { + Ok(core_conn) => { + if self.conn_state.enter_closing(&error).is_none() { + return Err(error.into()); + } + + let event_broker = core_conn.event_broker.clone(); + *conn = Err(core_conn.clone().enter_closing(error.clone().into())); + event_broker + } + Err(termination) => return Err(termination.error()), + } + }; + + if keep_alive { + self.keep_alive_until_closed(); + } + event_broker.emit(Event::ApplicationClose(error)); + Ok(()) + } + // called by event pub fn enter_closing(&self, error: QuicError) -> Result<(), Error> { let _span = (self.qlog_span.enter(), self.tracing_span.enter()); let mut conn = self.state.write().unwrap(); - let core_conn = conn.as_ref().map_err(|t| t.error())?; + match conn.as_ref() { + Ok(core_conn) => { + if self.conn_state.enter_closing(&error).is_none() { + return Err(error.into()); + } - *conn = Err(core_conn.clone().enter_closing(error.into())); - Ok(()) + *conn = Err(core_conn.clone().enter_closing(error.into())); + Ok(()) + } + Err(termination) => Err(termination.error()), + } } /// Close the connection with application close frame. @@ -449,22 +486,17 @@ impl Connection { /// Return error if the connection is already closed. #[doc(alias = "application_close")] pub fn close(&self, reason: impl Into>, code: u64) -> Result<(), Error> { - let _span = (self.qlog_span.enter(), self.tracing_span.enter()); - let mut conn = self.state.write().unwrap(); - let core_conn = conn.as_ref().map_err(|t| t.error())?; - let error_code = code.try_into().expect("application error code overflow"); let error = AppError::new(error_code, reason); - let event = Event::ApplicationClose(error.clone()); - core_conn.event_broker.emit(event); - *conn = Err(core_conn.clone().enter_closing(error.into())); - - Ok(()) + self.close_by_application(error, true) } pub fn enter_draining(&self, ccf: ConnectionCloseFrame) -> bool { let _span = (self.qlog_span.enter(), self.tracing_span.enter()); let mut conn = self.state.write().unwrap(); + if self.conn_state.enter_draining(&ccf).is_none() { + return false; + } match conn.as_mut() { Ok(core_conn) => { *conn = Err(core_conn.clone().enter_draining(ccf)); @@ -672,7 +704,8 @@ impl Connection { impl Drop for Connection { fn drop(&mut self) { let _span = self.tracing_span.enter(); - if self.validate().is_ok() && self.close("", 0).is_ok() { + let error = AppError::new(0_u64.try_into().expect("zero app error code"), ""); + if self.close_by_application(error, false).is_ok() { #[cfg(debug_assertions)] tracing::warn!(target: "quic", "connection is still active when dropped, close it automatically."); #[cfg(not(debug_assertions))] diff --git a/qconnection/src/state.rs b/qconnection/src/state.rs index 63557dab..dfe7127c 100644 --- a/qconnection/src/state.rs +++ b/qconnection/src/state.rs @@ -28,6 +28,7 @@ pub struct ArcConnState { state: Arc, handshaked: Arc>, terminated: Arc>, + closed: Arc>, } impl Default for ArcConnState { @@ -36,6 +37,7 @@ impl Default for ArcConnState { state: Default::default(), handshaked: Arc::new(SetOnce::new()), terminated: Arc::new(SetOnce::new()), + closed: Arc::new(SetOnce::new()), } } } @@ -154,6 +156,14 @@ impl ArcConnState { None } + pub fn enter_closed(&self) -> Option { + if let Some(old_state) = self.update(BaseConnectionStates::Closed.into()) { + _ = self.closed.set(()); + return Some(old_state); + } + None + } + pub fn handshaked(&self) -> impl Future> + Send + use<> { let handshaked = self.handshaked.clone(); let terminated = self.terminated.clone(); @@ -174,6 +184,15 @@ impl ArcConnState { .in_current_span() } + pub fn closed(&self) -> impl Future + Send + use<> { + let closed = self.closed.clone(); + async move { + closed.wait().await; + } + .instrument_in_current() + .in_current_span() + } + pub fn current(&self) -> Option { decode(self.state.load(Ordering::Acquire)) } @@ -221,3 +240,30 @@ pub const DRAINING: QlogConnectionState = pub const CLOSED: QlogConnectionState = QlogConnectionState::Granular(GranularConnectionStates::Closed); + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn closed_waits_for_final_closed_state() { + let conn_state = ArcConnState::new(); + let waiter = conn_state.closed(); + + assert!( + tokio::time::timeout(std::time::Duration::from_millis(1), waiter) + .await + .is_err() + ); + + assert!(conn_state.enter_closed().is_some()); + + tokio::time::timeout(std::time::Duration::from_secs(1), conn_state.closed()) + .await + .expect("closed wait should complete after enter_closed"); + assert_eq!( + conn_state.current(), + Some(BaseConnectionStates::Closed.into()) + ); + } +} diff --git a/qconnection/src/termination.rs b/qconnection/src/termination.rs index 6fb17e82..1c51025d 100644 --- a/qconnection/src/termination.rs +++ b/qconnection/src/termination.rs @@ -26,8 +26,6 @@ use tokio::time::Instant; use crate::{ArcLocalCids, Components, path::ArcPathContexts}; /// Keep a few states to support sending packets with ccf. -/// -/// when it is dropped all paths will be destroyed pub struct Terminator { last_recv_time: Mutex, rcvd_packets: AtomicUsize, @@ -37,12 +35,6 @@ pub struct Terminator { paths: ArcPathContexts, } -impl Drop for Terminator { - fn drop(&mut self) { - self.paths.clear(); - } -} - impl ProductHeader for Terminator { fn new_header(&self) -> Result { let (Some(dcid), Some(scid)) = (self.dcid, self.scid) else { @@ -137,7 +129,10 @@ impl Terminator { #[derive(Clone)] enum State { - Closing(Arc), + Closing { + rcvd_pkt_q: Arc, + paths: ArcPathContexts, + }, Draining, } @@ -151,11 +146,16 @@ pub struct Termination { } impl Termination { - pub fn closing(error: Error, local_cids: ArcLocalCids, state: Arc) -> Self { + pub fn closing( + error: Error, + local_cids: ArcLocalCids, + rcvd_pkt_q: Arc, + paths: ArcPathContexts, + ) -> Self { Self { error, _local_cids: local_cids, - state: State::Closing(state), + state: State::Closing { rcvd_pkt_q, paths }, } } @@ -174,11 +174,80 @@ impl Termination { // Close packets queues, dont send and receive any more packets. pub fn enter_draining(&mut self) -> bool { match mem::replace(&mut self.state, State::Draining) { - State::Closing(rcvd_pkt_q) => { + State::Closing { rcvd_pkt_q, paths } => { rcvd_pkt_q.close_all(); + paths.clear(); true } - _ => false, + State::Draining => false, } } } + +#[cfg(test)] +mod tests { + use std::sync::Arc; + + use qbase::{ + cid::GenUniqueCid, + error::{Error, ErrorKind, QuicError}, + net::tx::ArcSendWakers, + }; + use qinterface::component::route::{QuicRouter, RcvdPacketQueue}; + use tokio::sync::mpsc; + + use super::*; + use crate::{ + ArcLocalCids, ArcReliableFrameDeque, events::ArcEventBroker, path::ArcPathContexts, + state::ArcConnState, + }; + + fn test_local_cids() -> ArcLocalCids { + let queue = Arc::new(RcvdPacketQueue::new()); + let tx_wakers = ArcSendWakers::default(); + let reliable = ArcReliableFrameDeque::with_capacity_and_wakers(8, tx_wakers); + let registry = Arc::new(QuicRouter::default()).registry_on_issuing_scid(queue, reliable); + let initial_scid = registry.gen_unique_cid(); + ArcLocalCids::new(initial_scid, registry) + } + + fn test_paths() -> ArcPathContexts { + let tx_wakers = ArcSendWakers::default(); + let (tx, _rx) = mpsc::unbounded_channel(); + let broker = ArcEventBroker::new(ArcConnState::new(), tx); + ArcPathContexts::new(tx_wakers, broker) + } + + fn test_error() -> Error { + QuicError::with_default_fty(ErrorKind::NoViablePath, "closed").into() + } + + #[tokio::test] + async fn enter_draining_closes_closing_packet_queues() { + let rcvd_pkt_q = Arc::new(RcvdPacketQueue::new()); + let packets = rcvd_pkt_q.one_rtt().clone(); + let paths = test_paths(); + let mut termination = + Termination::closing(test_error(), test_local_cids(), rcvd_pkt_q, paths); + + assert!(termination.enter_draining()); + + assert!( + tokio::time::timeout(std::time::Duration::from_secs(1), packets.recv()) + .await + .expect("closed queue should wake") + .is_none() + ); + } + + #[test] + fn enter_draining_is_idempotent() { + let rcvd_pkt_q = Arc::new(RcvdPacketQueue::new()); + let paths = test_paths(); + let mut termination = + Termination::closing(test_error(), test_local_cids(), rcvd_pkt_q, paths); + + assert!(termination.enter_draining()); + assert!(!termination.enter_draining()); + } +}