AWS SDK

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/cell/h2/request.rs

@@ -0,1 +0,639 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! HTTP/2 request authority and two-sided completion.
 //!
 //! A cell creates [`H2Activation`] only after reserving one prospective request
 //! against an exact generation. The activation owns the transient Hyper sender
 //! and the right to convert that prospective count into an accepted request.
 //! Dropping it before acceptance cancels the prospective count.
 //!
 //! Hyper acceptance creates one [`H2RequestClaim`]. The request upload and
 //! response lifetime hold distinct guard types so transposing the two sides is
 //! a type error. Either side may finish first. The claim releases the accepted
 //! generation count only after both sides finish, including drop and error
 //! paths.
 //!
 //! The claim mutex is never held while locking cell or connection state.
-+
 use super::super::super::connection::{ConnectionInfo, ConnectionState, DispatchGuard};
 use super::super::super::partition::PartitionId;
 use super::super::{OriginCell, WaiterId};
 use super::{H2CloseHandle, H2GenerationId, H2RouteId, H2Sender};
 use crate::sync::{Arc, Mutex, Weak};
-+
 /// Sender and connection cloned while a prospective request is cell-owned.
 pub(super) struct H2ActivationResources {
     /// Transient sender clone for one activation.
     pub(super) sender: H2Sender,
     /// Whether the generation accepted an earlier request.
     pub(super) reused: bool,
     /// Connection retained through prospective dispatch.
     pub(super) connection: Arc<ConnectionState>,
-+
+}
-+
 /// Resources retained before and after dispatch transfer.
 enum H2ActivationState {
     /// Sender and connection still owned by the activation.
     Ready(H2ActivationResources),
     /// Immutable identity retained after the sender leaves the activation.
     Dispatched(H2RequestIdentity),
-+
+}
-+
 /// Values transferred together when an activation begins Hyper dispatch.
 pub(in crate::client::pool) struct H2DispatchParts {
     /// Transient sender clone for one dispatch attempt.
     pub(in crate::client::pool) sender: H2Sender,
     /// Upload-completion guard retained by the request body.
     pub(in crate::client::pool) upload: H2UploadGuard,
     /// Response-lifetime guard retained by the response future or body.
     pub(in crate::client::pool) response: H2ResponseGuard,
-+
+}
-+
 /// Prospective request authority for one exact generation.
 ///
 /// The activation leaves the cell lock with one counted prospective request.
 /// Taking its dispatch parts transfers the sender and creates the two request
 /// guards. Hyper acceptance converts the prospective count to an accepted
 /// claim; rejection or drop cancels it.
 pub(in crate::client::pool) struct H2Activation {
     /// Cell that owns the selected generation.
     connection_cell: Arc<OriginCell>,
     /// Exact generation that owns the prospective count.
     generation: H2GenerationId,
     /// Partition issuing this request.
     request_partition: PartitionId,
     /// Mutually exclusive pre-dispatch resources or post-transfer identity.
     state: Option<H2ActivationState>,
     /// Shared accounting for the upload and response sides.
     claim: Arc<H2RequestClaim>,
     /// Requesting-cell priority released at acceptance or cancellation.
     turn: Option<H2ActivationTurnGuard>,
     /// Whether drop must cancel the prospective generation count.
     active: bool,
-+
+}
-+
 impl std::fmt::Debug for H2Activation {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("H2Activation")
             .field("generation", &self.generation)
             .field(
                 "connection_id",
                 &self.state.as_ref().map(|state| match state {
                     H2ActivationState::Ready(resources) => resources.connection.id(),
                     H2ActivationState::Dispatched(identity) => identity.connection.id(),
                 }),
-+
+            )
             .field("active", &self.active)
             .finish()
-+
+    }
-+
+}
-+
 impl H2Activation {
     /// Builds an activation after the generation records its prospective claim.
     pub(super) fn new(
         connection_cell: Arc<OriginCell>,
         generation: H2GenerationId,
         resources: H2ActivationResources,
         request_partition: PartitionId,
         turn: Option<H2ActivationTurnGuard>,
     ) -> Self {
         let claim = Arc::new(H2RequestClaim {
             connection_cell: Weak::from_arc(&connection_cell),
             generation,
             state: Mutex::new(H2RequestClaimState::Prospective {
                 upload_finished: false,
                 response_finished: false,
             }),
         });
         Self {
             connection_cell,
             generation,
             request_partition,
             state: Some(H2ActivationState::Ready(resources)),
             claim,
             turn,
             active: true,
-+
+        }
-+
+    }
-+
     /// Returns the exact selected generation.
     #[cfg(test)]
     pub(in crate::client::pool) fn generation(&self) -> H2GenerationId {
         self.generation
-+
+    }
-+
     /// Returns whether Hyper previously accepted a request on this generation.
     ///
     /// # Panics
     ///
     /// Panics if dispatch parts were already taken.
     pub(in crate::client::pool) fn is_reused(&self) -> bool {
         match self
             .state
             .as_ref()
             .expect("HTTP/2 activation state missing")
-+
+        {
             H2ActivationState::Ready(resources) => resources.reused,
             H2ActivationState::Dispatched(_) => {
                 panic!("HTTP/2 activation resources already taken")
-+
+            }
-+
+        }
-+
+    }
-+
     /// Returns the selected protocol-neutral connection.
     ///
     /// # Panics
     ///
     /// Panics if dispatch parts were already taken.
     pub(in crate::client::pool) fn connection(&self) -> &Arc<ConnectionState> {
         match self
             .state
             .as_ref()
             .expect("HTTP/2 activation state missing")
-+
+        {
             H2ActivationState::Ready(resources) => &resources.connection,
             H2ActivationState::Dispatched(_) => {
                 panic!("HTTP/2 activation resources already taken")
-+
+            }
-+
+        }
-+
+    }
-+
     /// Transfers the sender and typed request guards into one dispatch attempt.
     ///
     /// # Panics
     ///
     /// Panics if dispatch parts were already taken.
     pub(in crate::client::pool) fn take_dispatch_parts(&mut self) -> H2DispatchParts {
         let state = self.state.take().expect("HTTP/2 activation state missing");
         let H2ActivationState::Ready(resources) = state else {
             self.state = Some(state);
             panic!("HTTP/2 activation dispatch parts already taken");
         };
         let identity = H2RequestIdentity {
             request_partition: self.request_partition,
             connection: resources.connection.info().clone(),
             generation: self.generation,
         };
         self.state = Some(H2ActivationState::Dispatched(identity.clone()));
         H2DispatchParts {
             sender: resources.sender,
             upload: H2UploadGuard::new(self.claim.clone(), Some(identity.clone())),
             response: H2ResponseGuard::new(self.claim.clone(), Some(identity)),
-+
+        }
-+
+    }
-+
     /// Retains a requesting-cell route turn until acceptance or cancellation.
     pub(super) fn attach_peer_turn(
         &mut self,
         requesting_cell: &Arc<OriginCell>,
         route: H2RouteId,
         waiter: WaiterId,
     ) {
         debug_assert_eq!(
             self.request_partition,
             requesting_cell.id().partition(),
             "peer HTTP/2 activation changed requesting partition"
         );
         assert!(
             self.turn.is_none(),
             "HTTP/2 activation acquired two requesting-cell turns"
         );
         self.turn = Some(H2ActivationTurnGuard::peer(requesting_cell, route, waiter));
-+
+    }
-+
     /// Returns close authority for the connection-owning generation.
     pub(in crate::client::pool) fn close_handle(&self) -> H2CloseHandle {
         H2CloseHandle::new(&self.connection_cell, self.generation)
-+
+    }
-+
     /// Converts the prospective reservation after Hyper accepts the request.
     ///
     /// # Panics
     ///
     /// Panics if dispatch parts were not transferred or the prospective
     /// reservation disappeared before acceptance.
     pub(in crate::client::pool) fn accept(mut self, dispatch: DispatchGuard) {
         let Some(H2ActivationState::Dispatched(identity)) = self.state.as_ref() else {
             panic!("HTTP/2 activation accepted before dispatch parts were taken");
         };
         assert!(
             OriginCell::accept_h2_activation(&self.connection_cell, self.generation),
             "prospective HTTP/2 activation disappeared before acceptance"
         );
         self.active = false;
         self.claim.accept(dispatch);
         if let Some(turn) = self.turn.take() {
             turn.release();
-+
+        }
         tracing::trace!(
             connection_id = %identity.connection.id(),
             request_partition = ?identity.request_partition,
             connection_partition = ?identity.connection.owner_partition(),
             origin_scheme = %identity.connection.origin().scheme(),
             origin_host = identity.connection.origin().host(),
             origin_port = ?identity.connection.origin().port(),
             h2_generation = ?identity.generation,
             "HTTP/2 request accepted"
         );
-+
+    }
-+
+}
-+
 impl Drop for H2Activation {
     fn drop(&mut self) {
         if self.active {
             OriginCell::cancel_h2_activation(&self.connection_cell, self.generation);
             self.claim.cancel();
             if let Some(turn) = self.turn.take() {
                 turn.release();
-+
+            }
             match &self.state {
                 Some(H2ActivationState::Ready(resources)) => trace_activation_cancelled(
                     self.request_partition,
                     self.generation,
                     resources.connection.info(),
                 ),
                 Some(H2ActivationState::Dispatched(identity)) => trace_activation_cancelled(
                     identity.request_partition,
                     identity.generation,
                     &identity.connection,
                 ),
                 None => {}
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// One activation-gate turn retained until Hyper acceptance or cancellation.
 pub(super) struct H2ActivationTurnGuard {
     /// Requesting cell whose gate owns this turn.
     cell: Weak<OriginCell>,
     /// Waiter that received the activation opportunity.
     waiter: WaiterId,
     /// Local-generation or peer-route gate identity.
     owner: H2ActivationGateOwner,
-+
+}
-+
 /// Cell transition run when an activation accepts or cancels.
 enum H2ActivationGateOwner {
     /// Gate attached to the connection cell's local generation.
     Local { generation: H2GenerationId },
     /// Gate attached to a requesting cell's peer route.
     Peer { route: H2RouteId },
-+
+}
-+
 impl H2ActivationTurnGuard {
     pub(super) fn local(
         cell: &Arc<OriginCell>,
         generation: H2GenerationId,
         waiter: WaiterId,
     ) -> Self {
         Self {
             cell: Weak::from_arc(cell),
             waiter,
             owner: H2ActivationGateOwner::Local { generation },
-+
+        }
-+
+    }
-+
     fn peer(cell: &Arc<OriginCell>, route: H2RouteId, waiter: WaiterId) -> Self {
         Self {
             cell: Weak::from_arc(cell),
             waiter,
             owner: H2ActivationGateOwner::Peer { route },
-+
+        }
-+
+    }
-+
     fn release(self) {
         if let Some(cell) = self.cell.upgrade() {
             match self.owner {
                 H2ActivationGateOwner::Local { generation } => {
                     OriginCell::release_local_h2_turn(&cell, generation, self.waiter)
-+
+                }
                 H2ActivationGateOwner::Peer { route } => {
                     OriginCell::release_peer_h2_turn(&cell, route, self.waiter)
-+
+                }
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// Shared accounting for one prospective or accepted HTTP/2 request.
 struct H2RequestClaim {
     /// Connection cell updated after both accepted sides finish.
     connection_cell: Weak<OriginCell>,
     /// Exact generation whose request count this claim owns.
     generation: H2GenerationId,
     /// Request phase; never held while locking cell or connection state.
     state: Mutex<H2RequestClaimState>,
-+
+}
-+
 /// Shared request phase and independent completion bits.
 enum H2RequestClaimState {
     /// Generation count reserved before Hyper accepts the request.
     Prospective {
         upload_finished: bool,
         response_finished: bool,
     },
     /// Hyper accepted the request and connection dispatch is retained.
     Accepted {
         dispatch: DispatchGuard,
         upload_finished: bool,
         response_finished: bool,
     },
     /// Both sides finished or prospective dispatch was cancelled.
     Complete,
-+
+}
-+
 impl H2RequestClaim {
     /// Converts prospective request state to accepted state.
     fn accept(&self, dispatch: DispatchGuard) {
         let completed_dispatch = {
             let mut state = self.state.lock();
             let previous = std::mem::replace(&mut *state, H2RequestClaimState::Complete);
             let (upload_finished, response_finished) = match previous {
                 H2RequestClaimState::Prospective {
                     upload_finished,
                     response_finished,
                 } => (upload_finished, response_finished),
                 other @ (H2RequestClaimState::Accepted { .. } | H2RequestClaimState::Complete) => {
                     *state = other;
                     drop(state);
                     panic!("HTTP/2 request claim accepted outside prospective state");
-+
+                }
             };
             if upload_finished && response_finished {
                 Some(dispatch)
             } else {
                 *state = H2RequestClaimState::Accepted {
                     dispatch,
                     upload_finished,
                     response_finished,
                 };
                 None
-+
+            }
         };
         if let Some(dispatch) = completed_dispatch {
             // Dispatch completion takes the connection lifecycle lock. Keep it
             // outside the claim lock so completion cannot nest pool locks.
             drop(dispatch);
             self.release_generation();
-+
+        }
-+
+    }
-+
     /// Cancels request state after the prospective reservation ends.
     fn cancel(&self) {
         let mut state = self.state.lock();
         if matches!(*state, H2RequestClaimState::Prospective { .. }) {
             *state = H2RequestClaimState::Complete;
-+
+        }
-+
+    }
-+
     fn finish_upload(&self) -> bool {
         self.finish_side(|upload_finished, _| *upload_finished = true)
-+
+    }
-+
     fn finish_response(&self) -> bool {
         self.finish_side(|_, response_finished| *response_finished = true)
-+
+    }
-+
     /// Marks one request side finished and releases the claim on the second.
     fn finish_side(&self, finish: impl FnOnce(&mut bool, &mut bool)) -> bool {
         let completed_dispatch = {
             let mut state = self.state.lock();
             let previous = std::mem::replace(&mut *state, H2RequestClaimState::Complete);
             match previous {
                 H2RequestClaimState::Prospective {
                     mut upload_finished,
                     mut response_finished,
                 } => {
                     finish(&mut upload_finished, &mut response_finished);
                     *state = H2RequestClaimState::Prospective {
                         upload_finished,
                         response_finished,
                     };
                     None
-+
+                }
                 H2RequestClaimState::Accepted {
                     dispatch,
                     mut upload_finished,
                     mut response_finished,
                 } => {
                     finish(&mut upload_finished, &mut response_finished);
                     if upload_finished && response_finished {
                         Some(dispatch)
                     } else {
                         *state = H2RequestClaimState::Accepted {
                             dispatch,
                             upload_finished,
                             response_finished,
                         };
                         None
-+
+                    }
-+
+                }
                 H2RequestClaimState::Complete => None,
-+
+            }
         };
         let request_complete = completed_dispatch.is_some();
         if let Some(dispatch) = completed_dispatch {
             // Dispatch completion takes the connection lifecycle lock. Keep it
             // outside the claim lock so completion cannot nest pool locks.
             drop(dispatch);
             self.release_generation();
-+
+        }
         request_complete
-+
+    }
-+
     /// Releases the generation count after the claim lock is released.
     fn release_generation(&self) {
         if let Some(cell) = self.connection_cell.upgrade() {
             OriginCell::release_h2_request(&cell, self.generation);
-+
+        }
-+
+    }
-+
     #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
     fn for_test(cell: &Arc<OriginCell>) -> Arc<Self> {
         Arc::new(Self {
             connection_cell: Weak::from_arc(cell),
             generation: H2GenerationId(0),
             state: Mutex::new(H2RequestClaimState::Prospective {
                 upload_finished: false,
                 response_finished: false,
             }),
         })
-+
+    }
-+
+}
-+
 /// Request and connection identity retained for completion logs.
 #[derive(Clone)]
 struct H2RequestIdentity {
     request_partition: PartitionId,
     connection: Arc<ConnectionInfo>,
     generation: H2GenerationId,
-+
+}
-+
 fn trace_activation_cancelled(
     request_partition: PartitionId,
     generation: H2GenerationId,
     connection: &ConnectionInfo,
 ) {
     tracing::trace!(
         connection_id = %connection.id(),
         request_partition = ?request_partition,
         connection_partition = ?connection.owner_partition(),
         origin_scheme = %connection.origin().scheme(),
         origin_host = connection.origin().host(),
         origin_port = ?connection.origin().port(),
         h2_generation = ?generation,
         "HTTP/2 activation cancelled before request acceptance"
     );
-+
+}
-+
 /// Linear guard for request-upload completion.
 pub(in crate::client::pool) struct H2UploadGuard {
     claim: Arc<H2RequestClaim>,
     identity: Option<H2RequestIdentity>,
     active: bool,
-+
+}
-+
 impl H2UploadGuard {
     fn new(claim: Arc<H2RequestClaim>, identity: Option<H2RequestIdentity>) -> Self {
         Self {
             claim,
             identity,
             active: true,
-+
+        }
-+
+    }
-+
     #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
     pub(in crate::client::pool) fn for_test(cell: &Arc<OriginCell>) -> (Self, H2RequestClaimProbe) {
         let claim = H2RequestClaim::for_test(cell);
-+
+        (
             Self::new(claim.clone(), None),
             H2RequestClaimProbe { claim },
-+
+        )
-+
+    }
-+
     /// Finishes upload ownership before dropping the guard.
     pub(in crate::client::pool) fn finish(mut self) {
         self.finish_once();
-+
+    }
-+
     fn finish_once(&mut self) {
         if self.active {
             self.active = false;
             let request_complete = self.claim.finish_upload();
             trace_request_side(self.identity.take(), "upload", request_complete);
-+
+        }
-+
+    }
-+
+}
-+
 impl Drop for H2UploadGuard {
     fn drop(&mut self) {
         self.finish_once();
-+
+    }
-+
+}
-+
 /// Linear guard for response-future and response-body completion.
 pub(in crate::client::pool) struct H2ResponseGuard {
     claim: Arc<H2RequestClaim>,
     identity: Option<H2RequestIdentity>,
     active: bool,
-+
+}
-+
 impl H2ResponseGuard {
     fn new(claim: Arc<H2RequestClaim>, identity: Option<H2RequestIdentity>) -> Self {
         Self {
             claim,
             identity,
             active: true,
-+
+        }
-+
+    }
-+
     #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
     pub(in crate::client::pool) fn for_test(cell: &Arc<OriginCell>) -> (Self, H2RequestClaimProbe) {
         let claim = H2RequestClaim::for_test(cell);
-+
+        (
             Self::new(claim.clone(), None),
             H2RequestClaimProbe { claim },
-+
+        )
-+
+    }
-+
     /// Finishes response ownership before dropping the guard.
     pub(in crate::client::pool) fn finish(mut self) {
         self.finish_once();
-+
+    }
-+
     fn finish_once(&mut self) {
         if self.active {
             self.active = false;
             let request_complete = self.claim.finish_response();
             trace_request_side(self.identity.take(), "response", request_complete);
-+
+        }
-+
+    }
-+
+}
-+
 impl Drop for H2ResponseGuard {
     fn drop(&mut self) {
         self.finish_once();
-+
+    }
-+
+}
-+
 fn trace_request_side(
     identity: Option<H2RequestIdentity>,
     request_side: &'static str,
     request_complete: bool,
 ) {
     if let Some(identity) = identity {
         tracing::trace!(
             connection_id = %identity.connection.id(),
             request_partition = ?identity.request_partition,
             connection_partition = ?identity.connection.owner_partition(),
             origin_scheme = %identity.connection.origin().scheme(),
             origin_host = identity.connection.origin().host(),
             origin_port = ?identity.connection.origin().port(),
             h2_generation = ?identity.generation,
             request_side,
             request_complete,
             "HTTP/2 request side finished"
         );
-+
+    }
-+
+}
-+
 /// Test observation of one request claim's production state.
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 pub(in crate::client::pool) struct H2RequestClaimProbe {
     claim: Arc<H2RequestClaim>,
-+
+}
-+
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 impl H2RequestClaimProbe {
     /// Returns whether the upload side reached its terminal transition.
     pub(in crate::client::pool) fn upload_finished(&self) -> bool {
         matches!(
             &*self.claim.state.lock(),
             H2RequestClaimState::Prospective {
                 upload_finished: true,
                 ..
             } | H2RequestClaimState::Accepted {
                 upload_finished: true,
                 ..
             } | H2RequestClaimState::Complete
-+
+        )
-+
+    }
-+
     /// Returns whether the response side reached its terminal transition.
     pub(in crate::client::pool) fn response_finished(&self) -> bool {
         matches!(
             &*self.claim.state.lock(),
             H2RequestClaimState::Prospective {
                 response_finished: true,
                 ..
             } | H2RequestClaimState::Accepted {
                 response_finished: true,
                 ..
             } | H2RequestClaimState::Complete
-+
+        )
-+
+    }
-+
+}

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/cell/waiters.rs

@@ -20,20 +671,676 @@

↥ Expand context

 //! DeliveryPending --capacity-------------------------> ReadyToEstablish
 //! DeliveryPending --borrowed H1----------------------> Ready
 //! DeliveryPending --local H1--> DeliveryPending(pending H1)
 //! DeliveryPending(pending H1) --delivery--> Ready + return delivered value
 //! ReadyToEstablish --local H1 return-----------------> Ready + return permit
 //! ReadyToEstablish --poll----------------------------> Launching(Submitted)
 //! Launching(Submitted) --first owner-runtime poll----> Launching(Started)
 //! Launching --local H1 or establishment result------> Ready
 //! Waiting --local or peer H2 activation--------------> Ready
 //! DeliveryPending --H2 activation--------------------> DeliveryPending(pending H2)
 //! ReadyToEstablish / Launching --H2 activation-------> Ready
 //! Ready --poll---------------------------------------> removed; result transferred
 //! ```
 //!
 //! Cancellation has one residence used only while a committed delivery is
 //! crossing the cell lock boundary:
 //!
 //! ```text
 //! Waiting ----------------------> removed; retire or advance demand
 //! DeliveryPending --------------> DeliveryCancelled

 //! DeliveryCancelled --delivery-> removed; return payloads
 //! ReadyToEstablish / Ready -----> removed; return the held event
 //! Launching --------------------> removed; task settles separately
 //! ```
 //!
 //! Admission updates, returned values, and task wakes are detached for the
 //! caller to process after releasing the cell lock.
-47
 use super::super::admission::{DemandId, DemandSnapshot, ProtocolRequirement, SnapshotVersion};
 use super::super::partition::EligibilityGroup;
 use super::{AcquisitionEvent, AcquisitionResult, EstablishmentPermit};
 use super::{AcquisitionOutcome, AcquisitionStep, EstablishmentPermit};
 use std::collections::{BTreeSet, HashMap};
 use std::num::NonZeroUsize;
 use std::task::{Context, Poll, Waker};
-54
 /// Local waiter identity within one cell.
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub(in crate::client::pool) struct WaiterId(pub(in crate::client::pool) u64);
-58
 /// Complete acquisition state protected by one origin cell lock.
 ///
 /// At every completed transition:
 ///
 /// - `records` owns every retained acquisition attempt.
 /// - [`WaitingQueueState::Active`] contains exactly the linked
 ///   [`WaiterState::Waiting`] records.
 /// - The active demand identifies and describes the FIFO head.
 /// - `h1_candidates` contains exactly each H1-compatible
 /// - `h1_compatible_waiters` contains exactly each H1-compatible
 ///   [`WaiterState::DeliveryPending`] without a pending H1,
 ///   [`WaiterState::ReadyToEstablish`], or [`WaiterState::Launching`] attempt.
 /// - `h2_activation_candidates` contains every delivery-pending, ready-to-establish, or
 /// - `h2_compatible_waiters` contains every delivery-pending, ready-to-establish, or
 ///   launching attempt that has not received another result. The waiting FIFO
 ///   head is compared with this set without duplicating its queue residence.
 /// - Delivery-pending, ready-to-establish, launching, delivery-cancelled, and
 ///   ready attempts remain in `records` but never in the bounded FIFO.
 ///
 /// Transitions return admission publications, values requiring fallback, and
 /// wakers to the caller; none runs while this state is locked.
 #[derive(Debug, Default)]
 pub(super) struct AcquisitionQueue {
     /// Every attempt still waiting, launching, or holding a terminal result.
     records: HashMap<WaiterId, WaiterRecord>,
     /// FIFO endpoints and demand for the records currently waiting.
     waiting: WaitingQueueState,
     /// H1-compatible attempts still competing with a returned sender.
     h1_candidates: BTreeSet<WaiterId>,
     h1_compatible_waiters: BTreeSet<WaiterId>,
     /// Attempts outside the FIFO that may accept an H2 activation.
     h2_activation_candidates: BTreeSet<WaiterId>,
     h2_compatible_waiters: BTreeSet<WaiterId>,
     /// Next cell-local waiter identity.
     next_waiter: u64,
     next_waiter_id: u64,
     /// Next identity for a head-waiter demand generation.
     next_demand_id: u64,
-92
+}
-93
 /// Owner-runtime progress of one submitted establishment task.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum EstablishmentPhase {
     /// The task was submitted but has not begun polling the connector.
     Submitted,
     /// The owner runtime polled the task and the pool owns completion.
     Started,
-101
+}
-102
 /// Whether the cell has an active FIFO and aggregate head demand.
 #[derive(Debug, Default)]
 enum WaitingQueueState {
     /// No waiter is linked and no demand is active.
     #[default]
     Empty,
     /// A nonempty FIFO represented by its endpoints, length, and head demand.
     Active {
         /// Oldest waiting record.
         head: WaiterId,
         /// Youngest waiting record.
         tail: WaiterId,
         /// Number of records linked from `head` through `tail`.
         len: NonZeroUsize,
         /// Demand generation represented by `head`.
         demand: DemandTicket,
     },
-120
+}
-121
 /// One request retained while waiting for or owning an acquisition result.
 /// One request retained while waiting for or owning an acquisition outcome.
 #[derive(Debug)]
 struct WaiterRecord {
     /// Protocol requirement published while this waiter is the head.
     requirement: ProtocolRequirement,
     /// Queue residence, delivery state, and any owned result.
     /// Queue residence, delivery state, and any owned outcome.
     state: WaiterState,
-129
+}
-130
 /// Authoritative cell-local ownership state for one waiter.
 #[derive(Debug)]
 enum WaiterState {
     /// The waiter is linked in the cell-local FIFO.
     Waiting {
         /// Older waiting record, or `None` at the head.
         previous: Option<WaiterId>,
         /// Newer waiting record, or `None` at the tail.
         next: Option<WaiterId>,
         /// Latest task waiting for an acquisition result.
         /// Latest task waiting for an acquisition step.
         waker: Option<Waker>,
     },
     /// A committed capacity or reuse delivery reserved this waiter but has
     /// not installed its payload.
     DeliveryPending {
         /// Latest task waiting for the crossing delivery.
         waker: Option<Waker>,
         /// Local protocol result that won while delivery was crossing.
         pending_result: Option<AcquisitionResult>,
         /// Local protocol outcome that won while delivery was crossing.
         pending_result: Option<AcquisitionOutcome>,
     },
     /// Cancellation won after reservation while the committed delivery was
     /// outside the lock.
     DeliveryCancelled {
         /// Task detached for wake after the crossing payload is returned.
         waker: Option<Waker>,
         /// Protocol result returned after the crossing closes, if one arrived first.
         pending_result: Option<AcquisitionResult>,
         /// Protocol outcome returned after the crossing closes, if one arrived first.
         pending_result: Option<AcquisitionOutcome>,
     },
     /// Capacity is ready to start an establishment attempt.
     ReadyToEstablish {
         /// Optional bounded-origin capacity owned until connection install.
         permit: EstablishmentPermit,
     },
     /// Establishment was submitted to the owner runtime and may be unpolled.
     Launching {
         /// Ownership phase used to distinguish submission from first poll.
         phase: EstablishmentPhase,
         /// Latest task waiting for the attempt or returned H1 to complete.
         waker: Option<Waker>,
     },
     /// The waiter owns its terminal acquisition result.
     Ready(AcquisitionResult),
     /// The waiter owns its terminal acquisition outcome.
     Ready(AcquisitionOutcome),
-173
+}
-174
 /// Cell-local ownership of the aggregate head demand.
 #[derive(Debug)]
 struct DemandTicket {
     /// Identity of this head-waiter generation.
     id: DemandId,
     /// Version assigned to the next complete publication for this demand
     /// identity.
     version: SnapshotVersion,
     /// Protocol capability required by this head waiter.
     requirement: ProtocolRequirement,
-185
+}
-186
 impl DemandTicket {
     /// Creates the complete active state published for this head generation.
     fn snapshot(&self, eligibility_group: &EligibilityGroup) -> DemandSnapshot {
         DemandSnapshot::active(
             self.id,
             self.version,
             self.requirement,
             eligibility_group.clone(),
-195
+        )
-196
+    }
-197
+}
-198
 impl AcquisitionQueue {
     /// Registers one acquisition attempt.
     ///
     /// An unbounded attempt starts ready to establish. A bounded attempt joins
     /// the capacity FIFO and publishes demand only when it becomes the head.
     pub(super) fn register_waiter(
         &mut self,
         requirement: ProtocolRequirement,
         eligibility_group: &EligibilityGroup,
         bounded: bool,
     ) -> (WaiterId, Option<DemandSnapshot>) {
         let waiter = self.take_waiter_id();
         if !bounded {
             self.records.insert(
                 waiter,
                 WaiterRecord {
                     requirement,
                     state: WaiterState::ReadyToEstablish {
                         permit: EstablishmentPermit::unbounded(),
                     },
                 },
             );
             self.add_protocol_candidates(waiter, requirement);
             self.add_protocol_waiters(waiter, requirement);
             self.assert_consistent();
             return (waiter, None);
-224
+        }
-225
         let previous = match &self.waiting {
             WaitingQueueState::Empty => None,
             WaitingQueueState::Active { tail, .. } => Some(*tail),
         };
         let initial_demand = previous.is_none().then(|| self.new_demand(requirement));
-231
         if let Some(previous) = previous {
             let previous = self
                 .records
                 .get_mut(&previous)
                 .expect("waiting tail disappeared");
             let WaiterState::Waiting { next, .. } = &mut previous.state else {
                 unreachable!("waiting tail left the waiting state");
             };
             debug_assert!(next.is_none());
             *next = Some(waiter);
-242
+        }
-243
         let replaced = self.records.insert(
             waiter,
             WaiterRecord {
                 requirement,
                 state: WaiterState::Waiting {
                     previous,
                     next: None,
                     waker: None,
                 },
             },
         );
         debug_assert!(replaced.is_none());
-256
         let snapshot = match (&mut self.waiting, initial_demand) {
             (waiting @ WaitingQueueState::Empty, Some(demand)) => {
                 let snapshot = demand.snapshot(eligibility_group);
                 *waiting = WaitingQueueState::Active {
                     head: waiter,
                     tail: waiter,
                     len: NonZeroUsize::MIN,
                     demand,
                 };
                 Some(snapshot)
-267
+            }
             (WaitingQueueState::Active { tail, len, .. }, None) => {
                 *tail = waiter;
                 *len = len.checked_add(1).expect("waiter queue length exhausted");
                 None
-272
+            }
             _ => unreachable!("waiter queue occupancy changed during registration"),
         };
         self.assert_consistent();
         (waiter, snapshot)
-277
+    }
-278
     /// Returns the oldest acquisition attempt that can accept HTTP/1.
     ///
     /// A bounded waiter remains a candidate after receiving capacity so a
     /// A bounded waiter remains compatible after receiving capacity so a
     /// returned sender can still beat its lazy or pool-owned establishment
     /// attempt. Comparing waiter identities preserves arrival order between
     /// that set and the current capacity-waiting head.
     fn oldest_h1_candidate(&self) -> Option<WaiterId> {
     fn oldest_h1_compatible_waiter(&self) -> Option<WaiterId> {
         let waiting = match &self.waiting {
             WaitingQueueState::Active { head, .. }
                 if self.records[head].requirement.accepts_h1() =>
-289
+            {
                 Some(*head)
-291
+            }
             WaitingQueueState::Empty | WaitingQueueState::Active { .. } => None,
         };
         match (waiting, self.h1_candidates.first().copied()) {
         match (waiting, self.h1_compatible_waiters.first().copied()) {
             (Some(waiting), Some(launching)) => Some(waiting.min(launching)),
             (Some(waiting), None) => Some(waiting),
             (None, launching) => launching,
-298
+        }
-299
+    }
-300
     /// Returns whether a returned H1 can satisfy a live acquisition attempt.
     pub(super) fn can_accept_h1(&self) -> bool {
         self.oldest_h1_candidate().is_some()
     pub(super) fn has_h1_compatible_waiter(&self) -> bool {
         self.oldest_h1_compatible_waiter().is_some()
-304
+    }
-305
     /// Returns whether a previously admitted local attempt still accepts H1.
     ///
     /// These candidates left the capacity FIFO before its current head and
     /// These waiters left the capacity FIFO before its current head and
     /// therefore precede cross-cell reuse for that newer aggregate demand.
     pub(super) fn has_prior_h1_candidate(&self) -> bool {
         !self.h1_candidates.is_empty()
     pub(super) fn has_prior_h1_waiter(&self) -> bool {
         !self.h1_compatible_waiters.is_empty()
-312
+    }
-313
     /// Returns the oldest attempt that may accept an HTTP/2 activation.
     pub(super) fn oldest_h2_candidate(&self) -> Option<WaiterId> {
     pub(super) fn oldest_h2_compatible_waiter(&self) -> Option<WaiterId> {
         let waiting = match &self.waiting {
             WaitingQueueState::Active { head, .. }
                 if self.records[head].requirement.accepts_h2() =>
-319
+            {
                 Some(*head)
-321
+            }
             WaitingQueueState::Active { .. } => None,
             WaitingQueueState::Empty => None,
         };
         match (waiting, self.h2_activation_candidates.first().copied()) {
         match (waiting, self.h2_compatible_waiters.first().copied()) {
             (Some(waiting), Some(active)) => Some(waiting.min(active)),
             (Some(waiting), None) => Some(waiting),
             (None, active) => active,
-329
+        }
-330
+    }
-331
     /// Advances the active demand beyond an accepted route publication.
     /// Advances the active demand beyond an accepted peer route.
     ///
     /// Admission retires the published version when it acknowledges route
     /// visibility. The waiter remains queued locally while the route gate
     /// offers activations. Reserving the following version here ensures that
     /// route invalidation can republish the same demand generation without an
     /// older inactive snapshot winning.
     pub(super) fn suppress_published_demand(&mut self, demand: DemandId) -> bool {
     pub(super) fn supersede_demand_snapshot(&mut self, demand: DemandId) -> bool {
         let WaitingQueueState::Active {
             demand: current, ..
         } = &mut self.waiting
         else {
             return false;
         };
         if current.id != demand {
             return false;
-348
+        }
         let acknowledged = current.version.next();
         current.version = acknowledged.next();
         self.assert_consistent();
         true
-353
+    }
-354
     /// Returns the current head demand for publication after route loss.
     pub(super) fn current_demand_snapshot(
         &self,
         eligibility_group: &EligibilityGroup,
     ) -> Option<DemandSnapshot> {
         match &self.waiting {
             WaitingQueueState::Empty => None,
             WaitingQueueState::Active { demand, .. } => Some(demand.snapshot(eligibility_group)),
-363
+        }
-364
+    }
-365
     /// Returns whether `waiter` is still the oldest H2 activation candidate.
     pub(super) fn is_oldest_h2_candidate(&self, waiter: WaiterId) -> bool {
         self.oldest_h2_candidate() == Some(waiter)
     /// Returns whether `waiter` is still the oldest H2-compatible waiter.
     pub(super) fn is_oldest_h2_compatible_waiter(&self, waiter: WaiterId) -> bool {
         self.oldest_h2_compatible_waiter() == Some(waiter)
-369
+    }
-370
     /// Returns the newest waiter identity committed before a publication.
     pub(super) fn publication_cutoff(&self) -> Option<WaiterId> {
     /// Returns the newest waiter identity committed before peer route installation.
     pub(super) fn route_cutoff(&self) -> Option<WaiterId> {
         (!self.records.is_empty()).then(|| {
             WaiterId(
                 self.next_waiter
                 self.next_waiter_id
                     .checked_sub(1)
                     .expect("nonempty waiter set had no allocated identity"),
-378
+            )
         })
-380
+    }
-381
     /// Returns whether an acquisition at or before `cutoff` still needs H2.
     pub(super) fn has_h2_candidate_through(&self, cutoff: WaiterId) -> bool {
         self.oldest_h2_candidate()
     pub(super) fn has_h2_compatible_waiter_through(&self, cutoff: WaiterId) -> bool {
         self.oldest_h2_compatible_waiter()
             .is_some_and(|waiter| waiter <= cutoff)
-386
+    }
-387
     /// Returns whether any acquisition must precede a new direct H2 arrival.
     pub(super) fn has_h2_candidate(&self) -> bool {
         self.oldest_h2_candidate().is_some()
     pub(super) fn has_h2_compatible_waiter(&self) -> bool {
         self.oldest_h2_compatible_waiter().is_some()
-391
+    }
-392
     /// Returns whether one transferred generation waiter remains launchable.
     pub(super) fn is_launching_h2_candidate(&self, waiter: WaiterId) -> bool {
         self.h2_activation_candidates.contains(&waiter)
     pub(super) fn is_launching_h2_waiter(&self, waiter: WaiterId) -> bool {
         self.h2_compatible_waiters.contains(&waiter)
             && self.records.get(&waiter).is_some_and(|record| {
                 record.requirement.accepts_h2()
                     && matches!(record.state, WaiterState::Launching { .. })
             })
-400
+    }
-401
     /// Adds one waiter to each protocol index its requirement accepts.
     fn add_protocol_candidates(&mut self, waiter: WaiterId, requirement: ProtocolRequirement) {
     /// Adds one waiter to each compatible-protocol index.
     fn add_protocol_waiters(&mut self, waiter: WaiterId, requirement: ProtocolRequirement) {
         if requirement.accepts_h1() {
             self.h1_candidates.insert(waiter);
             self.h1_compatible_waiters.insert(waiter);
-406
+        }
         if requirement.accepts_h2() {
             self.h2_activation_candidates.insert(waiter);
             self.h2_compatible_waiters.insert(waiter);
-409
+        }
-410
+    }
-411
     /// Removes one waiter from both protocol candidate indexes.
     fn remove_protocol_candidates(&mut self, waiter: WaiterId) {
         self.h1_candidates.remove(&waiter);
         self.h2_activation_candidates.remove(&waiter);
     /// Removes one waiter from both compatible-protocol indexes.
     fn remove_protocol_waiters(&mut self, waiter: WaiterId) {
         self.h1_compatible_waiters.remove(&waiter);
         self.h2_compatible_waiters.remove(&waiter);
-416
+    }
-417
     /// Commits a returned HTTP/1 sender to the oldest compatible attempt.
     ///
     /// Payload construction is delayed until all panic-capable requesting-cell checks
     /// have completed. Once constructed, the result is either state-owned or
     /// returned to the caller for cleanup after the cell lock is released.
     pub(super) fn install_returned_h1(
     pub(super) fn offer_returned_h1(
         &mut self,
         result: impl FnOnce() -> AcquisitionResult,
         result: impl FnOnce() -> AcquisitionOutcome,
         eligibility_group: &EligibilityGroup,
     ) -> WaiterInstall {
         let Some(waiter) = self.oldest_h1_candidate() else {
             return WaiterInstall::rejected(result());
     ) -> (Option<WaiterId>, WaiterResolution) {
         let Some(waiter) = self.oldest_h1_compatible_waiter() else {
             return (None, WaiterResolution::refused(result()));
         };
         self.install_protocol_result(waiter, result, eligibility_group)
-+
+        (
             Some(waiter),
             self.assign_protocol_outcome(waiter, result, eligibility_group),
-+
+        )
-435
+    }
-436
     /// Installs a selected protocol result through the shared waiter transition.
     fn install_protocol_result(
     fn assign_protocol_outcome(
         &mut self,
         waiter: WaiterId,
         result: impl FnOnce() -> AcquisitionResult,
         result: impl FnOnce() -> AcquisitionOutcome,
         eligibility_group: &EligibilityGroup,
     ) -> WaiterInstall {
     ) -> WaiterResolution {
         if matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::Waiting { .. })
         ) {
             let removed = self.pop_head(eligibility_group);
             debug_assert_eq!(removed.waiter, waiter);
             let record = self
                 .records
                 .get_mut(&waiter)
                 .expect("selected requesting waiter disappeared");
             let WaiterState::Waiting { waker, .. } = &mut record.state else {
                 unreachable!("selected requesting waiter left the waiting state");
             };
             let waker = waker.take();
             record.state = WaiterState::Ready(result());
             let retired =
                 DemandSnapshot::inactive(removed.demand.id, removed.demand.version.next());
             self.assert_consistent();
             return WaiterInstall {
                 waiter: Some(waiter),
             return WaiterResolution {
                 demand_updates: [Some(retired), removed.successor],
                 returned_event: None,
                 returned_step: None,
                 waker,
             };
-467
+        }
-468
         let record = self
             .records
             .get_mut(&waiter)
             .expect("selected candidate waiter disappeared");
         let (returned_event, waker) = match &mut record.state {
             .expect("selected compatible waiter disappeared");
         let (returned_step, waker) = match &mut record.state {
             WaiterState::DeliveryPending {
                 waker: _,
                 pending_result,
             } => {
                 debug_assert!(pending_result.is_none());
                 *pending_result = Some(result());
                 (None, None)
-481
+            }
             WaiterState::ReadyToEstablish { .. } | WaiterState::Launching { .. } => {
                 let previous = std::mem::replace(&mut record.state, WaiterState::Ready(result()));
                 match previous {
                     WaiterState::ReadyToEstablish { permit } => {
                         (Some(AcquisitionEvent::Establish(permit)), None)
                         (Some(AcquisitionStep::StartEstablishment(permit)), None)
-487
+                    }
                     WaiterState::Launching { waker, .. } => (None, waker),
                     _ => unreachable!("selected candidate changed state under the cell lock"),
                     _ => {
                         unreachable!("selected compatible waiter changed state under the cell lock")
-+
+                    }
-492
+                }
-493
+            }
             WaiterState::Waiting { .. }
             | WaiterState::DeliveryCancelled { .. }
             | WaiterState::Ready(_) => {
                 unreachable!("selected waiter was not eligible for a protocol result")
-498
+            }
         };
         self.remove_protocol_candidates(waiter);
         if returned_event.is_none() {
         self.remove_protocol_waiters(waiter);
         if returned_step.is_none() {
             // The result is state-owned and no permit is detached, so an
             // invariant failure cannot drop a pool value under this lock.
             self.assert_consistent();
-505
+        }
         WaiterInstall {
             waiter: Some(waiter),
         WaiterResolution {
             demand_updates: [None, None],
             returned_event,
             returned_step,
             waker,
-510
+        }
-511
+    }
-512
     /// Cancels one waiter and detaches all cross-lock cleanup work.
     ///
     /// Removing the head retires its demand and starts a successor generation.
     /// Removing another waiting record leaves the active demand unchanged.
     /// Cancellation during delivery leaves a marker for result installation to
     /// observe. A ready result is returned to the caller for cleanup after unlock.
     /// Cancellation while delivery is pending leaves a marker for the exact
     /// waiter commit to observe. A ready step is returned for cleanup after unlock.
     pub(super) fn cancel_waiter(
         &mut self,
         waiter: WaiterId,
         eligibility_group: &EligibilityGroup,
     ) -> Option<WaiterCancellation> {
         let state = &self.records.get(&waiter)?.state;
         let cancellation = if matches!(state, WaiterState::Waiting { .. }) {
             let is_head = matches!(
                 self.waiting,
                 WaitingQueueState::Active { head, .. } if head == waiter
             );
             let demand_updates = if is_head {
                 let removed = self.pop_head(eligibility_group);
                 debug_assert_eq!(waiter, removed.waiter);
                 let record = self
                     .records
                     .remove(&waiter)
                     .expect("cancelled head waiter disappeared");
                 debug_assert!(matches!(record.state, WaiterState::Waiting { .. }));
                 let retired =
                     DemandSnapshot::inactive(removed.demand.id, removed.demand.version.next());
                 [Some(retired), removed.successor]
             } else {
                 self.remove_non_head(waiter);
                 [None, None]
             };
             WaiterCancellation {
                 demand_updates,
                 returned_events: [None, None],
                 returned_steps: [None, None],
-548
+            }
         } else if matches!(state, WaiterState::DeliveryPending { .. }) {
             self.remove_protocol_candidates(waiter);
             self.remove_protocol_waiters(waiter);
             let record = self.records.get_mut(&waiter)?;
             let WaiterState::DeliveryPending {
                 waker,
                 pending_result,
             } = &mut record.state
             else {
                 unreachable!("delivery-pending waiter changed state under the cell lock");
             };
             let waker = waker.take();
             let pending_result = pending_result.take();
             record.state = WaiterState::DeliveryCancelled {
                 waker,
                 pending_result,
             };
             WaiterCancellation {
                 demand_updates: [None, None],
                 returned_events: [None, None],
                 returned_steps: [None, None],
-568
+            }
         } else if matches!(
             state,
             WaiterState::ReadyToEstablish { .. } | WaiterState::Ready(_)
         ) {
             // Validate before detaching the event. Once it is local,
             // this method must return without another panic-capable check.
             self.assert_consistent();
             let record = self.records.remove(&waiter)?;
             self.remove_protocol_candidates(waiter);
             self.remove_protocol_waiters(waiter);
             let event = match record.state {
                 WaiterState::ReadyToEstablish { permit } => AcquisitionEvent::Establish(permit),
                 WaiterState::Ready(result) => AcquisitionEvent::Complete(result),
                 WaiterState::ReadyToEstablish { permit } => {
                     AcquisitionStep::StartEstablishment(permit)
-+
+                }
                 WaiterState::Ready(result) => AcquisitionStep::Resolved(result),
                 _ => unreachable!("ready waiter changed state under the cell lock"),
             };
             return Some(WaiterCancellation {
                 demand_updates: [None, None],
                 returned_events: [Some(event), None],
                 returned_steps: [Some(event), None],
             });
         } else if matches!(state, WaiterState::Launching { .. }) {
             self.assert_consistent();
             self.remove_protocol_candidates(waiter);
             self.remove_protocol_waiters(waiter);
             self.records.remove(&waiter)?;
             return Some(WaiterCancellation {
                 demand_updates: [None, None],
                 returned_events: [None, None],
                 returned_steps: [None, None],
             });
         } else {
             debug_assert!(matches!(state, WaiterState::DeliveryCancelled { .. }));
             return None;
         };
-601
         self.assert_consistent();
         Some(cancellation)
-604
+    }
-605
     /// Returns the next event or records the latest waker for a pending waiter.
     ///
     /// # Panics
     ///
     /// Panics if `waiter` is unknown, was cancelled, or was already consumed
     /// by an earlier ready poll.
     pub(super) fn poll_waiter(
         &mut self,
         waiter: WaiterId,
         cx: &mut Context<'_>,
     ) -> Poll<AcquisitionEvent> {
     ) -> Poll<AcquisitionStep> {
         if matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::ReadyToEstablish { .. })
         ) {
             self.assert_consistent();
             let record = self
                 .records
                 .get_mut(&waiter)
                 .expect("ready establishment waiter disappeared");
             let previous = std::mem::replace(
                 &mut record.state,
                 WaiterState::Launching {
                     phase: EstablishmentPhase::Submitted,
                     waker: None,
                 },
             );
             let WaiterState::ReadyToEstablish { permit } = previous else {
                 unreachable!("ready establishment waiter changed state under the cell lock");
             };
             return Poll::Ready(AcquisitionEvent::Establish(permit));
             return Poll::Ready(AcquisitionStep::StartEstablishment(permit));
-637
+        }
-638
         if matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::Ready(_))
         ) {
             return Poll::Ready(
                 self.take_ready_result(waiter)
                     .map(AcquisitionEvent::Complete)
                     .expect("ready waiter lost its acquisition result"),
                     .map(AcquisitionStep::Resolved)
                     .expect("ready waiter lost its acquisition outcome"),
             );
-648
+        }
-649
         let record = self
             .records
             .get_mut(&waiter)
             .expect("polled a cancelled, consumed, or unknown acquisition waiter");
         let waker = match &mut record.state {
             WaiterState::Waiting { waker, .. }
             | WaiterState::DeliveryPending { waker, .. }

↧ Expand context

             | WaiterState::Launching { waker, .. } => waker,
             WaiterState::DeliveryCancelled { .. } => {
                 panic!("polled a cancelled acquisition waiter")
-660
+            }
             WaiterState::ReadyToEstablish { .. } | WaiterState::Ready(_) => {
                 unreachable!("ready waiter changed state under the cell lock")
-663
+            }
         };
         if waker
             .as_ref()
             .is_none_or(|waker| !waker.will_wake(cx.waker()))
-668
+        {
             *waker = Some(cx.waker().clone());
-670
+        }
         Poll::Pending
-672
+    }
-673
     /// Transfers completion ownership to the pool immediately before the first poll.
     ///
     /// A returned H1 or cancellation that wins first removes the submitted

@@ -703,708 +991,976 @@

↥ Expand context

         &mut self,
         demand: DemandId,
         eligibility_group: &EligibilityGroup,
     ) -> DeliveryReservation {
         let current = matches!(
             &self.waiting,
             WaitingQueueState::Active {
                 demand: current,
                 ..
             } if current.id == demand
         );
         if !current {
             return DeliveryReservation::Rejected;
-721
+        }
-722
         let removed = self.pop_head(eligibility_group);
         debug_assert_eq!(removed.demand.id, demand);
         let record = self
             .records
             .get_mut(&removed.waiter)

             .expect("reserved queue head disappeared");
         let WaiterState::Waiting { waker, .. } = &mut record.state else {
             unreachable!("reserved queue head left the waiting state");
         };
         let waker = waker.take();
         let requirement = record.requirement;
         record.state = WaiterState::DeliveryPending {
             waker,
             pending_result: None,
         };
         self.add_protocol_candidates(removed.waiter, requirement);
         self.add_protocol_waiters(removed.waiter, requirement);
-739
         self.assert_consistent();
         DeliveryReservation::Reserved {
             waiter: removed.waiter,
             successor: removed.successor,
-744
+        }
-745
+    }
-746
     /// Completes the unlocked capacity crossing without panicking while the
     /// incoming permit remains a droppable local.
     ///
     /// A live receiver takes ownership of the permit before another
     /// panic-capable operation. If an H1 return already won, both resources
     /// leave the lock as separate events and the permit is refunnelled.
     pub(super) fn install_capacity(
     pub(super) fn commit_capacity(
         &mut self,
         waiter: WaiterId,
         permit: EstablishmentPermit,
     ) -> ResultInstall {
     ) -> CellCommitOutcome {
         let Some(record) = self.records.get_mut(&waiter) else {
             return ResultInstall::invalid(
                 AcquisitionEvent::Establish(permit),
                 ResultInstallError::MissingWaiter,
             return CellCommitOutcome::invalid(
                 AcquisitionStep::StartEstablishment(permit),
                 CellCommitError::MissingWaiter,
             );
         };
-764
         match &mut record.state {
             WaiterState::DeliveryPending {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 if let Some(result) = pending_result.take() {
                     record.state = WaiterState::Ready(result);
                     self.remove_protocol_candidates(waiter);
                     self.remove_protocol_waiters(waiter);
                     // The rejected permit must cross the lock boundary even
                     // if other state is already inconsistent.
                     return ResultInstall {
                         returned_events: [Some(AcquisitionEvent::Establish(permit)), None],
                     return CellCommitOutcome::Refused {
                         returned: [Some(AcquisitionStep::StartEstablishment(permit)), None],
                         waker,
                         error: None,
                         accepted: false,
                     };
-780
+                }
                 record.state = WaiterState::ReadyToEstablish { permit };
                 // The permit is state-owned before this check can panic.
                 self.assert_consistent();
                 ResultInstall {
                     returned_events: [None, None],
                     waker,
                     error: None,
                     accepted: true,
--
+                }
                 CellCommitOutcome::Committed { waker }
-785
+            }
             WaiterState::DeliveryCancelled {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 let pending_result = pending_result.take();
                 self.records.remove(&waiter);
                 ResultInstall {
                     returned_events: [
                         pending_result.map(AcquisitionEvent::Complete),
                         Some(AcquisitionEvent::Establish(permit)),
                 CellCommitOutcome::Refused {
                     returned: [
                         pending_result.map(AcquisitionStep::Resolved),
                         Some(AcquisitionStep::StartEstablishment(permit)),
                     ],
                     waker,
                     error: None,
                     accepted: false,
-799
+                }
-800
+            }
             WaiterState::Waiting { .. }
             | WaiterState::ReadyToEstablish { .. }
             | WaiterState::Launching { .. }
             | WaiterState::Ready(_) => ResultInstall::invalid(
                 AcquisitionEvent::Establish(permit),
                 ResultInstallError::UnexpectedState,
             | WaiterState::Ready(_) => CellCommitOutcome::invalid(
                 AcquisitionStep::StartEstablishment(permit),
                 CellCommitError::UnexpectedState,
             ),
-808
+        }
-809
+    }
-810
     /// Installs a borrowed HTTP/1 result after its requesting waiter was reserved.
     ///
     /// A local return may have won while the borrowed sender crossed its
     /// owning-cell and requesting-cell locks. In that case the local result stays ready and
     /// the borrowed result leaves the lock for ordinary owning-cell return.
     pub(super) fn install_borrowed_h1(
     pub(super) fn commit_borrowed_h1(
         &mut self,
         waiter: WaiterId,
         result: AcquisitionResult,
     ) -> ResultInstall {
         result: AcquisitionOutcome,
     ) -> CellCommitOutcome {
         let Some(record) = self.records.get_mut(&waiter) else {
             return ResultInstall::invalid(
                 AcquisitionEvent::Complete(result),
                 ResultInstallError::MissingWaiter,
             return CellCommitOutcome::invalid(
                 AcquisitionStep::Resolved(result),
                 CellCommitError::MissingWaiter,
             );
         };
-827
         match &mut record.state {
             WaiterState::DeliveryPending {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 if let Some(local_result) = pending_result.take() {
                     record.state = WaiterState::Ready(local_result);
                     self.remove_protocol_candidates(waiter);
                     self.remove_protocol_waiters(waiter);
                     // The rejected borrowed sender must cross the lock
                     // boundary before any panic-capable consistency check.
                     return ResultInstall {
                         returned_events: [Some(AcquisitionEvent::Complete(result)), None],
                     return CellCommitOutcome::Refused {
                         returned: [Some(AcquisitionStep::Resolved(result)), None],
                         waker,
                         error: None,
                         accepted: false,
                     };
-843
+                }
                 record.state = WaiterState::Ready(result);
                 self.remove_protocol_candidates(waiter);
                 self.remove_protocol_waiters(waiter);
                 self.assert_consistent();
                 ResultInstall {
                     returned_events: [None, None],
                     waker,
                     error: None,
                     accepted: true,
--
+                }
                 CellCommitOutcome::Committed { waker }
-848
+            }
             WaiterState::DeliveryCancelled {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 let pending_result = pending_result.take();
                 self.records.remove(&waiter);
                 ResultInstall {
                     returned_events: [
                         pending_result.map(AcquisitionEvent::Complete),
                         Some(AcquisitionEvent::Complete(result)),
                 CellCommitOutcome::Refused {
                     returned: [
                         pending_result.map(AcquisitionStep::Resolved),
                         Some(AcquisitionStep::Resolved(result)),
                     ],
                     waker,
                     error: None,
                     accepted: false,
-862
+                }
-863
+            }
             WaiterState::Waiting { .. }
             | WaiterState::ReadyToEstablish { .. }
             | WaiterState::Launching { .. }
             | WaiterState::Ready(_) => ResultInstall::invalid(
                 AcquisitionEvent::Complete(result),
                 ResultInstallError::UnexpectedState,
             | WaiterState::Ready(_) => CellCommitOutcome::invalid(
                 AcquisitionStep::Resolved(result),
                 CellCommitError::UnexpectedState,
             ),
-871
+        }
-872
+    }
-873
     /// Commits one terminal establishment result when its waiter is still live.
     ///
     /// Completion may occur from either launch phase: normal execution starts
     /// the connector first, while dropping an unpolled submitted task installs
     /// a terminal runtime error.
     pub(super) fn install_establishment_result(
     pub(super) fn commit_establishment(
         &mut self,
         waiter: WaiterId,
         result: AcquisitionResult,
     ) -> ResultInstall {
         result: AcquisitionOutcome,
     ) -> CellCommitOutcome {
         let Some(record) = self.records.get_mut(&waiter) else {
             return ResultInstall::rejected(AcquisitionEvent::Complete(result));
             return CellCommitOutcome::refused(AcquisitionStep::Resolved(result));
         };
         if matches!(record.state, WaiterState::Ready(_)) {
             return ResultInstall::rejected(AcquisitionEvent::Complete(result));
             return CellCommitOutcome::refused(AcquisitionStep::Resolved(result));
-889
+        }
         if !matches!(record.state, WaiterState::Launching { .. }) {
             return ResultInstall::invalid(
                 AcquisitionEvent::Complete(result),
                 ResultInstallError::UnexpectedState,
             return CellCommitOutcome::invalid(
                 AcquisitionStep::Resolved(result),
                 CellCommitError::UnexpectedState,
             );
-895
+        }
         let WaiterState::Launching { waker, .. } = &mut record.state else {
             unreachable!("validated launching waiter changed state")
         };
         let waker = waker.take();
         record.state = WaiterState::Ready(result);
         self.remove_protocol_candidates(waiter);
         self.remove_protocol_waiters(waiter);
         self.assert_consistent();
         ResultInstall {
             returned_events: [None, None],
             waker,
             error: None,
             accepted: true,
--
+        }
         CellCommitOutcome::Committed { waker }
-904
+    }
-905
     /// Removes a ready waiter and transfers ownership of its result.
     pub(super) fn take_ready_result(&mut self, waiter: WaiterId) -> Option<AcquisitionResult> {
     pub(super) fn take_ready_result(&mut self, waiter: WaiterId) -> Option<AcquisitionOutcome> {
         if !matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::Ready(_))
         ) {
             return None;
-913
+        }
-914
         // Validate before detaching the result. Returning it must be the last
         // operation performed while the cell lock is held.
         self.assert_consistent();
         let record = self.records.remove(&waiter)?;
         let WaiterState::Ready(result) = record.state else {
             unreachable!("ready waiter changed state under the cell lock");
         };
         Some(result)
-923
+    }
-924
     /// Offers an H2 activation to the oldest compatible acquisition.
     ///
     /// `cutoff` retains publication priority. A later waiter is left pending
     /// `cutoff` retains peer-route priority. A later waiter is left pending
     /// until every live acquisition at or before the cutoff has received an
     /// activation opportunity.
     pub(super) fn install_h2(
     pub(super) fn offer_h2_activation(
         &mut self,
         cutoff: Option<WaiterId>,
         result: impl FnOnce(WaiterId) -> AcquisitionResult,
         result: impl FnOnce(WaiterId) -> AcquisitionOutcome,
         eligibility_group: &EligibilityGroup,
     ) -> WaiterInstall {
         let Some(waiter) = self.oldest_h2_candidate() else {
             return WaiterInstall::empty();
     ) -> (Option<WaiterId>, WaiterResolution) {
         let Some(waiter) = self.oldest_h2_compatible_waiter() else {
             return (None, WaiterResolution::empty());
         };
         if cutoff.is_some_and(|cutoff| waiter > cutoff) {
             return WaiterInstall::empty();
             return (None, WaiterResolution::empty());
-941
+        }
-942
         self.install_protocol_result(waiter, || result(waiter), eligibility_group)
-+
+        (
             Some(waiter),
             self.assign_protocol_outcome(waiter, || result(waiter), eligibility_group),
-+
+        )
-947
+    }
-948
     /// Unlinks the FIFO head and installs any successor demand.
     ///
     /// The head record remains in `records`; the caller either changes it to
     /// `DeliveryPending` or removes it for cancellation.
     fn pop_head(&mut self, eligibility_group: &EligibilityGroup) -> RemovedHead {
         let waiting = std::mem::take(&mut self.waiting);
         let WaitingQueueState::Active {
             head,

↧ Expand context

             tail,
             len,
             demand,
         } = waiting
         else {
             unreachable!("removed a head from an empty waiter queue");
         };
         let next = match &self
             .records
             .get(&head)
             .expect("waiting head disappeared")
             .state
-969
+        {
             WaiterState::Waiting { previous, next, .. } => {
                 debug_assert!(previous.is_none());
                 *next
-973
+            }
             _ => unreachable!("waiting head left the waiting state"),
         };
-976

@@ -1071,1056 +1158,1143 @@

↥ Expand context

             } = &mut next_record.state
             else {
                 unreachable!("next waiter left the waiting state");
             };
             debug_assert_eq!(*next_previous, Some(waiter));
             *next_previous = Some(previous);
-1062
+        }
-1063
         let WaitingQueueState::Active {
             head, tail, len, ..
         } = &mut self.waiting
         else {
             unreachable!("removed a waiter from an empty queue");
         };
         debug_assert_ne!(*head, waiter);
         if next.is_none() {
             debug_assert_eq!(*tail, waiter);
             *tail = previous;
-1074
+        }
         *len = NonZeroUsize::new(

             len.get()
                 .checked_sub(1)
                 .expect("waiter queue length underflowed"),
-1079
+        )
         .expect("removing a non-head waiter emptied the queue");
         record
-1082
+    }
-1083
     /// Allocates an identity that is never reused within this cell.
     fn take_waiter_id(&mut self) -> WaiterId {
         let value = self.next_waiter;
         self.next_waiter = value.checked_add(1).expect("waiter identity exhausted");
         let value = self.next_waiter_id;
         self.next_waiter_id = value.checked_add(1).expect("waiter identity exhausted");
         WaiterId(value)
-1089
+    }
-1090
     /// Starts a demand generation for a waiter that became the FIFO head.
     fn new_demand(&mut self, requirement: ProtocolRequirement) -> DemandTicket {
         let value = self.next_demand_id;
         self.next_demand_id = value.checked_add(1).expect("demand identity exhausted");
         DemandTicket {
             id: DemandId::from_u64(value),
             version: SnapshotVersion::INITIAL,
             requirement,
-1099
+        }
-1100
+    }
-1101
     /// Checks map, FIFO, and demand relationships in debug and test builds.
     pub(super) fn assert_consistent(&self) {
         #[cfg(debug_assertions)]
         #[cfg(any(debug_assertions, test))]
-1105
+        {
             if std::thread::panicking() {
                 return;
-1108
+            }
             self.assert_consistent_debug();
-1110
+        }
-1111
+    }
-1112
     #[cfg(debug_assertions)]
     #[cfg(any(debug_assertions, test))]
     fn assert_consistent_debug(&self) {
         let waiting_records = self
             .records
             .values()
             .filter(|record| matches!(record.state, WaiterState::Waiting { .. }))
             .count();
         match &self.waiting {
             WaitingQueueState::Empty => {
                 assert_eq!(0, waiting_records, "empty queue retained waiting records");
-1123
+            }

↧ Expand context

             WaitingQueueState::Active {
                 head,
                 tail,
                 len,
                 demand,
             } => {
                 let head_record = self.records.get(head).expect("waiting head disappeared");
                 assert_eq!(
                     demand.requirement, head_record.requirement,
                     "aggregate demand did not describe the waiting head"
                 );
-1135
                 let mut current = Some(*head);
                 let mut previous = None;
                 let mut traversed = 0;
                 while let Some(waiter) = current {
                     assert!(
                         traversed < self.records.len(),
                         "waiter queue contains a cycle"
                     );

@@ -1163,1148 +1412,1400 @@

↥ Expand context

                     let WaiterState::Waiting {
                         previous: linked_previous,
                         next,
                         ..
                     } = &record.state
                     else {
                         panic!("linked waiter left the waiting state");
                     };
                     assert_eq!(
                         previous, *linked_previous,
                         "waiter queue contains inconsistent backward links"
                     );
                     traversed += 1;
                     previous = Some(waiter);
                     current = *next;
-1163
+                }
-1164
                 assert_eq!(Some(*tail), previous, "waiting tail was not reachable");
                 assert_eq!(
                     len.get(),

                     traversed,
                     "waiter queue length did not match its links"
                 );
                 assert_eq!(
                     waiting_records, traversed,
                     "waiting record was not reachable from the queue head"
                 );
-1175
+            }
-1176
+        }
-1177
         let expected_h1_candidates = self
         let expected_h1_waiters = self
             .records
             .iter()
             .filter_map(|(waiter, record)| {
                 (record.requirement.accepts_h1()
                     && matches!(
                         record.state,
                         WaiterState::DeliveryPending {
                             pending_result: None,
                             ..
                         } | WaiterState::ReadyToEstablish { .. }
                             | WaiterState::Launching { .. }
                     ))
                 .then_some(*waiter)
             })
             .collect::<BTreeSet<_>>();
         assert_eq!(
             expected_h1_candidates, self.h1_candidates,
             "HTTP/1 acquisition candidates did not match waiter state"
             expected_h1_waiters, self.h1_compatible_waiters,
             "HTTP/1 compatible-waiter index did not match waiter state"
         );
-1198
         let expected_h2_activation_candidates = self
         let expected_h2_waiters = self
             .records
             .iter()
             .filter_map(|(waiter, record)| {
                 (record.requirement.accepts_h2()
                     && matches!(
                         record.state,
                         WaiterState::DeliveryPending {
                             pending_result: None,
                             ..
                         } | WaiterState::ReadyToEstablish { .. }
                             | WaiterState::Launching { .. }
                     ))
                 .then_some(*waiter)
             })
             .collect::<BTreeSet<_>>();
         assert_eq!(
             expected_h2_activation_candidates, self.h2_activation_candidates,
             "HTTP/2 acquisition candidates did not match waiter state"
             expected_h2_waiters, self.h2_compatible_waiters,
             "HTTP/2 compatible-waiter index did not match waiter state"
         );
-1219
+    }
-1220
     #[cfg(test)]
     pub(super) fn snapshot(&self) -> CellSnapshot {
     pub(super) fn probe(&self) -> AcquisitionProbe {
         let (waiting, demand) = match &self.waiting {
             WaitingQueueState::Empty => (0, None),
             WaitingQueueState::Active { len, demand, .. } => (len.get(), Some(demand.id)),
         };
         CellSnapshot {
         AcquisitionProbe {
             waiting,
             retained: self.records.len(),
             demand,
-1231
+        }
-1232
+    }
-1233
+}
-1234
 /// Result of reserving a requesting waiter for one unlocked delivery.
 pub(super) enum DeliveryReservation {
     /// The current head was reserved and may have a successor demand.
     Reserved {
         waiter: WaiterId,
         successor: Option<DemandSnapshot>,
     },
     /// The delivery no longer matches live cell demand.
     Rejected,
-1244
+}
-1245
 /// Values detached from cell state by cancellation.
 pub(super) struct WaiterCancellation {
     /// Demand retirement and optional successor published after unlocking.
     pub(super) demand_updates: [Option<DemandSnapshot>; 2],
     /// Intermediate and terminal events returned for cleanup after unlocking.
     pub(super) returned_events: [Option<AcquisitionEvent>; 2],
     pub(super) returned_steps: [Option<AcquisitionStep>; 2],
-1252
+}
-1253
 /// Values detached after installing a selected protocol result.
 pub(super) struct WaiterInstall {
     /// Waiter that accepted the result, when one remained eligible.
     pub(super) waiter: Option<WaiterId>,
 /// Values detached after offering a selected protocol result.
 pub(super) struct WaiterResolution {
     /// Demand retirement and optional successor published after unlocking.
     pub(super) demand_updates: [Option<DemandSnapshot>; 2],
     /// Establishment permit or rejected result returned after unlocking.
     pub(super) returned_event: Option<AcquisitionEvent>,
     pub(super) returned_step: Option<AcquisitionStep>,
     /// Waiting task woken after demand publication.
     pub(super) waker: Option<Waker>,
-1262
+}
-1263
 impl WaiterInstall {
 impl WaiterResolution {
     /// Preserves a result when no live attempt can accept HTTP/1.
     fn rejected(result: AcquisitionResult) -> Self {
     fn refused(result: AcquisitionOutcome) -> Self {
         Self {
             waiter: None,
             demand_updates: [None, None],
             returned_event: Some(AcquisitionEvent::Complete(result)),
             returned_step: Some(AcquisitionStep::Resolved(result)),
             waker: None,
-1271
+        }
-1272
+    }
-1273
     /// Reports that no live attempt can accept an HTTP/2 activation.
     fn empty() -> Self {
         Self {
             waiter: None,
             demand_updates: [None, None],
             returned_event: None,
             returned_step: None,
             waker: None,
-1280
+        }
-1281
+    }
-1282
+}
-1283
 /// Invalid state observed while a committed event was crossing to the cell.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) enum ResultInstallError {
 pub(super) enum CellCommitError {
     /// The waiter reserved for the delivery no longer exists.
     MissingWaiter,
     /// The waiter exists but cannot receive the reserved delivery.
     UnexpectedState,
-1291
+}
-1292
 /// Values produced after installing or rejecting a committed event.
 pub(super) struct ResultInstall {
     /// Events rejected by cancellation and returned after unlocking.
     ///
     /// When both are present, the H1 fallback precedes the capacity fallback
     /// so successor demand can reuse the connection before starting another.
     pub(super) returned_events: [Option<AcquisitionEvent>; 2],
     /// Waiting task woken after the delivery fence closes.
     pub(super) waker: Option<Waker>,
     /// Invalid state reported only after the returned result runs its fallback.
     pub(super) error: Option<ResultInstallError>,
     /// Whether cell state became authoritative for the incoming event.
     pub(super) accepted: bool,
 /// Result of committing one exact acquisition step to a reserved waiter.
 pub(super) enum CellCommitOutcome {
     /// Cell state accepted the step and became authoritative for it.
     Committed {
         /// Waiting task woken after assignment settlement.
         waker: Option<Waker>,
     },
     /// The waiter no longer accepted the step; fallback remains valid.
     Refused {
         /// Steps returned after unlocking.
         returned: [Option<AcquisitionStep>; 2],
         /// Waiting task woken after assignment settlement.
         waker: Option<Waker>,
     },
     /// The reserved waiter disappeared or entered an impossible state.
     Invalid {
         /// Steps returned before reporting the invariant failure.
         returned: [Option<AcquisitionStep>; 2],
         /// Waiting task, if one remained attached.
         waker: Option<Waker>,
         /// Invalid cell transition observed under the lock.
         error: CellCommitError,
     },
-1316
+}
-1317
 impl ResultInstall {
     /// Preserves the incoming result for unlocked cleanup after invalid state.
     fn invalid(event: AcquisitionEvent, error: ResultInstallError) -> Self {
         Self {
             returned_events: [Some(event), None],
 impl CellCommitOutcome {
     /// Preserves the incoming step for unlocked cleanup after invalid state.
     fn invalid(step: AcquisitionStep, error: CellCommitError) -> Self {
         Self::Invalid {
             returned: [Some(step), None],
             waker: None,
             error: Some(error),
             accepted: false,
             error,
-1325
+        }
-1326
+    }
-1327
     /// Preserves a losing result for its ordinary unlocked fallback.
     fn rejected(event: AcquisitionEvent) -> Self {
         Self {
             returned_events: [Some(event), None],
     /// Preserves a losing step for its ordinary unlocked fallback.
     fn refused(step: AcquisitionStep) -> Self {
         Self::Refused {
             returned: [Some(step), None],
             waker: None,
             error: None,
             accepted: false,
-1333
+        }
-1334
+    }
-1335
+}
-1336
 /// FIFO state detached while advancing the active head.
 struct RemovedHead {
     /// Identity of the unlinked waiter.
     waiter: WaiterId,
     /// Retired demand generation that represented this head.
     demand: DemandTicket,
     /// New demand published if another waiter became the head.
     successor: Option<DemandSnapshot>,
-1345
+}
-1346
 #[cfg(test)]
 #[derive(Debug)]
 pub(super) struct CellSnapshot {
 pub(super) struct AcquisitionProbe {
     pub(super) waiting: usize,
     pub(super) retained: usize,
     pub(super) demand: Option<DemandId>,
-1353
+}
-1354
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
-1358
     #[test]
     fn h1_required_waiter_is_not_an_h2_candidate() {
     fn h1_required_waiter_is_not_h2_compatible() {
         let mut queue = AcquisitionQueue::default();
         let (waiter, snapshot) = queue.register_waiter(
             ProtocolRequirement::H1Required,
             &EligibilityGroup::Pool,
             false,
         );
-1367
         assert!(snapshot.is_none());
         assert_eq!(Some(waiter), queue.oldest_h1_candidate());
         assert_eq!(None, queue.oldest_h2_candidate());
         assert_eq!(Some(waiter), queue.oldest_h1_compatible_waiter());
         assert_eq!(None, queue.oldest_h2_compatible_waiter());
-1371
+    }
-1372
     #[test]
     fn head_cancellation_uses_the_successors_protocol_requirement() {
         let mut queue = AcquisitionQueue::default();
         let (head, initial) = queue.register_waiter(
             ProtocolRequirement::H2Required,
             &EligibilityGroup::Pool,
             true,
         );

↧ Expand context

         let (_successor, no_new_demand) = queue.register_waiter(
             ProtocolRequirement::H1Compatible,
             &EligibilityGroup::Pool,
             true,
         );
         assert!(initial.is_some());
         assert!(no_new_demand.is_none());
-1388
         let cancelled = queue
             .cancel_waiter(head, &EligibilityGroup::Pool)
             .expect("head waiter was not cancelled");
         assert_eq!(
             Some(DemandSnapshot::active(
                 DemandId::from_u64(1),
                 SnapshotVersion::INITIAL,
                 ProtocolRequirement::H1Compatible,
                 EligibilityGroup::Pool,
             )),
             cancelled.demand_updates[1]
         );

AWS SDK

Files changed:

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/cell/h2/request.rs

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/cell/waiters.rs