AWS SDK

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

@@ -1,1 +188,256 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5

 //! State local to one partition and origin.
 //!
 //! An [`OriginCell`] owns its acquisition order and protocol records. A
 //! bounded cell publishes aggregate demand to [`OriginAdmission`] after
 //! releasing its lock, and admission returns capacity through the same
 //! unlocked boundary.
 //!
 //! [`AcquisitionQueue`] owns every live acquisition attempt, including the
 //! bounded FIFO, delivery crossings, establishment launch, cancellation, and
 //! terminal results. Its module documents the complete transition model.
 //! [`CellState`] couples that model to HTTP/1 sender residence so a local
 //! return either satisfies one acquisition or becomes idle, never both.
 //! [`CellState`] couples that model to HTTP/1 sender residence, HTTP/2
 //! flight and generation residence, generation gates, and the cross-cell reuse
 //! reservation. A local HTTP/1 return either satisfies one acquisition or
 //! becomes idle, never both. An HTTP/2 activation becomes visible only with a
 //! prospective generation lease and one recorded gate opportunity.
 //!
 //! Permits, senders, admission updates, and task wakes are detached from
 //! mutable state before their fallback or callback runs. This keeps drop and
 //! wake paths outside the cell lock.
-25
 pub(super) mod h1;
 pub(super) mod h2;
 mod waiters;
-29
 #[cfg(test)]
 use self::h1::H1CloseHandle;
 #[cfg(all(test, not(smithy_http_client_loom)))]
 use self::h1::H1DriverGuard;
 use self::h1::{H1Records, H1ReuseReservation, H1Selection, H1Sender, OwnedH1, ProvisionalH1};
 #[cfg(test)]
 use self::h1::H1Sender;
 use self::h1::{H1Records, H1ReuseReservation, H1Selection, OwnedH1};
 #[cfg(test)]
 use self::waiters::CellSnapshot;
 pub(in crate::client::pool) use self::waiters::WaiterId;
 use self::waiters::{AcquisitionQueue, DeliveryReservation, ResultInstallError};
 use super::admission::reuse::{
     H1Availability, H1AvailabilitySnapshot, PreparedReuseInstall, ReuseCandidate, ReuseId,
     ReuseInstallResult,
 };
 #[cfg(test)]
 use super::admission::DemandSnapshot;
 use super::admission::reuse::{H1Availability, H1AvailabilitySnapshot, ReuseId};
 use super::admission::{
     AdmissionAction, CapacityLease, DeliveryGuard, OriginAdmission, ProtocolRequirement,
     AdmissionAction, CapacityLease, DeliveryGuard, DemandSnapshot, H2AdvertisementSnapshot,
     OriginAdmission, ProtocolRequirement,
 };
 use super::connection::CloseReason;
 #[cfg(test)]
 use super::connection::ConnectionInfo;
 use super::connection::{CloseReason, ConnectionState};
 #[cfg(test)]
 use super::connection::ConnectionState;
 use super::maintenance::PartitionMaintenance;
 use super::origin::OriginKey;
 use super::partition::{EligibilityGroup, PartitionId};
 use crate::sync::{Arc, Mutex};
 #[cfg(test)]
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use std::task::{Context, Poll};
 use std::time::SystemTime;
-60
 /// Stable identity of an [`OriginCell`].
 #[derive(Clone, Debug, Eq, Hash, PartialEq)]
 pub(crate) struct CellId {
     /// Partition that owns this cell's connection work.
     partition: PartitionId,
     /// Canonical origin whose requests share this cell.
     origin: OriginKey,
-68
+}
-69
 impl CellId {
     pub(super) fn new(partition: PartitionId, origin: OriginKey) -> Self {
         Self { partition, origin }
-73
+    }
-74
     pub(crate) fn partition(&self) -> PartitionId {
         self.partition
-77
+    }
-78
     pub(crate) fn origin(&self) -> &OriginKey {
         &self.origin
-81
+    }
-82
+}
-83
 /// Stable state shared by requests for one partition and canonical origin.
 pub(crate) struct OriginCell {
     /// Complete identity used by partition and admission indexes.
     id: CellId,
     /// Partitions whose connections may satisfy this cell's demand.
     eligibility_group: EligibilityGroup,
     /// Origin-wide authority present only when capacity is bounded.
     admission: Option<Arc<OriginAdmission>>,
     /// Partition-owned idle deadline scheduler.
     maintenance: Option<Arc<PartitionMaintenance>>,
     /// Waiter order and protocol records protected by one cell-local lock.
     state: Mutex<CellState>,
-96
+}
-97
 /// Mutable state protected by one partition-origin cell lock.
 ///
 /// Waiter outcomes and HTTP/1 residence share this lock so a returning sender
 /// either becomes visible to one waiter or enters idle storage, never both.
 /// Waiter outcomes and protocol residence share this lock. HTTP/1 returns,
 /// HTTP/2 activation gates, flight participants, and cancellation therefore
 /// commit against one acquisition order.
 #[derive(Debug, Default)]
 struct CellState {
     /// Cell-local acquisition order and delivered results.
     waiters: AcquisitionQueue,
     /// Cell-owned HTTP/1 records and reusable sender order.
     h1: H1Records,
     /// Cell-owned HTTP/2 flight and installed generations.
     h2: h2::H2Records,
     /// One cross-cell reuse reservation and its local fairness debt.
     reuse: H1ReuseReservation,
     /// Last HTTP/1 availability published to bounded-origin admission.
     published_h1_availability: Option<H1Availability>,
     /// Revision used to reject availability reports that cross out of order.
     h1_availability_revision: u64,
     /// Last complete H2 advertisement reported to origin admission.
     published_h2_advertisement: Option<(h2::H2GenerationId, bool)>,
     /// Revision used to reject H2 advertisements that cross out of order.
     h2_advertisement_revision: u64,
-121
+}
-122
 impl CellState {
     /// Checks the coupled waiter, sender, and reuse-reservation state machines.
     fn assert_consistent(&self) {
         #[cfg(any(debug_assertions, test))]
         if std::thread::panicking() {
             return;
-+
+        }
         self.waiters.assert_consistent();
         self.h1.assert_consistent();
         self.h2.assert_consistent();
         #[cfg(any(debug_assertions, test))]
         self.h2.assert_pending_waiters(&self.waiters);
         self.reuse
             .assert_consistent(self.h1.supports_installed_reuse());
-137
+    }
-138
     /// Reports the HTTP/1 availability admission needs without exposing cell internals.
     fn h1_availability(&self) -> H1Availability {
         let local_h1_demand = self.waiters.can_accept_h1();
         H1Availability {
             advertised: self.h1.has_returnable(),
             blocked: !self.reuse.is_available()
                 || self.reuse.blocks_peer_reuse(local_h1_demand)
                 || self.waiters.has_prior_h1_candidate(),
-147
+        }
-148
+    }
-149
     /// Returns an availability update only when admission's view must change.
     fn take_h1_availability_update(&mut self) -> Option<H1AvailabilitySnapshot> {
         let current = self.h1_availability();
         if self.published_h1_availability == Some(current) {
             return None;
-155
+        }
         Some(self.record_h1_availability(current))
-157
+    }
-158
     /// Returns the current versioned availability for a reuse transition.
     fn report_h1_availability(&mut self) -> H1AvailabilitySnapshot {
         let current = self.h1_availability();
         if self.published_h1_availability == Some(current) {
             return H1AvailabilitySnapshot::new(self.h1_availability_revision, current);
-164
+        }
         self.record_h1_availability(current)
-166
+    }
-167
     /// Advances the revision and records changed HTTP/1 availability.
     fn record_h1_availability(&mut self, availability: H1Availability) -> H1AvailabilitySnapshot {
         self.h1_availability_revision = self
             .h1_availability_revision
             .checked_add(1)
             .expect("HTTP/1 availability revision exhausted");
         self.published_h1_availability = Some(availability);
         H1AvailabilitySnapshot::new(self.h1_availability_revision, availability)
-176
+    }
-177
     /// Returns an H2 advertisement only when admission's view must change.
     fn take_h2_advertisement_update(&mut self) -> Option<H2AdvertisementSnapshot> {
         let current = self.h2_advertisement();
         if self.published_h2_advertisement == current {
             return None;
-+
+        }
         Some(self.record_h2_advertisement(current))
-+
+    }
-+
     /// Returns the current versioned H2 advertisement for a reclaim transition.
     fn report_h2_advertisement(&mut self) -> H2AdvertisementSnapshot {
         let current = self.h2_advertisement();
         if self.published_h2_advertisement == current {
             return Self::h2_advertisement_snapshot(self.h2_advertisement_revision, current);
-+
+        }
         self.record_h2_advertisement(current)
-+
+    }
-+
     /// Returns the exact publishable generation and its idle state.
     fn h2_advertisement(&self) -> Option<(h2::H2GenerationId, bool)> {
         self.h2
             .publishable_generation()
             .map(|generation| (generation, self.h2.is_idle(generation)))
-+
+    }
-+
     /// Advances the revision and records changed H2 availability.
     fn record_h2_advertisement(
         &mut self,
         advertisement: Option<(h2::H2GenerationId, bool)>,
     ) -> H2AdvertisementSnapshot {
         self.h2_advertisement_revision = self
             .h2_advertisement_revision
             .checked_add(1)
             .expect("HTTP/2 advertisement revision exhausted");
         self.published_h2_advertisement = advertisement;
         Self::h2_advertisement_snapshot(self.h2_advertisement_revision, advertisement)
-+
+    }
-+
     fn h2_advertisement_snapshot(
         revision: u64,
         advertisement: Option<(h2::H2GenerationId, bool)>,
     ) -> H2AdvertisementSnapshot {
         match advertisement {
             Some((generation, true)) => H2AdvertisementSnapshot::idle(revision, generation),
             Some((generation, false)) => H2AdvertisementSnapshot::accepting(revision, generation),
             None => H2AdvertisementSnapshot::unavailable(revision),
-+
+        }
-+
+    }
-+
     /// Atomically decides whether a peer may reserve this cell's connection.
     fn install_reuse(&mut self, reuse_id: ReuseId) -> ReuseInstall {
         let local_h1_demand = self.waiters.can_accept_h1();
         if self.reuse.blocks_peer_reuse(local_h1_demand)
             || self.waiters.has_prior_h1_candidate()
             || !self.reuse.is_available()
-233
+        {
             let report = self.report_h1_availability();
             self.assert_consistent();
             return ReuseInstall::Rejected(report);

↧ Expand context

-237
+        }
-238
         if let Some(owner) = self.h1.take_idle_for_reuse() {
             assert!(
                 self.reuse.install_resolving(reuse_id),
                 "available HTTP/1 connection could not install its reuse reservation"
             );
             self.assert_consistent();
             return ReuseInstall::Candidate(owner);
-246
+        }
         if self.h1.has_returnable() {
             assert!(
                 self.reuse.install(reuse_id),
                 "available HTTP/1 connection could not install its reuse reservation"
             );
             self.assert_consistent();
             return ReuseInstall::Installed;
-254
+        }
-255
         let report = self.report_h1_availability();

@@ -196,264 +1120,824 @@

↥ Expand context

         let local_h1_demand = self.waiters.can_accept_h1();
         self.reuse.clear_unused_turn(local_h1_demand);
         self.assert_consistent();
         self.report_h1_availability()
-268
+    }
-269
     /// Clears a reservation before its provisional sender follows ordinary return.
     fn reject_reuse_candidate(&mut self, reuse_id: ReuseId) {
         self.reuse.reject(reuse_id);
         self.assert_consistent();
-274
+    }
-275
     /// Revalidates a reuse operation and sender residence as one cell transition.
     fn commit_reuse(&mut self, reuse_id: ReuseId, owner: &OwnedH1) -> bool {
         let committed = self.reuse.names(reuse_id) && self.h1.commit_return_to_waiter(owner);
         self.assert_consistent();
         committed
-281
+    }
-282
     /// Completes or rejects a reservation after its external action resolves.

     fn finish_reuse(&mut self, reuse_id: ReuseId, transferred: bool) -> H1AvailabilitySnapshot {
         if transferred {
             let local_h1_demand = self.waiters.can_accept_h1();
             self.reuse.complete_transfer(reuse_id, local_h1_demand);
         } else {
             self.reuse.reject(reuse_id);
-290
+        }
         self.assert_consistent();
         self.report_h1_availability()
-293
+    }
-+
     /// Removes active HTTP/2-compatible demand already served by visible H2 state.
     fn publishable_demand_updates(
         &self,
         updates: [Option<DemandSnapshot>; 2],
     ) -> [Option<DemandSnapshot>; 2] {
         let h2_visible = self.h2.has_visible_h2();
         updates.map(|snapshot| {
             snapshot
                 .filter(|snapshot| !h2_visible || !snapshot.is_active() || !snapshot.accepts_h2())
         })
-+
+    }
-306
+}
-307
 /// Cell-local result of attempting to install one peer reuse reservation.
 enum ReuseInstall {
     /// The reservation will intercept a future sender return.
     Installed,
     /// An idle sender was detached for immediate resolution.
     Candidate(OwnedH1),
     /// The cell could not reserve a connection for the operation.
     Rejected(H1AvailabilitySnapshot),
-316
+}
-317
 /// A terminal result that may satisfy an acquisition waiter.
 #[derive(Debug)]
 pub(super) enum AcquisitionResult {
     /// An exclusive HTTP/1 sender selected from an installed cell-owned record.
     H1(H1Selection),
     /// A prospective request lease against one HTTP/2 generation.
     H2(h2::H2Activation),
     /// Establishment failed before producing a dispatchable connection.
     Failed(ConnectorError),
     /// An HTTP/2 generation closed before serving a transferred attempt.
     Reacquire,
-329
+}
-330
 /// One event observed while driving an acquisition attempt.
 #[derive(Debug)]
 pub(super) enum AcquisitionEvent {
     /// Capacity is available and one establishment attempt may start.
     Establish(EstablishmentPermit),
     /// A returned sender or establishment produced the terminal result.
     Complete(AcquisitionResult),
-338
+}
-339
 /// Optional bounded-origin capacity for one establishment attempt.
 ///
 /// Unbounded origins carry no lease. For bounded origins, dropping this value
 /// before connection installation returns the permit to admission.
 #[derive(Debug)]
 pub(super) struct EstablishmentPermit {
     /// Bounded-origin slot transferred into a connection on success.
     lease: Option<CapacityLease>,
-348
+}
-349
 impl EstablishmentPermit {
     /// Creates a permit for an origin without a connection bound.
     fn unbounded() -> Self {
         Self { lease: None }
-354
+    }
-355
     /// Takes ownership of one bounded-origin capacity lease.
     pub(in crate::client::pool) fn bounded(lease: CapacityLease) -> Self {
         Self { lease: Some(lease) }
-359
+    }
-360
     /// Transfers bounded capacity into an installed connection, when present.
     pub(super) fn into_lease(self) -> Option<CapacityLease> {
         self.lease
-364
+    }
-365
+}
-366
 impl OriginCell {
     /// Creates a cell before its one-time publication in a partition registry.
     pub(super) fn new(
         partition: PartitionId,
         origin: OriginKey,
         eligibility_group: EligibilityGroup,
         admission: Option<Arc<OriginAdmission>>,
         maintenance: Option<Arc<PartitionMaintenance>>,
     ) -> Self {
         Self {
             id: CellId::new(partition, origin),
             eligibility_group,
             admission,
             maintenance,
             state: Mutex::new(CellState::default()),
-382
+        }
-383
+    }
-384
     /// Returns the next idle deadline from immutable pool policy.
     fn idle_deadline(&self) -> Option<SystemTime> {
     pub(in crate::client::pool) fn idle_deadline(&self) -> Option<SystemTime> {
         self.maintenance
             .as_ref()
             .and_then(|maintenance| maintenance.idle_deadline())
-390
+    }
-391
     /// Publishes a newly idle deadline to partition maintenance.
     fn notify_maintenance(&self, deadline: Option<SystemTime>) {
         if let Some(maintenance) = &self.maintenance {
             maintenance.notify_deadline(deadline);
-396
+        }
-397
+    }
-398
     pub(crate) fn id(&self) -> &CellId {
         &self.id
-401
+    }
-402
     #[cfg(test)]
     pub(crate) fn eligibility_group(&self) -> &EligibilityGroup {
         &self.eligibility_group
-406
+    }
-407
     #[cfg(test)]
     pub(crate) fn admission(&self) -> Option<&Arc<OriginAdmission>> {
         self.admission.as_ref()
-411
+    }
-412
     /// Installs a fresh H1 record selected by its launching request.
     ///
     /// # Panics
     ///
     /// Panics if this cell already contains the connection identity.
     pub(super) fn install_selected_h1(
         cell: &Arc<Self>,
         connection: Arc<ConnectionState>,
         sender: H1Sender,
     ) -> H1Selection {
         let (installed, availability) = {
             let mut state = cell.state.lock();
             let installed = state.h1.install_selected(connection, sender);
             state.assert_consistent();
             (installed, state.take_h1_availability_update())
         };
         cell.publish_h1_availability(availability);
         match installed {
             Ok(owner) => H1Selection::new(cell, owner),
             Err(owner) => {
                 owner
                     .connection()
                     .logical_close(CloseReason::ProtocolClosed);
                 drop(owner);
                 panic!("duplicate HTTP/1 connection identity installed in one cell");
--
+            }
--
+        }
--
+    }
--
     /// Installs an H1 record whose launching acquisition already completed.
     ///
     /// # Panics
     ///
     /// Panics if this cell already contains the connection identity.
     #[cfg(test)]
     pub(super) fn install_idle_h1(
         cell: &Arc<Self>,
         connection: Arc<ConnectionState>,
         sender: H1Sender,
     ) {
         let deadline = cell.idle_deadline();
         let (installed, availability) = {
             let mut state = cell.state.lock();
             let installed = state.h1.install_idle(connection, sender, deadline);
             state.assert_consistent();
             (installed, state.take_h1_availability_update())
         };
         cell.publish_h1_availability(availability);
         if let Err(owner) = installed {
             owner
                 .connection()
                 .logical_close(CloseReason::ProtocolClosed);
             drop(owner);
             panic!("duplicate HTTP/1 connection identity installed in one cell");
--
+        }
         cell.notify_maintenance(deadline);
--
+    }
--
     /// Selects the newest reusable H1 sender without origin-wide coordination.
     pub(super) fn select_h1(cell: &Arc<Self>) -> Option<H1Selection> {
         let (owner, availability) = {
             let mut state = cell.state.lock();
             let owner = state.h1.select_idle();
             if owner.is_some() {
                 state.reuse.consume_local_turn();
--
+            }
             state.assert_consistent();
             let availability = cell
                 .admission
                 .as_ref()
                 .and_then(|_| state.take_h1_availability_update());
             (owner, availability)
         };
         cell.publish_h1_availability(availability);
         let owner = owner?;
         Some(H1Selection::new(cell, owner))
--
+    }
--
     /// Reserves this cell's selected connection for one peer reuse operation.
     pub(in crate::client::pool) fn install_h1_reuse(
         cell: &Arc<Self>,
         origin: Arc<OriginAdmission>,
         prepared: PreparedReuseInstall,
     ) -> Option<AdmissionAction> {
         let decision = {
             let mut state = cell.state.lock();
             state.install_reuse(prepared.id)
         };
--
         let result = match decision {
             ReuseInstall::Installed => ReuseInstallResult::Installed,
             ReuseInstall::Candidate(owner) => {
                 let provisional = ProvisionalH1::new(cell, owner);
                 ReuseInstallResult::Candidate(ReuseCandidate::new(
                     origin.clone(),
                     prepared.id,
                     cell.id.partition(),
                     provisional,
                 ))
--
+            }
             ReuseInstall::Rejected(availability) => ReuseInstallResult::Rejected(availability),
         };
         OriginAdmission::finish_h1_reuse_install(&origin, prepared.id, cell.id.partition(), result)
--
+    }
--
     /// Clears an installed or resolving reservation after request cancellation.
     pub(in crate::client::pool) fn cancel_h1_reuse(
         &self,
         reuse_id: ReuseId,
     ) -> H1AvailabilitySnapshot {
         self.state.lock().cancel_reuse(reuse_id)
--
+    }
--
     /// Returns a rejected provisional sender through ordinary connection-owning cell handling.
     pub(in crate::client::pool) fn reject_h1_reuse_candidate(
         cell: &Arc<Self>,
         reuse_id: ReuseId,
         provisional: ProvisionalH1,
     ) -> H1AvailabilitySnapshot {
--
+        {
             let mut state = cell.state.lock();
             state.reject_reuse_candidate(reuse_id);
--
+        }
         drop(provisional);
         let mut state = cell.state.lock();
         state.assert_consistent();
         state.report_h1_availability()
--
+    }
--
     /// Revalidates a reuse operation and commits its provisional sender for dispatch.
     pub(in crate::client::pool) fn commit_h1_reuse(
         cell: &Arc<Self>,
         reuse_id: ReuseId,
         provisional: ProvisionalH1,
     ) -> Result<H1Selection, ProvisionalH1> {
         let (connection_cell, owner) = provisional.into_parts();
         let committed = cell.state.lock().commit_reuse(reuse_id, &owner);
         if committed {
             Ok(H1Selection::new(cell, owner))
         } else {
             Err(ProvisionalH1::from_parts(connection_cell, owner))
--
+        }
--
+    }
--
     /// Closes a selected sender and records fairness only if close wins.
     pub(in crate::client::pool) fn reclaim_h1_reuse(
         cell: &Arc<Self>,
         reuse_id: ReuseId,
         provisional: ProvisionalH1,
     ) -> Result<(H1AvailabilitySnapshot, bool), ProvisionalH1> {
         let (connection_cell, owner) = provisional.into_parts();
         if !cell.state.lock().reuse.names(reuse_id) {
             return Err(ProvisionalH1::from_parts(connection_cell, owner));
--
+        }
--
         let close_won = Self::retire_h1_owner(cell, owner, CloseReason::Reclaimed);
         let availability = {
             let mut state = cell.state.lock();
             let availability = state.finish_reuse(reuse_id, close_won);
             state.assert_consistent();
             availability
         };
         Ok((availability, close_won))
--
+    }
--
     /// Completes local reuse state after a requesting cell accepts the sender.
     pub(in crate::client::pool) fn complete_h1_reuse(
         &self,
         reuse_id: ReuseId,
         transferred: bool,
     ) -> H1AvailabilitySnapshot {
         self.state.lock().finish_reuse(reuse_id, transferred)
--
+    }
--
     /// Publishes this cell's HTTP/1 availability when the origin is bounded.
     fn publish_h1_availability(&self, availability: Option<H1AvailabilitySnapshot>) {
         if let (Some(admission), Some(availability)) = (&self.admission, availability) {
             OriginAdmission::update_h1_availability(
                 admission,
                 self.id.partition(),
                 self.eligibility_group.clone(),
                 availability,
             );
--
+        }
--
+    }
--
     /// Returns a reusable sender to the oldest compatible waiter or idle set.
     ///
     /// Demand publication, task wakeup, and any rejected-result fallback all
     /// run after the cell lock is released. `owner` remains outside the locked
     /// scope so sender or connection drop cannot run while the cell guard is
     /// live.
     fn return_h1_owner(cell: &Arc<Self>, owner: OwnedH1) {
         let connection_id = owner.id();
         let mut owner = Some(owner);
         let mut installation = None;
         let mut intercepted = None;
         let mut rejected_reuse = None;
         let idle_deadline = cell.idle_deadline();
         let should_retire = {
             let mut state = cell.state.lock();
             let returnable = state
                 .h1
                 .accepts_return(owner.as_ref().expect("HTTP/1 owner disappeared"));
             if !returnable {
                 if let Some(reuse_id) = state.reuse.intercept_return() {
                     let snapshot = state.finish_reuse(reuse_id, false);
                     rejected_reuse = Some((reuse_id, snapshot));
                 } else {
                     state.assert_consistent();
--
+                }
                 true
             } else if let Some(reuse_id) = state.reuse.intercept_return() {
                 if state
                     .h1
                     .reserve_for_reuse(owner.as_ref().expect("HTTP/1 owner disappeared"))
--
+                {
                     state.assert_consistent();
                     intercepted = Some((
                         reuse_id,
                         ProvisionalH1::new(cell, owner.take().expect("HTTP/1 owner disappeared")),
                     ));
                     false
                 } else {
                     let snapshot = state.finish_reuse(reuse_id, false);
                     rejected_reuse = Some((reuse_id, snapshot));
                     true
--
+                }
             } else if state.waiters.can_accept_h1()
                 && state
                     .h1
                     .commit_return_to_waiter(owner.as_ref().expect("HTTP/1 owner disappeared"))
--
+            {
                 state.reuse.consume_local_turn();
                 state.assert_consistent();
                 let mut returned = owner.take().expect("HTTP/1 owner disappeared");
                 returned.mark_reused();
                 installation = Some(state.waiters.install_returned_h1(
                     || AcquisitionResult::H1(H1Selection::new(cell, returned)),
                     &cell.eligibility_group,
                 ));
                 false
             } else {
                 let returned = state.h1.return_idle(
                     owner.take().expect("HTTP/1 owner disappeared"),
                     idle_deadline,
                 );
                 match returned {
                     Ok(()) => {
                         state.assert_consistent();
                         false
--
+                    }
                     Err(returned) => {
                         owner = Some(returned);
                         true
--
+                    }
--
+                }
--
+            }
         };
--
         if should_retire {
             if tracing::level_enabled!(tracing::Level::TRACE) {
                 trace_h1_return(cell, connection_id, H1ReturnTrace::Rejected);
--
+            }
             if let Some((reuse_id, snapshot)) = rejected_reuse {
                 let admission = cell
                     .admission
                     .as_ref()
                     .expect("an HTTP/1 reuse operation requires bounded admission");
                 OriginAdmission::reject_returned_h1_reuse(
                     admission,
                     reuse_id,
                     cell.id.partition(),
                     snapshot,
                 );
--
+            }
             Self::retire_h1_owner(
                 cell,
                 owner.take().expect("retired HTTP/1 owner disappeared"),
                 CloseReason::ProtocolClosed,
             );
             return;
--
+        }
--
         if let Some((reuse_id, provisional)) = intercepted {
             if tracing::level_enabled!(tracing::Level::TRACE) {
                 trace_h1_return(cell, connection_id, H1ReturnTrace::ReuseIntercepted);
--
+            }
             let admission = cell
                 .admission
                 .as_ref()
                 .expect("an HTTP/1 reuse operation requires bounded admission");
             let candidate = ReuseCandidate::new(
                 admission.clone(),
                 reuse_id,
                 cell.id.partition(),
                 provisional,
             );
             let action = OriginAdmission::resolve_h1_reuse(admission, reuse_id, candidate);
             OriginAdmission::drive(action);
             return;
--
+        }
--
         let Some(installation) = installation else {
             if tracing::level_enabled!(tracing::Level::TRACE) {
                 trace_h1_return(cell, connection_id, H1ReturnTrace::Idle);
--
+            }
             cell.notify_maintenance(idle_deadline);
             if cell.admission.is_some() {
                 let availability = cell.state.lock().take_h1_availability_update();
                 cell.publish_h1_availability(availability);
--
+            }
             return;
         };
         if tracing::level_enabled!(tracing::Level::TRACE) {
             trace_h1_return(cell, connection_id, H1ReturnTrace::LocalDemand);
--
+        }
         if let Some(admission) = &cell.admission {
             for snapshot in installation.demand_updates.into_iter().flatten() {
                 OriginAdmission::publish_demand(admission, cell.id.partition(), snapshot);
--
+            }
--
+        }
         drop(installation.returned_event);
         if let Some(waker) = installation.waker {
             waker.wake();
--
+        }
         if cell.admission.is_some() {
             let availability = cell.state.lock().take_h1_availability_update();
             cell.publish_h1_availability(availability);
--
+        }
--
+    }
--
     /// Retires an externally owned H1 sender and removes its connection-owning cell record.
     ///
     /// Returns whether this path won the connection's logical-close race.
     fn retire_h1_owner(cell: &Arc<Self>, owner: OwnedH1, reason: CloseReason) -> bool {
         let should_close = {
             let mut state = cell.state.lock();
             let should_close = state.h1.close_owned(&owner);
             state.assert_consistent();
             should_close
         };
--
         let id = owner.id();
         let won = should_close && owner.connection().logical_close(reason);
         drop(owner);
--
         let mut state = cell.state.lock();
         state.h1.finish_close(id);
         state.assert_consistent();
         let availability = state.take_h1_availability_update();
         drop(state);
         cell.publish_h1_availability(availability);
         won
--
+    }
--
     /// Begins close for an installed H1 generation named without its sender.
     ///
     /// Returns whether this signal won the connection's logical-close race.
     fn close_h1(cell: &Arc<Self>, id: ConnectionId, reason: CloseReason) -> bool {
         let Some((close, availability)) = ({
             let mut state = cell.state.lock();
             let close = state.h1.begin_close(id);
             state.assert_consistent();
             close.map(|close| (close, state.take_h1_availability_update()))
         }) else {
             return false;
         };
--
         cell.publish_h1_availability(availability);
         let remove_record = close.sender.is_some();
         let won = close.connection.logical_close(reason);
         drop(close.sender);
         if remove_record {
             let mut state = cell.state.lock();
             state.h1.finish_close(id);
             state.assert_consistent();
--
+        }
         won
--
+    }
--
     /// Closes idle records whose maintenance deadline has elapsed.
     /// Closes H1 records and H2 generations whose idle deadline elapsed.
     pub(super) fn expire_idle(cell: &Arc<Self>, now: SystemTime) {
         let expired = cell.state.lock().h1.expired_idle(now);
         for id in expired {
         let (expired_h1, expired_h2) = {
             let state = cell.state.lock();
             (state.h1.expired_idle(now), state.h2.expired(now))
         };
         for id in expired_h1 {
             if Self::close_h1(cell, id, CloseReason::IdleTimeout) {
                 tracing::trace!(
                     connection_id = %id,
                     connection_partition = ?cell.id.partition(),
                     origin_scheme = %cell.id.origin().scheme(),
                     origin_host = cell.id.origin().host(),
                     origin_port = ?cell.id.origin().port(),
                     "HTTP/1 idle connection expired"
                 );
-429
+            }
-430
+        }
         if let Some(generation) = expired_h2 {
             if Self::close_h2(cell, generation, CloseReason::IdleTimeout) {
                 tracing::trace!(
                     connection_partition = ?cell.id.partition(),
                     origin_scheme = %cell.id.origin().scheme(),
                     origin_host = cell.id.origin().host(),
                     origin_port = ?cell.id.origin().port(),
                     h2_generation = ?generation,
                     "HTTP/2 idle generation expired"
                 );
-+
+            }
-+
+        }
-443
+    }
-444
     /// Returns this cell's nearest reusable H1 deadline.
     /// Returns this cell's nearest reusable connection deadline.
     pub(super) fn nearest_idle_deadline(&self) -> Option<SystemTime> {
         self.state.lock().h1.nearest_idle_deadline()
         let state = self.state.lock();
-+
+        [
             state.h1.nearest_idle_deadline(),
             state.h2.nearest_idle_deadline(),
-+
+        ]
         .into_iter()
         .flatten()
         .min()
-455
+    }
-456
     /// Logically closes every retained H1 record for pool shutdown.
     pub(super) fn close_all_h1(cell: &Arc<Self>, reason: CloseReason) {
         let ids = cell.state.lock().h1.connection_ids();
         for id in ids {
     /// Logically closes every retained connection for pool shutdown.
     pub(super) fn close_all(cell: &Arc<Self>, reason: CloseReason) {
         let (h1, h2) = {
             let state = cell.state.lock();
             (state.h1.connection_ids(), state.h2.generation_ids())
         };
         for id in h1 {
             Self::close_h1(cell, id, reason);
-465
+        }
         for generation in h2 {
             Self::close_h2(cell, generation, reason);
-+
+        }
-469
+    }
-470
     /// Applies one materialized acquisition delivery after admission unlock.
     ///
     /// Connection-cell H1 revalidation has already completed, so reserving
     /// the requesting cell cannot be followed by another fallible connection-
     /// cell transition.
     ///
     /// # Panics
     ///
     /// Panics if a waiter reserved by this function disappears or enters a
     /// state that cannot accept the committed acquisition event.
     pub(in crate::client::pool) fn receive_delivery(
         cell: &Arc<Self>,
         delivery: DeliveryGuard,
     ) -> Option<AdmissionAction> {
         let reservation = {
             let mut state = cell.state.lock();
             state
                 .waiters
                 .reserve_delivery_waiter(delivery.demand(), &cell.eligibility_group)
         };
-491
         let DeliveryReservation::Reserved { waiter, successor } = reservation else {
             return delivery.reject(None);
         };
-495
         let (event, acknowledgement) = delivery.commit(successor);
         let installation = {
         let (event, mut acknowledgement) = delivery.commit(successor);
         let (installation, suppress_successor) = {
             let mut state = cell.state.lock();
             match event {
             let installation = match event {
                 AcquisitionEvent::Establish(permit) => {
                     state.waiters.install_capacity(waiter, permit)
-502
+                }
                 AcquisitionEvent::Complete(result) => {
                     state.waiters.install_borrowed_h1(waiter, result)
-505
+                }
--
+            }
             };
             (installation, state.h2.has_visible_h2())
         };
         if suppress_successor {
             acknowledgement.suppress_h2_successor();
-+
+        }
-512
         let accepted = installation.accepted;
         let error = installation.error;
         let waker = installation.waker;
         let next = if accepted {
             drop(installation.returned_events);
             acknowledgement.accept()
         } else {
             acknowledgement.reject(installation.returned_events)
         };
         if let Some(waker) = waker {
             waker.wake();
-524
+        }
         if let Some(error) = error {
             match error {
                 ResultInstallError::MissingWaiter => {
                     panic!("reserved waiter disappeared before acquisition delivery")
-529
+                }
                 ResultInstallError::UnexpectedState => {
                     panic!("reserved waiter entered an invalid acquisition-delivery state")
-532
+                }
-533
+            }
-534
+        }
         next
-536
+    }
-537
     /// Returns the number of retained acquisition waiters for boundary tests.
     #[cfg(all(test, feature = "rt-tokio"))]
     #[cfg(test)]
     pub(super) fn retained_waiters_for_test(&self) -> usize {
         self.state.lock().waiters.snapshot().retained
-542
+    }
-543
     /// Returns H1 record counts for focused ownership tests.
     #[cfg(test)]
     fn h1_counts(&self) -> (usize, usize) {
         self.state.lock().h1.counts()
--
+    }
--
     /// Returns the sole installed HTTP/1 connection for focused dispatch tests.
     #[cfg(all(test, feature = "rt-tokio"))]
     pub(super) fn only_h1_connection_for_test(&self) -> Arc<ConnectionState> {
         self.state.lock().h1.only_connection_for_test()
--
+    }
--
     /// Registers one acquisition waiter in cell-local arrival order.
     ///
     /// The waiter and its demand snapshot are committed under the cell lock.
     /// A bounded cell publishes the snapshot to origin admission only after
     /// that lock is released. An unbounded cell retains only local order.
     pub(super) fn register_waiter(&self, requirement: ProtocolRequirement) -> WaiterId {
         let (waiter, snapshot) = {
             let mut state = self.state.lock();
             state.waiters.register_waiter(
     pub(super) fn register_waiter(cell: &Arc<Self>, requirement: ProtocolRequirement) -> WaiterId {
         let (waiter, snapshot, h2_visible) = {
             let mut state = cell.state.lock();
             let (waiter, snapshot) = state.waiters.register_waiter(
                 requirement,
                 &self.eligibility_group,
                 self.admission.is_some(),
                 &cell.eligibility_group,
                 cell.admission.is_some(),
             );
-+
+            (
                 waiter,
                 snapshot,
                 state.h2.has_visible_h2() && requirement.accepts_h2(),
-561
+            )
         };
-563
         if let (Some(admission), Some(snapshot)) = (&self.admission, snapshot) {
             OriginAdmission::publish_demand(admission, self.id.partition(), snapshot);
         if let (Some(admission), Some(snapshot), false) = (&cell.admission, snapshot, h2_visible) {
             OriginAdmission::publish_demand(admission, cell.id.partition(), snapshot);
-566
+        }
         Self::service_h2_waiters(cell);
         Self::service_peer_h2_waiters(cell);
         waiter
-570
+    }
-571
     /// Cancels a waiter and re-examines local idle H1 state atomically.
     ///
     /// A newly compatible successor may be serviceable by an H1 that an older
     /// HTTP/2-only head could not use. The local sender is selected under the
     /// same cell lock as cancellation, so it never needs a same-cell peer
     /// reuse operation. Demand publication, fallback, and waking remain outside the lock.
     pub(super) fn cancel_waiter(cell: &Arc<Self>, waiter: WaiterId) -> bool {
         let transition = {
             let mut state = cell.state.lock();
             state.h2.cancel_flight_participant(waiter);
             state.h2.cancel_peer_activation(waiter);
             state.h2.cancel_pending_waiter(waiter);
             state
                 .waiters
                 .cancel_waiter(waiter, &cell.eligibility_group)
                 .map(|cancelled| {
                     let local_install = if state.waiters.can_accept_h1() {
                 .map(|mut cancelled| {
                     let mut local_install = if state.waiters.can_accept_h1() {
                         state.h1.select_idle().map(|owner| {
                             state.reuse.consume_local_turn();
                             state.waiters.install_returned_h1(
                                 || AcquisitionResult::H1(H1Selection::new(cell, owner)),
                                 &cell.eligibility_group,
-594
+                            )
                         })
                     } else {
                         None
                     };
                     cancelled.demand_updates =
                         state.publishable_demand_updates(cancelled.demand_updates);
                     if let Some(install) = &mut local_install {
                         if let Some(waiter) = install.waiter {
                             state.h2.cancel_pending_waiter(waiter);
-+
+                        }
                         install.demand_updates = state.publishable_demand_updates(std::mem::take(
                             &mut install.demand_updates,
                         ));
-+
+                    }
                     (cancelled, local_install)
                 })
         };
         let Some((cancelled, local_install)) = transition else {
             return false;
         };
-615
         if let Some(admission) = &cell.admission {
             for snapshot in cancelled
                 .demand_updates
                 .into_iter()
                 .chain(
                     local_install
                         .as_ref()
                         .into_iter()
                         .flat_map(|install| install.demand_updates.iter().cloned()),
-625
+                )
                 .flatten()
-627
+            {
                 OriginAdmission::publish_demand(admission, cell.id.partition(), snapshot);
-629
+            }
-630
+        }
-631
         // Ready results and any locally rejected event cross the lock boundary
         // before their fallback can re-enter the pool.
         drop(cancelled.returned_events);
         if let Some(install) = local_install {
             drop(install.returned_event);
             if let Some(waker) = install.waker {
                 waker.wake();
-639
+            }
-640
+        }
-641
         let availability = {
             let mut state = cell.state.lock();
             let local_h1_demand = state.waiters.can_accept_h1();
             state.reuse.clear_unused_turn(local_h1_demand);
             state.assert_consistent();
             state.take_h1_availability_update()
         };
         cell.publish_h1_availability(availability);
         Self::service_h2_waiters(cell);
         Self::service_peer_h2_waiters(cell);
         true
-653
+    }
-654
     /// Polls the next event for one acquisition waiter.
     ///
     /// # Panics
     ///
     /// Panics if `waiter` is unknown, was cancelled, or is polled again after
     /// its ready result was consumed.
     pub(super) fn poll_waiter(
         &self,
         waiter: WaiterId,
         cx: &mut Context<'_>,
     ) -> Poll<AcquisitionEvent> {
         let mut state = self.state.lock();
         state.waiters.poll_waiter(waiter, cx)
-668
+    }
-669
     /// Marks an establishment attempt as started before its first connector poll.
     pub(super) fn start_establishment(&self, waiter: WaiterId) -> bool {
         self.state.lock().waiters.start_establishment(waiter)
-673
+    }
-674
     /// Commits one terminal establishment result to its launching waiter.
     ///
     /// A returned H1 may already have served or cancelled the waiter. In that
     /// case the losing result leaves the cell lock and follows its ordinary
     /// sender-return or error-drop fallback.
     ///
     /// # Panics
     ///
     /// Panics if the launching waiter disappeared or left the launching state
     /// before its owned establishment completed.
     pub(super) fn complete_establishment(&self, waiter: WaiterId, result: AcquisitionResult) {
         let served_with_h1 = matches!(&result, AcquisitionResult::H1(_));
         let installation = {
             let mut state = self.state.lock();
             let installation = state.waiters.install_establishment_result(waiter, result);
             let served_with_h1 = served_with_h1
                 && !installation.returned_events.iter().any(|event| {
                     matches!(
                         event,
                         Some(AcquisitionEvent::Complete(AcquisitionResult::H1(_)))
-+
+                    )
                 });
             if installation.accepted {
                 if served_with_h1 {
                     state.reuse.consume_local_turn();
                 } else {
                     let local_h1_demand = state.waiters.can_accept_h1();
                     state.reuse.clear_unused_turn(local_h1_demand);
-703
+                }
                 state.assert_consistent();
-705
+            }
             installation
         };
         drop(installation.returned_events);
         if let Some(waker) = installation.waker {
             waker.wake();
-711
+        }
         if let Some(error) = installation.error {
             match error {
                 ResultInstallError::MissingWaiter => {
                     unreachable!("establishment result reported a missing waiter")
-716
+                }
                 ResultInstallError::UnexpectedState => {
                     panic!("establishment completed for a waiter that was not launching")
-719
+                }
-720
+            }
-721
+        }
         let availability = {
             let mut state = self.state.lock();
             state.assert_consistent();
             state.take_h1_availability_update()
         };
         self.publish_h1_availability(availability);
-728
+    }
-729
     #[cfg(test)]
     pub(super) fn take_ready_lease(cell: &Arc<Self>, waiter: WaiterId) -> Option<CapacityLease> {
         let permit = match cell.take_ready_event(waiter)? {
             AcquisitionEvent::Establish(permit) => permit,
             AcquisitionEvent::Complete(_) => {
                 panic!("capacity test received a terminal acquisition result")
-736
+            }
         };
         assert!(Self::cancel_waiter(cell, waiter));
         permit.into_lease()
-740
+    }
-741
     #[cfg(test)]
     fn take_ready_h1(&self, waiter: WaiterId) -> Option<H1Selection> {
         match self.take_ready_event(waiter)? {
             AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => Some(selection),
             AcquisitionEvent::Complete(AcquisitionResult::H2(_)) => {
                 panic!("HTTP/1 ownership test received an HTTP/2 activation")
-+
+            }
             AcquisitionEvent::Complete(AcquisitionResult::Failed(_)) => {
                 panic!("HTTP/1 ownership test received establishment failure")
-751
+            }
             AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
                 panic!("HTTP/1 ownership test received an internal reacquisition")
-+
+            }
             AcquisitionEvent::Establish(_) => {
                 panic!("HTTP/1 ownership test received establishment capacity")
-757
+            }
-758
+        }
-759
+    }
-760
     #[cfg(test)]
     fn take_ready_event(&self, waiter: WaiterId) -> Option<AcquisitionEvent> {
         let mut state = self.state.lock();
         match state
             .waiters
             .poll_waiter(waiter, &mut Context::from_waker(std::task::Waker::noop()))
-767
+        {
             Poll::Ready(event) => Some(event),
             Poll::Pending => None,
-770
+        }
-771
+    }
-772
     #[cfg(test)]
     pub(super) fn register_waiter_without_publish(
         &self,
         requirement: ProtocolRequirement,
     ) -> (WaiterId, DemandSnapshot) {
         let (waiter, snapshot) =
             self.state
                 .lock()
                 .waiters
                 .register_waiter(requirement, &self.eligibility_group, true);
-783
+        (
             waiter,
             snapshot.expect("first unpublished waiter did not create demand"),
-786
+        )
-787
+    }
-788
     #[cfg(test)]
     fn snapshot(&self) -> CellSnapshot {
         self.state.lock().waiters.snapshot()
-792
+    }
-793
+}
-794
 /// Terminal outcome for one reusable HTTP/1 sender return.
 enum H1ReturnTrace {
     /// The owning cell no longer accepts the sender.
     Rejected,
     /// Origin admission reserved the sender for peer demand.
     ReuseIntercepted,
     /// No compatible demand exists, so the sender became idle.
     Idle,
     /// A waiter in the owning cell accepted the sender.
     LocalDemand,
--
+}
--
 /// Emits a committed HTTP/1 return outcome outside the cell lock.
 // Keep field formatting out of a return transition that can synchronously
 // enter admission and sender fallbacks.
 #[inline(never)]
 fn trace_h1_return(cell: &OriginCell, connection_id: ConnectionId, outcome: H1ReturnTrace) {
     match outcome {
         H1ReturnTrace::Rejected => tracing::trace!(
             connection_id = %connection_id,
             connection_partition = ?cell.id.partition(),
             origin_scheme = %cell.id.origin().scheme(),
             origin_host = cell.id.origin().host(),
             origin_port = ?cell.id.origin().port(),
             "HTTP/1 return was rejected by its connection-owning cell"
         ),
         H1ReturnTrace::ReuseIntercepted => tracing::trace!(
             connection_id = %connection_id,
             connection_partition = ?cell.id.partition(),
             origin_scheme = %cell.id.origin().scheme(),
             origin_host = cell.id.origin().host(),
             origin_port = ?cell.id.origin().port(),
             "HTTP/1 return was intercepted by a cross-cell reuse operation"
         ),
         H1ReturnTrace::Idle => tracing::trace!(
             connection_id = %connection_id,
             connection_partition = ?cell.id.partition(),
             origin_scheme = %cell.id.origin().scheme(),
             origin_host = cell.id.origin().host(),
             origin_port = ?cell.id.origin().port(),
             "HTTP/1 connection returned to idle storage"
         ),
         H1ReturnTrace::LocalDemand => tracing::trace!(
             connection_id = %connection_id,
             request_partition = ?cell.id.partition(),
             connection_partition = ?cell.id.partition(),
             origin_scheme = %cell.id.origin().scheme(),
             origin_host = cell.id.origin().host(),
             origin_port = ?cell.id.origin().port(),
             "HTTP/1 return satisfied local demand"
         ),
--
+    }
--
+}
--
 impl std::fmt::Debug for OriginCell {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.debug_struct("OriginCell")
             .field("id", &self.id)
             .field("eligibility_group", &self.eligibility_group)
             .field("admission", &self.admission)
             .field("state", &self.state)
             .finish()
-803
+    }
-804
+}

↧ Expand context

-805
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use http_1x::uri::Scheme;
     use std::num::NonZeroUsize;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::sync::Arc as StdArc;
     use std::task::{Wake, Waker};
-814
     struct WakeCounter(AtomicUsize);
-816
     impl Wake for WakeCounter {
         fn wake(self: StdArc<Self>) {
             self.0.fetch_add(1, Ordering::Relaxed);
-820
+        }
-821
         fn wake_by_ref(self: &StdArc<Self>) {
             self.0.fetch_add(1, Ordering::Relaxed);
-824
+        }

@@ -1276,980 +1812,1527 @@

↥ Expand context

         let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
         assert!(!selection.is_reused());
-982
         selection.into_exchange().offer_for_reuse();
         assert_eq!((1, 1), cell.h1_counts());
-985
+    }
-986
     #[test]
     fn incomplete_h1_return_closes_and_releases_capacity() {
         let (admission, cell) = bounded_cell();
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         let selection =
             OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
-994
         drop(selection.into_exchange());
-996
         assert_eq!((0, 0), cell.h1_counts());
         assert_eq!(1, admission.available_capacity_for_test());
         assert_eq!(

             Some(CloseReason::IncompleteH1Exchange),
             connection.snapshot().close_reason
         );
-1003
+    }
-1004
     #[test]
     fn returned_h1_satisfies_an_unbounded_waiter() {
         let cell = unbounded_cell();
         let (connection, _physical) = unbounded_connection(1);
         let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-1011
         drop(selection);
-1013
         let selected = cell
             .take_ready_h1(waiter)
             .expect("returned sender did not satisfy the waiter");
         assert_eq!(11, selected.test_sender_id());
         assert!(selected.is_reused());
         assert_eq!(0, cell.snapshot().retained);
         drop(selected);
         assert_eq!((1, 1), cell.h1_counts());
-1022
+    }
-1023
     #[test]
     fn returned_h1_satisfies_bounded_demand_without_releasing_capacity() {
         let (admission, cell) = bounded_cell();
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         assert_eq!(1, admission.ordered_demand_count_for_test());
-1032
         drop(selection);
-1034
         assert_eq!(0, admission.ordered_demand_count_for_test());
         assert_eq!(0, admission.available_capacity_for_test());
         let selected = cell
             .take_ready_h1(waiter)
             .expect("returned sender did not satisfy bounded demand");
         drop(selected);
         assert_eq!((1, 1), cell.h1_counts());
-1042
+    }
-1043
     #[test]
     fn eligible_peer_borrows_an_idle_h1_without_moving_capacity() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         OriginCell::install_idle_h1(&connection_cell, connection, H1Sender::test(11));
-1051
         let waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter =
             OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
         let borrowed = requesting_cell
             .take_ready_h1(waiter)
             .expect("eligible peer did not receive the idle HTTP/1 sender");
-1057
         assert_eq!(11, borrowed.test_sender_id());
         assert_eq!(0, admission.available_capacity_for_test());
         assert_eq!((1, 0), connection_cell.h1_counts());
         drop(borrowed);
         assert_eq!((1, 1), connection_cell.h1_counts());
-1063
+    }
-1064
     #[test]
     fn ineligible_peer_reclaims_h1_capacity_without_moving_the_sender() {
         let (admission, connection_cell, requesting_cell) = bounded_peer_cells(
 ,
             EligibilityGroup::Partition(PartitionId::from_index(1)),
             EligibilityGroup::Partition(PartitionId::from_index(2)),
         );
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         OriginCell::install_idle_h1(&connection_cell, connection.clone(), H1Sender::test(11));
-1075
         let waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter =
             OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
         let replacement = OriginCell::take_ready_lease(&requesting_cell, waiter)
             .expect("ineligible peer did not receive reclaimed capacity");
-1080
         assert_eq!(
             Some(CloseReason::Reclaimed),
             connection.snapshot().close_reason
         );
         assert_eq!((0, 0), connection_cell.h1_counts());
         assert_eq!(0, admission.available_capacity_for_test());
         drop(replacement);
         assert_eq!(1, admission.available_capacity_for_test());
-1089
+    }
-1090
     #[test]
     fn installed_peer_reuse_intercepts_the_next_active_h1_return() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         let selected =
             OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
         let waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter =
             OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
-1101
         drop(selected);
-1103
         let borrowed = requesting_cell
             .take_ready_h1(waiter)
             .expect("installed reuse operation did not intercept the active return");
         assert_eq!(11, borrowed.test_sender_id());
         drop(borrowed);
         assert_eq!((1, 1), connection_cell.h1_counts());
-1110
+    }
-1111
     #[test]
     fn cancelling_requesting_cell_releases_connection_reservation() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         let selected =
             OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
         let waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter =
             OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
-1122
         assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
         assert!(
             connection_cell.state.lock().reuse.is_available(),
             "requesting cell cancellation did not reconcile the installed connection-owning cell reuse operation"
         );
         drop(selected);
-1129
         assert_eq!((1, 1), connection_cell.h1_counts());
         assert_eq!(0, admission.ordered_demand_count_for_test());
-1132
+    }
-1133
     #[test]
     fn cross_cell_borrow_owes_one_usable_turn_to_the_owning_cell() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         let selected =
             OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
-1142
         let requesting_waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
         let local_waiter = connection_cell.register_waiter(ProtocolRequirement::H1Compatible);
         let requesting_waiter =
             OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
         let local_waiter =
             OriginCell::register_waiter(&connection_cell, ProtocolRequirement::H1Compatible);
         drop(selected);
-1148
         let borrowed = requesting_cell
             .take_ready_h1(requesting_waiter)
             .expect("older peer demand did not receive the intercepted sender");
         assert!(connection_cell.state.lock().reuse.local_turn_owed());
         drop(borrowed);
-1154
         let local = connection_cell
             .take_ready_h1(local_waiter)
             .expect("connection-owning cell local demand did not receive its owed turn");
         assert!(!connection_cell.state.lock().reuse.local_turn_owed());
         drop(local);
         assert_eq!((1, 1), connection_cell.h1_counts());
-1161
+    }
-1162
     #[test]
     fn cancelling_ready_h1_returns_it_after_unlock() {
         let cell = unbounded_cell();
         let (connection, _physical) = unbounded_connection(1);
         let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         drop(selection);
-1170
         assert!(OriginCell::cancel_waiter(&cell, waiter));
         assert_eq!((1, 1), cell.h1_counts());
         assert_eq!(
 ,
             OriginCell::select_h1(&cell)
                 .expect("cancelled result did not return its sender")
                 .test_sender_id()
         );
-1179
+    }
-1180
     #[test]
     fn returned_h1_and_establishment_complete_one_acquisition_attempt() {
         let cell = unbounded_cell();
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not start establishment")
         else {
             panic!("unbounded miss completed before establishment started");
         };
         assert!(cell.start_establishment(waiter));
         assert!(permit.into_lease().is_none());
-1193
         let (returning_connection, _returning_physical) = unbounded_connection(1);
         let returning =
             OriginCell::install_selected_h1(&cell, returning_connection, H1Sender::test(11));
         let (fresh_connection, _fresh_physical) = unbounded_connection(2);
         let fresh = OriginCell::install_selected_h1(&cell, fresh_connection, H1Sender::test(22));
-1199
         drop(returning);
         cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
-1202
         assert_eq!((2, 1), cell.h1_counts());
         let winner = cell
             .take_ready_h1(waiter)
             .expect("acquisition attempt had no terminal H1");
         assert_eq!(11, winner.test_sender_id());
         drop(winner);
         assert_eq!((2, 2), cell.h1_counts());
-1210
+    }
-1211
     #[test]
     fn bounded_attempt_remains_eligible_for_a_returned_h1() {
         let (admission, cell) = bounded_cell_with_limit(2);
         let installed_lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), installed_lease);
         let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
-1218
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("bounded miss received no capacity")
         else {
             panic!("bounded miss completed before establishment started");
         };
         let attempt_lease = permit
             .into_lease()
             .expect("bounded establishment carried no capacity");
         assert_eq!(0, admission.available_capacity_for_test());
-1230
         drop(returning);
         let selected = cell
             .take_ready_h1(waiter)
             .expect("returned H1 did not beat the pending establishment");
         drop(attempt_lease);
         assert_eq!(1, admission.available_capacity_for_test());
         drop(selected);
         assert_eq!((1, 1), cell.h1_counts());
-1239
+    }
-1240
     #[test]
     fn older_launching_waiter_precedes_younger_capacity_waiter_for_returned_h1() {
         let (admission, cell) = bounded_cell_with_limit(2);
         let installed_lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), installed_lease);
         let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
-1247
         let older = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let older = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(older)
             .expect("older waiter received no establishment capacity")
         else {
             panic!("older waiter completed before establishment started");
         };
         let attempt_lease = permit
             .into_lease()
             .expect("bounded establishment carried no capacity");
-1258
         let younger = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let younger = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         assert!(cell.take_ready_event(younger).is_none());
-1261
         drop(returning);
-1263
         let selected = cell
             .take_ready_h1(older)
             .expect("younger capacity waiter overtook the older launching waiter");
         assert_eq!(11, selected.test_sender_id());
         assert!(cell.take_ready_event(younger).is_none());
-1269
         assert!(OriginCell::cancel_waiter(&cell, younger));
         drop(attempt_lease);
         drop(selected);
         assert_eq!((1, 1), cell.h1_counts());
         assert_eq!(1, admission.available_capacity_for_test());
-1275
+    }
-1276
     #[test]
     fn establishment_failure_is_a_terminal_acquisition_result() {
         let cell = unbounded_cell();
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not start establishment")
         else {
             panic!("unbounded miss completed before establishment started");
         };
         assert!(cell.start_establishment(waiter));
         drop(permit);
-1289
         cell.complete_establishment(
             waiter,
             AcquisitionResult::Failed(ConnectorError::io(
                 std::io::Error::other("synthetic establishment failure").into(),
             )),
         );
-1296
         let AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) = cell
             .take_ready_event(waiter)
             .expect("establishment failure was not delivered")
         else {
             panic!("establishment failure produced the wrong acquisition event");
         };
         assert!(error.is_io());
         assert_eq!(0, cell.snapshot().retained);
-1305
+    }
-1306
     #[test]
     fn establishment_completion_clears_the_served_waiters_local_turn() {
         let cell = unbounded_cell();
         let reuse_id = ReuseId::for_test(1);
-1311
+        {
             let mut state = cell.state.lock();
             assert!(state.reuse.install_resolving(reuse_id));
             assert!(state.reuse.complete_transfer(reuse_id, true));
-1315
+        }
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not prepare establishment")
         else {
             panic!("unbounded miss completed before establishment was prepared");
         };
         assert!(cell.start_establishment(waiter));
         drop(permit);
         let (connection, _physical) = unbounded_connection(1);
         let fresh = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
-1327
         cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
-1329
         assert!(!cell.state.lock().reuse.local_turn_owed());
         drop(
             cell.take_ready_h1(waiter)
                 .expect("successful establishment result was not retained"),
         );
-1335
+    }
-1336
     #[test]
     fn failed_establishment_preserves_a_turn_for_compatible_successor() {
         let cell = unbounded_cell();
         let reuse_id = ReuseId::for_test(1);
-1341
+        {
             let mut state = cell.state.lock();
             assert!(state.reuse.install_resolving(reuse_id));
             assert!(state.reuse.complete_transfer(reuse_id, true));
-1345
+        }
         let failed = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let failed = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(failed)
             .expect("unbounded miss did not prepare establishment")
         else {
             panic!("unbounded miss completed before establishment was prepared");
         };
         assert!(cell.start_establishment(failed));
         drop(permit);
         let successor = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let successor = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-1356
         cell.complete_establishment(
             failed,
             AcquisitionResult::Failed(ConnectorError::io(
                 std::io::Error::other("synthetic establishment failure").into(),
             )),
         );
-1363
         assert!(cell.state.lock().reuse.local_turn_owed());
         assert!(matches!(
             cell.take_ready_event(failed),
             Some(AcquisitionEvent::Complete(AcquisitionResult::Failed(_)))
         ));
         assert!(OriginCell::cancel_waiter(&cell, successor));
         assert!(!cell.state.lock().reuse.local_turn_owed());
-1371
+    }
-1372
     #[test]
     fn failed_establishment_clears_a_turn_after_local_demand_drains() {
         let cell = unbounded_cell();
         let reuse_id = ReuseId::for_test(1);
-1377
+        {
             let mut state = cell.state.lock();
             assert!(state.reuse.install_resolving(reuse_id));
             assert!(state.reuse.complete_transfer(reuse_id, true));
-1381
+        }
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not prepare establishment")
         else {
             panic!("unbounded miss completed before establishment was prepared");
         };
         assert!(cell.start_establishment(waiter));
         drop(permit);
-1391
         cell.complete_establishment(
             waiter,
             AcquisitionResult::Failed(ConnectorError::io(
                 std::io::Error::other("synthetic establishment failure").into(),
             )),
         );
-1398
         assert!(!cell.state.lock().reuse.local_turn_owed());
         assert!(matches!(
             cell.take_ready_event(waiter),
             Some(AcquisitionEvent::Complete(AcquisitionResult::Failed(_)))
         ));
-1404
+    }
-1405
     #[test]
     fn completion_after_waiter_cancellation_returns_the_new_h1() {
         let cell = unbounded_cell();
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not start establishment")
         else {
             panic!("unbounded miss completed before establishment started");
         };
         assert!(cell.start_establishment(waiter));
         assert!(OriginCell::cancel_waiter(&cell, waiter));
         drop(permit);
-1419
         let (connection, _physical) = unbounded_connection(1);
         let fresh = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
         cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
-1423
         assert_eq!((1, 1), cell.h1_counts());
         assert_eq!(
 ,
             OriginCell::select_h1(&cell)
                 .expect("cancelled acquisition lost its completed H1")
                 .test_sender_id()
         );
-1431
+    }
-1432
     #[test]
     fn returned_h1_prevents_the_first_establishment_poll() {
         let cell = unbounded_cell();
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not prepare establishment")
         else {
             panic!("unbounded miss completed before establishment was prepared");
         };
         let (connection, _physical) = unbounded_connection(1);
         let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
-1445
         drop(returning);
         assert!(!cell.start_establishment(waiter));
         drop(permit);
         let selected = cell
             .take_ready_h1(waiter)
             .expect("returned H1 did not win before establishment started");
         assert_eq!(11, selected.test_sender_id());
-1453
+    }
-1454
     #[test]
     fn h2_required_head_does_not_take_h1() {
         let cell = unbounded_cell();
         let (connection, _physical) = unbounded_connection(1);
         let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
         let waiter = cell.register_waiter(ProtocolRequirement::H2Required);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
-1461
         drop(selection);
-1463
         assert_eq!((1, 1), cell.h1_counts());
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(waiter)
             .expect("unbounded miss did not start establishment")
         else {
             panic!("H2-required waiter accepted an HTTP/1 selection");
         };
         assert!(OriginCell::cancel_waiter(&cell, waiter));
         drop(permit);
-1473
+    }
-1474
     #[test]
     fn cancelling_h2_head_serves_compatible_successor_locally() {
     fn cancelling_h2_head_refunnels_reclaimed_capacity_to_successor() {
         let (admission, cell) = bounded_cell();
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         OriginCell::install_idle_h1(&cell, connection, H1Sender::test(11));
         let h2_head = cell.register_waiter(ProtocolRequirement::H2Required);
         let h1_successor = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let h2_head = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
         let h1_successor = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-1483
         assert!(cell.take_ready_event(h2_head).is_none());
         let AcquisitionEvent::Establish(permit) = cell
             .take_ready_event(h2_head)
             .expect("HTTP/2 demand did not receive reclaimed local capacity")
         else {
             panic!("HTTP/2 demand received a protocol handle instead of capacity");
         };
         assert!(OriginCell::cancel_waiter(&cell, h2_head));
         drop(permit);
-1492
         let selected = cell
             .take_ready_h1(h1_successor)
             .expect("local idle H1 did not serve the compatible successor");
         assert_eq!(11, selected.test_sender_id());
         let successor = OriginCell::take_ready_lease(&cell, h1_successor)
             .expect("reclaimed capacity did not reach the compatible successor");
         assert_eq!((0, 0), cell.h1_counts());
         assert!(cell.state.lock().reuse.is_available());
         drop(selected);
         drop(successor);
-1498
+    }
-1499
     #[test]
     fn close_marks_selected_record_until_sender_returns() {
         let cell = unbounded_cell();
         let (connection, _physical) = unbounded_connection(1);
         let selection =
             OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
         let close = H1CloseHandle::new(&cell, selection.connection());
-1507

↧ Expand context

         assert!(close.close(CloseReason::Poisoned));
         assert!(OriginCell::select_h1(&cell).is_none());
         assert_eq!((1, 0), cell.h1_counts());
-1511
         drop(selection);
         assert_eq!((0, 0), cell.h1_counts());
         assert_eq!(
             Some(CloseReason::Poisoned),
             connection.snapshot().close_reason
         );
-1518
+    }
-1519
     #[test]
     fn confirmed_upgrade_refines_an_earlier_driver_close() {
         let (admission, cell) = bounded_cell();
         let lease = OriginAdmission::lease_for_test(&admission);
         let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
         let selection =
             OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
         let driver = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));

@@ -1848,1563 +1984,1699 @@

↥ Expand context

             OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
         let driver = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));
         let runtime = tokio::runtime::Builder::new_multi_thread()
             .worker_threads(1)
             .enable_all()
             .build()
             .unwrap();
         let (started, observed) = std::sync::mpsc::sync_channel(0);
-1571
         runtime.spawn(async move {
             let _driver = driver;
             started.send(()).unwrap();
             std::future::pending::<()>().await;
         });
         observed
             .recv_timeout(std::time::Duration::from_secs(1))
             .expect("driver task did not start");
         runtime.shutdown_timeout(std::time::Duration::from_secs(1));
-1581
         assert_eq!(

             Some(CloseReason::OwnerRuntimeShutdown),
             connection.snapshot().close_reason
         );
         drop(selection);
         assert_eq!((0, 0), cell.h1_counts());
-1588
+    }
-1589
     #[test]
     fn three_waiters_receive_capacity_in_fifo_order() {
         let (_admission, cell) = bounded_cell();
         let first = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let second = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let third = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-1596
         let first_lease = OriginCell::take_ready_lease(&cell, first).unwrap();
         assert!(OriginCell::take_ready_lease(&cell, second).is_none());
         assert!(OriginCell::take_ready_lease(&cell, third).is_none());
-1600
         drop(first_lease);
         let second_lease = OriginCell::take_ready_lease(&cell, second).unwrap();
         assert!(OriginCell::take_ready_lease(&cell, third).is_none());
-1604
         drop(second_lease);
         let third_lease = OriginCell::take_ready_lease(&cell, third).unwrap();
         drop(third_lease);
         assert_eq!(0, cell.snapshot().retained);
-1609
+    }
-1610
     #[test]
     fn cancelling_middle_and_tail_waiters_repairs_fifo_links() {
         let (_admission, cell, held) = saturated_bounded_cell();
         let first = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let second = cell.register_waiter(ProtocolRequirement::H2Required);
         let third = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let fourth = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
         let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let fourth = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let demand = cell.snapshot().demand;
-1619
         assert!(OriginCell::cancel_waiter(&cell, second));
         assert!(OriginCell::cancel_waiter(&cell, fourth));
         assert_eq!(demand, cell.snapshot().demand);
         assert_eq!(2, cell.snapshot().waiting);
         let fifth = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let fifth = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         assert_eq!(3, cell.snapshot().waiting);
-1626
         drop(held);
         let first_lease = OriginCell::take_ready_lease(&cell, first).unwrap();
         assert!(OriginCell::take_ready_lease(&cell, third).is_none());
         assert!(OriginCell::take_ready_lease(&cell, fifth).is_none());
         drop(first_lease);
-1632
         let third_lease = OriginCell::take_ready_lease(&cell, third).unwrap();
         assert!(OriginCell::take_ready_lease(&cell, fifth).is_none());
         drop(third_lease);
-1636
         drop(OriginCell::take_ready_lease(&cell, fifth).unwrap());
         assert_eq!(0, cell.snapshot().retained);
-1639
+    }
-1640
     #[test]
     fn cancelling_head_waiter_preserves_remaining_fifo_order() {
         let (_admission, cell, held) = saturated_bounded_cell();
         let first = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let second = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let third = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let fourth = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let fourth = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-1648
         assert!(OriginCell::cancel_waiter(&cell, first));
         drop(held);
-1651
         let second_lease = OriginCell::take_ready_lease(&cell, second).unwrap();
         assert!(OriginCell::take_ready_lease(&cell, third).is_none());
         assert!(OriginCell::take_ready_lease(&cell, fourth).is_none());
         drop(second_lease);
-1656
         let third_lease = OriginCell::take_ready_lease(&cell, third).unwrap();
         assert!(OriginCell::take_ready_lease(&cell, fourth).is_none());
         drop(third_lease);
-1660
         drop(OriginCell::take_ready_lease(&cell, fourth).unwrap());
         assert_eq!(0, cell.snapshot().retained);
-1663
+    }
-1664
     #[test]
     fn cancelling_ready_waiter_refunnels_capacity() {
         let (_admission, cell) = bounded_cell();
         let first = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let second = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-1670
         assert!(OriginCell::cancel_waiter(&cell, first));
         let lease = OriginCell::take_ready_lease(&cell, second).unwrap();
         drop(lease);
-1674
+    }
-1675
     #[test]
     fn cancelling_during_delivery_refunnels_capacity_after_unlock() {
         let (admission, cell) = bounded_cell();
         let (waiter, demand) =

↧ Expand context

             cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
         let mut delivery =
             OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
                 .expect("published demand did not reserve capacity");
         assert!(delivery.materialize_for_test());
         let reservation = {
             let mut state = cell.state.lock();
             state
                 .waiters
                 .reserve_delivery_waiter(delivery.demand(), &cell.eligibility_group)
         };
         let DeliveryReservation::Reserved {
             waiter: reserved,
             successor,
         } = reservation
         else {
             panic!("current delivery was rejected");
         };
         assert_eq!(waiter, reserved);
         let (event, acknowledgement) = delivery.commit(successor);

@@ -2092,1807 +2152,1867 @@

↥ Expand context

         drop(returning);
         assert!(OriginCell::cancel_waiter(&cell, waiter));
         let (event, acknowledgement) = delivery.commit(successor);
         let AcquisitionEvent::Establish(permit) = event else {
             panic!("capacity delivery materialized a non-capacity event");
         };
         let installation = {
             let mut state = cell.state.lock();
             state.waiters.install_capacity(waiter, permit)
         };
         assert!(installation.returned_events.iter().all(Option::is_some));
         drop(installation.returned_events);
         assert!(acknowledgement.accept().is_none());
-1820
         assert_eq!(1, admission.available_capacity_for_test());
         assert_eq!(0, cell.snapshot().retained);
         assert_eq!((1, 1), cell.h1_counts());
         assert_eq!(
 ,
             OriginCell::select_h1(&cell)

                 .expect("cancelled crossing lost its H1 sender")
                 .test_sender_id()
         );
-1830
+    }
-1831
     #[test]
     fn stale_delivery_is_rejected_after_head_cancellation() {
         let (admission, cell) = bounded_cell();
         let (first, first_demand) =
             cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
         let second = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let stale = OriginAdmission::publish_without_driving(
             &admission,
             cell.id().partition(),
             first_demand,
-1842
+        )
         .expect("first demand did not reserve capacity");
-1844
         assert!(OriginCell::cancel_waiter(&cell, first));
         OriginAdmission::drive(Some(AdmissionAction::Delivery(stale)));
-1847

↧ Expand context

         let lease = OriginCell::take_ready_lease(&cell, second)
             .expect("replacement demand did not receive refunnelled capacity");
         drop(lease);
-1851
+    }
-1852
     #[test]
     fn stale_delivery_returns_its_successor_to_the_driver() {
         let admission = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let stale = OriginAdmission::register_cell(
             &admission,
             Arc::new(OriginCell::new(
                 PartitionId::from_index(1),
                 OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
                 EligibilityGroup::Pool,
                 Some(admission.clone()),
                 None,
             )),
         );
         let target = OriginAdmission::register_cell(
             &admission,

@@ -2172,1887 +2276,2289 @@

↥ Expand context

         OriginAdmission::publish_demand(&admission, target.id().partition(), target_demand);
-1888
         // Model cancellation winning in the cell before its retirement
         // publication reaches admission.
         let cancelled = stale
             .state
             .lock()
             .waiters
             .cancel_waiter(stale_waiter, &stale.eligibility_group)
             .expect("stale waiter was not cancelled");
         drop(cancelled.returned_events);
-1898
         assert!(delivery.materialize_for_test());
         let next = OriginCell::receive_delivery(&stale, delivery)
             .expect("stale rejection did not return the successor delivery");
         assert!(
             OriginCell::take_ready_lease(&target, target_waiter).is_none(),
             "stale rejection recursively drove its successor"
         );
-1906

         OriginAdmission::drive(Some(next));
         drop(
             OriginCell::take_ready_lease(&target, target_waiter)
                 .expect("existing driver did not deliver the successor"),
         );
-1912
+    }
-1913
     #[test]
     fn cancelling_the_only_waiter_retires_admission_demand() {
         let (admission, cell, held) = saturated_bounded_cell();
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         assert_eq!(1, admission.ordered_demand_count_for_test());
-1919
         assert!(OriginCell::cancel_waiter(&cell, waiter));
         assert_eq!(0, admission.ordered_demand_count_for_test());
         drop(held);
-1923
+    }
-1924
     #[test]
     fn unknown_waiter_cancellation_is_a_noop() {
         let (_admission, cell) = bounded_cell();
         assert!(!OriginCell::cancel_waiter(&cell, WaiterId(99)));
-1929
+    }
-1930
     #[test]
     fn unbounded_waiter_registers_and_cancels_locally() {
         let cell = Arc::new(OriginCell::new(
             PartitionId::from_index(1),
             OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
             EligibilityGroup::Pool,
             None,
             None,
         ));
-1940
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         assert_eq!(0, cell.snapshot().waiting);
         assert_eq!(1, cell.snapshot().retained);
         assert!(OriginCell::cancel_waiter(&cell, waiter));
         assert_eq!(0, cell.snapshot().retained);
-1946
+    }
-1947
     #[test]
     fn delivered_waiter_is_woken_once() {
         let (_admission, cell, held) = saturated_bounded_cell();
         let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
         let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
         let counter = StdArc::new(WakeCounter(AtomicUsize::new(0)));
         let waker = Waker::from(counter.clone());
         let mut context = Context::from_waker(&waker);
         assert!(cell.poll_waiter(waiter, &mut context).is_pending());
-1956
         drop(held);
         assert_eq!(1, counter.0.load(Ordering::Relaxed));
         drop(OriginCell::take_ready_lease(&cell, waiter).expect("woken waiter had no capacity"));
         assert_eq!(1, counter.0.load(Ordering::Relaxed));
-1961
+    }
-+
     fn install_bounded_h2(
         admission: &Arc<OriginAdmission>,
         cell: &Arc<OriginCell>,
         connection_id: u64,
     ) -> (
         h2::H2GenerationId,
         Arc<ConnectionState>,
         super::super::connection::PhysicalConnectionGuard,
     ) {
         let lease = OriginAdmission::lease_for_test(admission);
         let (connection, physical) = ConnectionState::bounded(
             ConnectionInfo::for_test(ConnectionId::new(connection_id), cell.id().partition()),
             lease,
         );
         let generation =
             OriginCell::install_h2_for_test(cell, connection.clone(), connection_id, None);
         (generation, connection, physical)
-+
+    }
-+
     fn take_ready_h2(cell: &OriginCell, waiter: WaiterId) -> h2::H2Activation {
         match cell
             .take_ready_event(waiter)
             .expect("HTTP/2 waiter did not receive an activation")
-+
+        {
             AcquisitionEvent::Complete(AcquisitionResult::H2(activation)) => activation,
             AcquisitionEvent::Establish(_) => {
                 panic!("HTTP/2 waiter received establishment capacity")
-+
+            }
             AcquisitionEvent::Complete(AcquisitionResult::H1(_)) => {
                 panic!("HTTP/2 waiter received an HTTP/1 sender")
-+
+            }
             AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
                 panic!("HTTP/2 waiter was returned for reacquisition")
-+
+            }
             AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
                 panic!("HTTP/2 waiter received an establishment failure: {error}")
-+
+            }
-+
+        }
-+
+    }
-+
     #[test]
     fn eligible_peer_h2_generation_satisfies_bounded_demand() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (generation, connection, _physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
-+
         let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
         let activation = take_ready_h2(&requesting_cell, waiter);
-+
         assert_eq!(generation, activation.generation());
         assert!(Arc::ptr_eq(&connection, activation.connection()));
         assert_eq!(0, admission.available_capacity_for_test());
         assert_eq!(Some(generation), connection_cell.accepting_h2_generation());
         drop(activation);
         assert_eq!(0, requesting_cell.snapshot().retained);
         assert!(OriginCell::close_h2(
             &connection_cell,
             generation,
             CloseReason::PoolDropped,
         ));
         assert_eq!(1, admission.available_capacity_for_test());
-+
+    }
-+
     #[test]
     fn peer_h2_publication_respects_eligibility_group() {
         let groups = [
-+
+            (
                 EligibilityGroup::Partition(PartitionId::from_index(1)),
                 EligibilityGroup::Partition(PartitionId::from_index(2)),
             ),
-+
+            (
                 EligibilityGroup::NetworkInterface(Some(Arc::from("eth0"))),
                 EligibilityGroup::NetworkInterface(Some(Arc::from("eth1"))),
             ),
         ];
         for (connection_group, requesting_group) in groups {
             let (admission, connection_cell, requesting_cell) =
                 bounded_peer_cells(1, connection_group, requesting_group);
             let (generation, _connection, _physical) =
                 install_bounded_h2(&admission, &connection_cell, 1);
             let waiter =
                 OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
-+
             assert!(requesting_cell.take_ready_event(waiter).is_none());
             assert_eq!(1, admission.ordered_demand_count_for_test());
             assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
             assert!(OriginCell::close_h2(
                 &connection_cell,
                 generation,
                 CloseReason::PoolDropped,
             ));
             assert_eq!(1, admission.available_capacity_for_test());
-+
+        }
-+
+    }
-+
     #[test]
     fn dropped_publication_guard_retries_the_same_demand() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (generation, _connection, _physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let (waiter, demand) =
             requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
         let action = OriginAdmission::publish_action_without_driving(
             &admission,
             requesting_cell.id().partition(),
             demand,
-+
+        )
         .expect("peer demand did not prepare publication");
-+
         drop(action);
-+
         drop(take_ready_h2(&requesting_cell, waiter));
         assert_eq!(0, admission.ordered_demand_count_for_test());
         assert!(OriginCell::close_h2(
             &connection_cell,
             generation,
             CloseReason::PoolDropped,
         ));
-+
+    }
-+
     #[test]
     fn requesting_cell_cancellation_closes_an_in_flight_publication_fence() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (generation, _connection, _physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let (waiter, demand) =
             requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
         let action = OriginAdmission::publish_action_without_driving(
             &admission,
             requesting_cell.id().partition(),
             demand,
-+
+        )
         .expect("peer demand did not prepare publication");
-+
         assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
         OriginAdmission::drive(Some(action));
-+
         assert_eq!(0, requesting_cell.snapshot().retained);
         assert_eq!(0, admission.ordered_demand_count_for_test());
         assert_eq!(0, admission.available_capacity_for_test());
         assert_eq!(Some(generation), connection_cell.accepting_h2_generation());
         assert!(OriginCell::close_h2(
             &connection_cell,
             generation,
             CloseReason::PoolDropped,
         ));
-+
+    }
-+
     #[test]
     fn stale_publication_retries_against_a_replacement_generation() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (first, _first_connection, _first_physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let (waiter, demand) =
             requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
         let action = OriginAdmission::publish_action_without_driving(
             &admission,
             requesting_cell.id().partition(),
             demand,
-+
+        )
         .expect("peer demand did not prepare publication");
-+
         assert!(OriginCell::close_h2(
             &connection_cell,
             first,
             CloseReason::ProtocolClosed
         ));
         let (second, _second_connection, _second_physical) =
             install_bounded_h2(&admission, &connection_cell, 2);
         OriginAdmission::drive(Some(action));
-+
         let activation = take_ready_h2(&requesting_cell, waiter);
         assert_eq!(second, activation.generation());
         drop(activation);
         assert!(OriginCell::close_h2(
             &connection_cell,
             second,
             CloseReason::PoolDropped
         ));
-+
+    }
-+
     #[test]
     fn stale_peer_route_republishes_demand_to_a_replacement_generation() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (first, _first_connection, _first_physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let (waiter, demand) =
             requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
         assert!(OriginCell::install_h2_route(
             &requesting_cell,
             h2::H2Route::new(&connection_cell, first),
             &EligibilityGroup::Pool,
             demand.id_for_test(),
         ));
-+
         assert!(OriginCell::close_h2(
             &connection_cell,
             first,
             CloseReason::ProtocolClosed
         ));
         let (second, _second_connection, _second_physical) =
             install_bounded_h2(&admission, &connection_cell, 2);
         OriginCell::service_peer_h2_waiters(&requesting_cell);
         let activation = take_ready_h2(&requesting_cell, waiter);
         assert_eq!(second, activation.generation());
         drop(activation);
         assert!(OriginCell::close_h2(
             &connection_cell,
             second,
             CloseReason::PoolDropped
         ));
-+
+    }
-+
     #[test]
     fn direct_selection_removes_a_stale_peer_route() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (generation, _connection, _physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
         drop(take_ready_h2(&requesting_cell, waiter));
-+
         assert!(OriginCell::close_h2(
             &connection_cell,
             generation,
             CloseReason::ProtocolClosed,
         ));
         assert!(OriginCell::select_h2(&requesting_cell).is_none());
         assert!(
             !requesting_cell.state.lock().h2.has_visible_h2(),
             "direct selection retained a stale peer route"
         );
-+
+    }
-+
     #[test]
     fn open_peer_route_allows_concurrent_prospective_activations() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (generation, _connection, _physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
         drop(take_ready_h2(&requesting_cell, waiter));
-+
         let first = OriginCell::select_h2(&requesting_cell)
             .expect("first direct peer activation was not selected");
         let second = OriginCell::select_h2(&requesting_cell)
             .expect("open peer route serialized the second activation");
         assert_eq!(generation, first.generation());
         assert_eq!(generation, second.generation());
         assert_eq!(Some((2, 0)), connection_cell.h2_request_counts(generation));
         drop(first);
         drop(second);
         assert!(OriginCell::close_h2(
             &connection_cell,
             generation,
             CloseReason::PoolDropped
         ));
-+
+    }
-+
     #[test]
     fn peer_generation_gate_preserves_prepublication_order() {
         let (admission, connection_cell, requesting_cell) =
             bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
         let (generation, _connection, _physical) =
             install_bounded_h2(&admission, &connection_cell, 1);
         let (first, demand) =
             requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
         let second = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
         let action = OriginAdmission::publish_action_without_driving(
             &admission,
             requesting_cell.id().partition(),
             demand,
-+
+        )
         .expect("peer demand did not prepare publication");
         OriginAdmission::drive(Some(action));
         let third = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
-+
         let first_activation = take_ready_h2(&requesting_cell, first);
         assert!(requesting_cell.take_ready_event(second).is_none());
         assert!(requesting_cell.take_ready_event(third).is_none());
         drop(first_activation);
-+
         let second_activation = take_ready_h2(&requesting_cell, second);
         assert!(requesting_cell.take_ready_event(third).is_none());
         drop(second_activation);
         drop(take_ready_h2(&requesting_cell, third));
         assert!(OriginCell::close_h2(
             &connection_cell,
             generation,
             CloseReason::PoolDropped,
         ));
-+
+    }
-2260
+}
-2261
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use http_1x::uri::Scheme;
     use std::num::NonZeroUsize;
-2267
     fn bounded_cell() -> (Arc<OriginAdmission>, Arc<OriginCell>) {
         bounded_cell_with_limit(1)

↧ Expand context

-2270
+    }
-2271
     fn bounded_cell_with_limit(limit: usize) -> (Arc<OriginAdmission>, Arc<OriginCell>) {
         let admission = OriginAdmission::for_test(NonZeroUsize::new(limit).unwrap());
         let candidate = Arc::new(OriginCell::new(
             PartitionId::from_index(1),
             OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
             EligibilityGroup::Pool,
             Some(admission.clone()),
             None,
         ));
         let cell = OriginAdmission::register_cell(&admission, candidate);
         (admission, cell)
-2283
+    }
-2284
     fn bounded_peer_cells(
         owning_group: EligibilityGroup,
         requesting_group: EligibilityGroup,
     ) -> (Arc<OriginAdmission>, Arc<OriginCell>, Arc<OriginCell>) {
         bounded_peer_cells_with_limit(1, owning_group, requesting_group)

@@ -2346,2359 +2471,2484 @@

↥ Expand context

-2359
     #[test]
     fn h1_return_linearizes_against_close() {
         loom::model(|| {
             let cell = unbounded_cell();
             let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
             let selection =
                 OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
             let exchange = selection.into_exchange();
             let close = H1CloseHandle::new(&cell, &connection);
-2369
             let returning = loom::thread::spawn(move || exchange.offer_for_reuse());
             let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
             returning.join().unwrap();
             assert!(closing.join().unwrap());
-2374
             assert_eq!((0, 0), cell.h1_counts());
             assert_eq!(
                 Some(CloseReason::Poisoned),
                 connection.snapshot().close_reason

             );
         });
-2381
+    }
-2382
     #[test]
     fn h1_delivery_and_waiter_cancellation_preserve_the_sender() {
         loom::model(|| {
             let cell = unbounded_cell();
             let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
             let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
             let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
             let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
-2390
             let returning = loom::thread::spawn(move || drop(selection));
             let cancel_cell = cell.clone();
             let cancelling =
                 loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter));
             returning.join().unwrap();
             assert!(cancelling.join().unwrap());
-2397
             assert_eq!((1, 1), cell.h1_counts());
             assert_eq!(0, cell.snapshot().retained);
             assert_eq!(
 ,
                 OriginCell::select_h1(&cell)
                     .expect("sender was lost during return/cancel")
                     .test_sender_id()
             );
         });
-2407
+    }
-2408
     #[test]
     fn returned_h1_and_establishment_race_for_one_waiter() {
         loom::model(|| {
             let cell = unbounded_cell();
             let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
             let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
             let AcquisitionEvent::Establish(permit) = cell
                 .take_ready_event(waiter)
                 .expect("unbounded miss did not start establishment")
             else {
                 panic!("unbounded miss completed before establishment started");
             };
             assert!(cell.start_establishment(waiter));
             drop(permit);
-2422
             let (returning_connection, _returning_physical) =
                 ConnectionState::unbounded(connection_info(1));
             let returning =
                 OriginCell::install_selected_h1(&cell, returning_connection, H1Sender::test(11));
             let (fresh_connection, _fresh_physical) =
                 ConnectionState::unbounded(connection_info(2));
             let fresh =
                 OriginCell::install_selected_h1(&cell, fresh_connection, H1Sender::test(22));
-2431
             let returning = loom::thread::spawn(move || drop(returning));
             let completion_cell = cell.clone();
             let completing = loom::thread::spawn(move || {
                 completion_cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
             });
             returning.join().unwrap();
             completing.join().unwrap();
-2439
             assert_eq!((2, 1), cell.h1_counts());
             let winner = cell
                 .take_ready_h1(waiter)
                 .expect("both acquisition results were lost");
             assert!(matches!(winner.test_sender_id(), 11 | 22));
             drop(winner);
             assert_eq!((2, 2), cell.h1_counts());
         });
-2448
+    }
-2449
     #[test]
     fn first_establishment_poll_races_returned_h1() {
         loom::model(|| {
             let cell = unbounded_cell();
             let waiter = cell.register_waiter(ProtocolRequirement::H1Compatible);
             let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
             let AcquisitionEvent::Establish(permit) = cell
                 .take_ready_event(waiter)
                 .expect("unbounded miss did not prepare establishment")
             else {
                 panic!("unbounded miss completed before establishment was prepared");
             };
-2461
             let (returning_connection, _returning_physical) =
                 ConnectionState::unbounded(connection_info(1));
             let returning =

↧ Expand context

                 OriginCell::install_selected_h1(&cell, returning_connection, H1Sender::test(11));
             let return_thread = loom::thread::spawn(move || drop(returning));
             let start_cell = cell.clone();
             let start_thread = loom::thread::spawn(move || start_cell.start_establishment(waiter));
-2469
             return_thread.join().unwrap();
             let started = start_thread.join().unwrap();
             drop(permit);
-2473
             if started {
                 let (fresh_connection, _fresh_physical) =
                     ConnectionState::unbounded(connection_info(2));
                 let fresh =
                     OriginCell::install_selected_h1(&cell, fresh_connection, H1Sender::test(22));
                 cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
-2480
+            }
-2481
             let winner = cell
                 .take_ready_h1(waiter)
                 .expect("returned H1 was lost at the first-poll boundary");

@@ -2498,2511 +2558,2572 @@

↥ Expand context

-2511
             let delivery_cell = cell.clone();
             let delivering = loom::thread::spawn(move || {
                 drop(OriginCell::receive_delivery(&delivery_cell, delivery));
             });
             let returning = loom::thread::spawn(move || drop(returning));
             delivering.join().unwrap();
             returning.join().unwrap();
-2519
             assert_eq!(1, admission.available_capacity_for_test());
             let selection = cell
                 .take_ready_h1(waiter)
                 .expect("return and capacity delivery lost the HTTP/1 sender");
             assert_eq!(11, selection.test_sender_id());
             drop(selection);
             admission.clear_modeled_cells_for_test();
         });
-2528
+    }
-2529
     #[test]

     fn peer_reuse_and_request_cancellation_preserve_the_sender() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound = Some(2);
         model.check(|| {
             let (admission, connection_cell, requesting_cell) =
                 bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
             let lease = OriginAdmission::lease_for_test(&admission);
             let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
             let returning =
                 OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
             let waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
             let waiter =
                 OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
-2543
             let returning = loom::thread::spawn(move || drop(returning));
             let cancel_requesting_cell = requesting_cell.clone();
             let cancelling = loom::thread::spawn(move || {
                 OriginCell::cancel_waiter(&cancel_requesting_cell, waiter)
             });
             returning.join().unwrap();
             assert!(cancelling.join().unwrap());
-2551
             assert_eq!((1, 1), connection_cell.h1_counts());

↧ Expand context

             assert_eq!(0, requesting_cell.snapshot().retained);
             assert_eq!(0, admission.ordered_demand_count_for_test());
             admission.clear_modeled_cells_for_test();
         });
-2557
+    }
-2558
     #[test]
     fn borrowed_delivery_racing_a_local_return_preserves_both_senders() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound = Some(2);
         model.check(|| {
             let (admission, connection_cell, requesting_cell) =
                 bounded_peer_cells_with_limit(2, EligibilityGroup::Pool, EligibilityGroup::Pool);
             let owning_lease = OriginAdmission::lease_for_test(&admission);
             let (owned_connection, _owning_physical) =
                 ConnectionState::bounded(connection_info(1), owning_lease);
             OriginCell::install_idle_h1(&connection_cell, owned_connection, H1Sender::test(11));
             let requesting_lease = OriginAdmission::lease_for_test(&admission);
             let (requesting_connection, _requesting_physical) =
                 ConnectionState::bounded(connection_info(2), requesting_lease);

@@ -2614,2628 +2721,2868 @@

↥ Expand context

                 &admission,
                 requesting_cell.id().partition(),
                 demand,
-2631
+            )
             .expect("peer demand did not prepare a reuse operation");
             let delivery = install
                 .drive_once_for_test()
                 .expect("reuse operation installation did not prepare a delivery");
-2636
             let delivering = loom::thread::spawn(move || {
                 OriginAdmission::drive(Some(delivery));
             });
             let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
             delivering.join().unwrap();
             closing.join().unwrap();
-2643
             let event = requesting_cell
                 .take_ready_event(waiter)
                 .expect("borrow/close race stranded its requesting cell waiter");
             match event {

                 AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => {
                     drop(selection);
-2650
+                }
                 AcquisitionEvent::Establish(permit) => {
                     assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
                     drop(permit);
-2654
+                }
                 AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
                     panic!("unexpected establishment failure: {error}")
-2657
+                }
                 AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
                     panic!("HTTP/1 reuse model requested reacquisition")
-+
+                }
                 AcquisitionEvent::Complete(AcquisitionResult::H2(_)) => {
                     panic!("HTTP/1 reuse model received an HTTP/2 activation")
-+
+                }
-2664
+            }
             assert_eq!(0, requesting_cell.snapshot().retained);
             assert_eq!(1, admission.available_capacity_for_test());
             admission.clear_modeled_cells_for_test();
         });
-2669
+    }
-2670
     #[test]
     fn reclaim_and_connection_close_release_exactly_one_capacity_slot() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound = Some(2);
         model.check(|| {
             let (admission, connection_cell, requesting_cell) = bounded_peer_cells(
                 EligibilityGroup::Partition(PartitionId::from_index(1)),
                 EligibilityGroup::Partition(PartitionId::from_index(2)),
             );
             let lease = OriginAdmission::lease_for_test(&admission);
             let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
             let returning = OriginCell::install_selected_h1(
                 &connection_cell,
                 connection.clone(),
                 H1Sender::test(11),
             );
             let close = H1CloseHandle::new(&connection_cell, &connection);
             let waiter = requesting_cell.register_waiter(ProtocolRequirement::H1Compatible);
             let local_waiter = connection_cell.register_waiter(ProtocolRequirement::H1Compatible);
             let waiter =
                 OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
             let local_waiter =
                 OriginCell::register_waiter(&connection_cell, ProtocolRequirement::H1Compatible);
-2692
             let returning = loom::thread::spawn(move || drop(returning));
             let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
             returning.join().unwrap();
             let close_won = closing.join().unwrap();
-2697
             assert_eq!(
                 !close_won,
                 connection_cell.state.lock().reuse.local_turn_owed(),
                 "only a completed reclaim may earn the connection-owning cell a local fairness turn"
             );
-2703
             let replacement = OriginCell::take_ready_lease(&requesting_cell, waiter)
                 .expect("close/reclaim race did not deliver released capacity");
             assert_eq!((0, 0), connection_cell.h1_counts());
             assert_eq!(0, admission.available_capacity_for_test());
             assert!(OriginCell::cancel_waiter(&connection_cell, local_waiter));
             drop(replacement);
             assert_eq!(1, admission.available_capacity_for_test());
             admission.clear_modeled_cells_for_test();
         });
-2713
+    }
-2714
     #[test]
     fn delivery_and_cancellation_race_refunnels_capacity() {
         loom::model(|| {
             let (admission, cell) = bounded_cell();
             let (first, demand) =
                 cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
             let mut delivery =
                 OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
                     .unwrap();
             assert!(
                 delivery.materialize_for_test(),
                 "capacity delivery did not materialize"
             );
-2728
             let delivery_cell = cell.clone();
             let deliver = loom::thread::spawn(move || {
                 drop(OriginCell::receive_delivery(&delivery_cell, delivery));
             });
             let cancel_cell = cell.clone();
             let cancel =
                 loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, first));
             deliver.join().unwrap();
             cancel.join().unwrap();
-2738
             assert_eq!(1, admission.available_capacity_for_test());
             admission.clear_modeled_cells_for_test();
         });
-2742
+    }
-+
     #[test]
     fn h2_route_installation_races_requesting_cell_cancellation() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound = Some(2);
         model.check(|| {
             let (admission, connection_cell, requesting_cell) =
                 bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
             let lease = OriginAdmission::lease_for_test(&admission);
             let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
             let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
             let (waiter, demand) =
                 requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
             let route = h2::H2Route::new(&connection_cell, generation);
-+
             let install_cell = requesting_cell.clone();
             let installing = loom::thread::spawn(move || {
                 OriginCell::install_h2_route(
                     &install_cell,
                     route,
                     &EligibilityGroup::Pool,
                     demand.id_for_test(),
-+
+                )
             });
             let cancel_cell = requesting_cell.clone();
             let cancelling =
                 loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter));
             let _installed = installing.join().unwrap();
             assert!(cancelling.join().unwrap());
-+
             assert_eq!(0, requesting_cell.snapshot().retained);
             assert_eq!(0, admission.ordered_demand_count_for_test());
             assert_eq!(0, admission.available_capacity_for_test());
             assert!(OriginCell::close_h2(
                 &connection_cell,
                 generation,
                 CloseReason::PoolDropped,
             ));
             assert_eq!(1, admission.available_capacity_for_test());
             admission.clear_modeled_cells_for_test();
         });
-+
+    }
-+
     #[test]
     fn h2_publication_acknowledgement_races_generation_close_and_route_service() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound = Some(3);
         model.check(|| {
             let (admission, connection_cell, requesting_cell) =
                 bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
             let lease = OriginAdmission::lease_for_test(&admission);
             let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
             let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
             let (waiter, demand) =
                 requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
             let action = OriginAdmission::publish_action_without_driving(
                 &admission,
                 requesting_cell.id().partition(),
                 demand,
-+
+            )
             .expect("peer demand did not prepare publication");
-+
             let publishing = loom::thread::spawn(move || {
                 OriginAdmission::drive(Some(action));
             });
             let closing_cell = connection_cell.clone();
             let closing = loom::thread::spawn(move || {
                 OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
             });
             let servicing_cell = requesting_cell.clone();
             let servicing = loom::thread::spawn(move || {
                 OriginCell::service_peer_h2_waiters(&servicing_cell);
             });
-+
             publishing.join().unwrap();
             assert!(closing.join().unwrap());
             servicing.join().unwrap();
-+
             assert!(
                 OriginCell::cancel_waiter(&requesting_cell, waiter),
                 "publication acknowledgement lost the live demand"
             );
             assert_eq!(0, requesting_cell.snapshot().retained);
             assert_eq!(0, admission.ordered_demand_count_for_test());
             assert_eq!(1, admission.available_capacity_for_test());
             admission.clear_modeled_cells_for_test();
         });
-+
+    }
-+
     #[test]
     fn h2_publication_close_race_preserves_capacity_ownership() {
         let mut model = loom::model::Builder::new();
         model.preemption_bound = Some(2);
         model.check(|| {
             let (admission, connection_cell, requesting_cell) =
                 bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
             let lease = OriginAdmission::lease_for_test(&admission);
             let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
             let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
             let (waiter, demand) =
                 requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
             let action = OriginAdmission::publish_action_without_driving(
                 &admission,
                 requesting_cell.id().partition(),
                 demand,
-+
+            )
             .expect("peer demand did not prepare publication");
-+
             let publishing = loom::thread::spawn(move || {
                 OriginAdmission::drive(Some(action));
             });
             let closing_cell = connection_cell.clone();
             let closing = loom::thread::spawn(move || {
                 OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
             });
             publishing.join().unwrap();
             assert!(closing.join().unwrap());
-+
             assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
             assert_eq!(0, requesting_cell.snapshot().retained);
             assert_eq!(0, admission.ordered_demand_count_for_test());
             assert_eq!(1, admission.available_capacity_for_test());
             admission.clear_modeled_cells_for_test();
         });
-+
+    }
-2868
+}

 //!
 //! Admission owns the operation record. The connection cell retains a
 //! local `H1ReuseReservation` serialized with sender return. Values crossing
 //! between those lock domains own their fallback: an uninstalled operation is
 //! cancelled, and an uncommitted candidate returns its sender before demand
 //! becomes schedulable again.
 //!
 //! Admission advances each operation through these phases:
 //!
 //! ```text
 //! Installing -- reservation installed ----------------------> Installed
 //! Installing -- sender extracted ---------------------------> Resolving
 //! Installing(cancelled) -- reservation installed -----------> Cancelling
 //! Installing(cancelled) -- sender extracted ----------------> Resolving
 //! Installing -- reservation rejected or cell expired -------> removed
 //! Installed -- sender return intercepted -------------------> Resolving
 //! Installed -- demand cancelled ----------------------------> Cancelling
 //! Installed -- terminal cell outcome ------------------------> removed
 //! Cancelling -- sender return intercepted ------------------> Resolving
 //! Cancelling -- reservation cleared -------------------------> removed

 //! Resolving -- repeated candidate handoff ------------------> Resolving
 //! Resolving -- borrow or reclaim completes ------------------> removed
 //! ```
 //!
 //! A sender extracted during installation takes the `Resolving` path even if
 //! cancellation raced with the install acknowledgement. The extracted sender
 //! must complete its terminal transition before cancellation can finish.
-39
 use super::{
     AdmissionAction, DeliveryAckResult, DeliveryGuard, DeliveryId, DemandId, DemandSchedule,
     IntrusiveLinks, IntrusiveOrder, OriginAdmission, ProtocolRequirement,
     IntrusiveLinks, IntrusiveOrder, OriginAdmission,
 };
 use crate::client::pool::cell::h1::{H1Selection, ProvisionalH1};
 use crate::client::pool::cell::OriginCell;
 use crate::client::pool::partition::{EligibilityGroup, PartitionId};
 use crate::sync::Arc;
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use std::collections::{HashMap, VecDeque};
 use std::fmt;
-51
 /// Identity of one cross-cell HTTP/1 reuse operation.

 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub(in crate::client::pool) struct ReuseId(u64);
-55
 impl ReuseId {
     /// Creates a deterministic identity for focused transition tests.
     #[cfg(test)]
     pub(in crate::client::pool) const fn for_test(value: u64) -> Self {
         Self(value)
-61
+    }
-62
+}
-63
 /// How an HTTP/1 reuse operation satisfies waiting demand.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(in crate::client::pool) enum ReuseMode {
     /// Move the selected sender to the requesting cell for dispatch.
     Borrow,
     /// Close the selected connection and deliver its capacity instead.
     Reclaim,
-71
+}
-72

         &mut self,
         requesting_partition: &PartitionId,
         schedule: &DemandSchedule,
     ) {
         let Some(reuse_id) = self
             .requesting_partitions
             .get(requesting_partition)
             .copied()
         else {
             return;
         };
         let Some(record) = self.operations.get_mut(&reuse_id) else {
             self.requesting_partitions.remove(requesting_partition);
             self.assert_consistent();
             return;
         };
         if schedule.is_current_queued(&record.requesting_partition, record.demand) {
             return;
-357
+        }
         record.cancelled = true;

         if record.phase == ReusePhase::Installed {
             record.phase = ReusePhase::Cancelling;
             self.cancellations.push_back(reuse_id);
-362
+        }
         self.assert_consistent();
-364
+    }
-365
     /// Selects another cell's connection for the oldest origin-wide demand.
     ///
     /// An eligible cell lends its sender. Otherwise, a connection is reclaimed
     /// so its capacity can satisfy the same demand. A cell is never selected for
     /// its own demand; local idle and returning senders are handled under the
     /// cell lock.
     /// so its capacity can satisfy the same demand. HTTP/1-compatible demand
     /// does not select its own cell because local senders are handled under the
     /// cell lock. HTTP/2-only demand may reclaim same-cell idle HTTP/1 capacity.
     pub(super) fn prepare_reuse(
         &mut self,
         schedule: &DemandSchedule,
     ) -> Option<PreparedReuseInstall> {
         let requesting_partition = schedule.queued_head()?;
         if self
             .requesting_partitions
             .contains_key(&requesting_partition.requesting_partition)
-380
+        {
             return None;
-382
+        }
-383
         let (connection_partition, mode) =
             if requesting_partition.requirement == ProtocolRequirement::H1Compatible {
         let (connection_partition, mode) = if requesting_partition.requirement.accepts_h1() {
             match self.take_group_peer(
                 &requesting_partition.eligibility_group,
                 &requesting_partition.requesting_partition,
             ) {
                 Some(connection_partition) => (connection_partition, ReuseMode::Borrow),
                 None => (
                     self.take_origin_peer(&requesting_partition.requesting_partition)?,
                     ReuseMode::Reclaim,
                 ),
-394
+            }
         } else {
--
+                (
                     self.take_origin_peer(&requesting_partition.requesting_partition)?,
                     ReuseMode::Reclaim,
--
+                )
             (self.take_origin_head()?, ReuseMode::Reclaim)
         };
-398
         let id = self.take_reuse_id();
         let availability_record = self
             .connection_partitions
             .get_mut(&connection_partition)
             .expect("selected HTTP/1 connection cell disappeared");
         debug_assert!(availability_record.reuse_id.is_none());
         availability_record.reuse_id = Some(id);
         self.requesting_partitions

AWS SDK

Files changed:

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/admission/reuse.rs

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

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

             .insert(requesting_partition.requesting_partition, id);
         self.operations.insert(
             id,
             ReuseOperation {
                 connection_partition,
                 requesting_partition: requesting_partition.requesting_partition,
                 demand: requesting_partition.demand,
                 mode,
                 phase: ReusePhase::Installing,
                 cancelled: false,
             },
         );
         self.assert_consistent();
         Some(PreparedReuseInstall {
             id,
             connection_partition,
         })
-424
+    }
-425
     /// Extracts cancellation work, discarding entries for completed operations.

                 self.connection_partitions.remove(&connection_partition);
-537
+            }
-538
+        }
-539
+    }
-540
     /// Removes and returns the first eligible peer in one reuse group.
     fn take_group_peer(
         &mut self,
         group: &EligibilityGroup,
         requesting_partition: &PartitionId,
     ) -> Option<PartitionId> {
         let connection_partition = {
             let order = self.group_orders.get(group)?;
             self.first_peer(order, requesting_partition, AvailabilityOrderView::Borrow)?
         };
         self.unlink_cell(&connection_partition);
         Some(connection_partition)
-553
+    }
-554
     /// Removes and returns the first origin-wide peer.

     fn take_origin_peer(&mut self, requesting_partition: &PartitionId) -> Option<PartitionId> {
         let connection_partition = self.first_peer(
             &self.origin_order,
             requesting_partition,
             AvailabilityOrderView::Reclaim,
         )?;
         self.unlink_cell(&connection_partition);
         Some(connection_partition)
-564
+    }
-565
     /// Removes and returns the oldest origin-wide connection, including local.
     fn take_origin_head(&mut self) -> Option<PartitionId> {
         let connection_partition = self.origin_order.head()?;
         self.unlink_cell(&connection_partition);
         Some(connection_partition)
-+
+    }
-+
     /// Returns the head, or its successor when the head is the requesting cell itself.
     fn first_peer(
         &self,
         order: &IntrusiveOrder<PartitionId>,
         requesting_partition: &PartitionId,
         view: AvailabilityOrderView,
     ) -> Option<PartitionId> {
         let head = order.head()?;
         if head != *requesting_partition {
             return Some(head);

-583
+        }
         let record = self
             .connection_partitions
             .get(&head)
             .expect("connection cell order head disappeared");
         record.residence.links(view).and_then(|links| links.next)
-589
+    }
-590
     /// Appends an available, unreserved cell to both scheduling views.
     fn enqueue_cell_if_available(&mut self, connection_partition: &PartitionId) {
         let Some(record) = self.connection_partitions.get(connection_partition) else {
             return;
         };
         if !record.is_schedulable() || record.residence.is_available() {
             return;
-598
+        }
         let group = record.group.clone();
         let origin_links = self.origin_order.push_back(*connection_partition);
         let group_links = self
             .group_orders

                 .get_mut(&previous)
                 .expect("previous connection cell disappeared")
                 .residence
                 .links_mut(view)
                 .next = links.next;
-678
+        }
         if let Some(next) = links.next {
             connection_partitions
                 .get_mut(&next)
                 .expect("next connection cell disappeared")
                 .residence
                 .links_mut(view)
                 .previous = links.previous;
-686
+        }
         debug_assert_ne!(links.previous.as_ref(), Some(connection_partition));
         debug_assert_ne!(links.next.as_ref(), Some(connection_partition));
-689
+    }
-690
     /// Allocates a reuse operation identity.
     fn take_reuse_id(&mut self) -> ReuseId {

         let value = self.next_reuse;
         self.next_reuse = value
             .checked_add(1)
             .expect("HTTP/1 reuse operation identity exhausted");
         ReuseId(value)
-698
+    }
-699
     /// Checks reuse-operation indexes and both availability orders after every mutation.
     fn assert_consistent(&self) {
         #[cfg(debug_assertions)]
-+
+        {
             if std::thread::panicking() {
                 return;
-+
+            }
             self.assert_consistent_debug();
-708
+        }
-+
+    }
-710
     #[cfg(debug_assertions)]
     fn assert_consistent_debug(&self) {
         for (id, operation) in &self.operations {
             assert_eq!(
                 self.requesting_partitions
                     .get(&operation.requesting_partition),
                 Some(id),
                 "reuse operation's requesting-cell index did not name the operation"
             );

             assert_eq!(
                 self.connection_partitions
                     .get(&operation.connection_partition)
                     .and_then(|connection_partition| connection_partition.reuse_id),
                 Some(*id),
                 "reuse operation's connection-cell index did not name the operation"
             );
-727
+        }
         for (requesting_partition, id) in &self.requesting_partitions {
             assert_eq!(
                 self.operations
                     .get(id)
                     .map(|operation| &operation.requesting_partition),
                 Some(requesting_partition),
                 "requesting cell index named a missing reuse operation"
             );
-736
+        }
         for (connection_partition, record) in &self.connection_partitions {
             if let Some(id) = record.reuse_id {
                 assert_eq!(

-1203
+            }
-1204
+        }
-1205
+    }
-1206
     /// Resolves a provisional sender through borrow or reclaim policy.
     pub(in crate::client::pool) fn resolve_h1_reuse(
         origin: &Arc<Self>,
         id: ReuseId,
         candidate: ReuseCandidate,
     ) -> Option<AdmissionAction> {
         let action = {
             let mut state = origin.state.lock();
             let Some(record) = state.h1.begin_resolution(id) else {
                 drop(state);
                 drop(candidate);
                 return None;
             };
             if record.cancelled
                 || !state
                     .demand_schedule

                     .is_current_queued(&record.requesting_partition, record.demand)
-1224
+            {
                 drop(state);
                 drop(candidate);
                 return None;
-1228
+            }
-1229
             match record.mode {
                 ReuseMode::Borrow => {
                     let delivery = state.take_delivery_id();
                     let old_group = state
                         .demand_schedule
                         .group_for(&record.requesting_partition);
                     let Some(scheduled) = state.demand_schedule.reserve_reuse_demand(
                         &record.requesting_partition,
                         record.demand,
                         delivery,
                     ) else {
                         drop(state);
                         drop(candidate);
                         return None;
                     };
                     state.reconcile_demand_indexes(&scheduled.requesting_partition, old_group);
                     Some(AdmissionAction::Delivery(DeliveryGuard::borrowed_h1(
                         origin.clone(),
                         delivery,
                         scheduled.requesting_partition,
                         scheduled.demand,
                         id,
                         record.connection_partition,
                         candidate,
                     )))
-1255
+                }

                 ReuseMode::Reclaim => {
                     Some(AdmissionAction::H1(H1ReuseAction::Reclaim(ReclaimAction {
                         origin: origin.clone(),
                         requesting_partition: record.requesting_partition,
                         reuse_id: id,
                         candidate: Some(candidate),
                     })))
-1263
+                }
-1264
+            }
         };
         action
-1267
+    }
-1268
     /// Applies a terminal connection cell outcome and schedules the next admission action.
     fn finish_h1_reuse(
         origin: &Arc<Self>,
         id: ReuseId,
         outcome: H1AvailabilityOutcome,
     ) -> Option<AdmissionAction> {
         let mut state = origin.state.lock();

         delivery: DeliveryId,
         requesting_partition: &PartitionId,
         result: DeliveryAckResult,
         transferred_connection_cell: Option<PartitionId>,
         rejected_outcome: Option<H1AvailabilityOutcome>,
     ) -> Option<AdmissionAction> {
         let mut state = origin.state.lock();
         state.finish_delivery(delivery, requesting_partition, result);
         if let Some(connection_partition) = transferred_connection_cell {
             return Some(AdmissionAction::H1(H1ReuseAction::CompleteConnectionCell(
                 ConnectionCellCompletion {
                     origin: origin.clone(),
                     reuse_id,
                     connection_partition,
                     transferred: true,
                     active: true,
                 },
             )));
-1302
+        }
         state.h1.finish_reuse(

             reuse_id,
             rejected_outcome.expect("rejected borrow had no connection cell outcome"),
         );
         Self::prepare_action(origin, &mut state)
-1308
+    }
-1309
+}
-1310
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use crate::client::pool::admission::{DemandSnapshot, SnapshotVersion};
     use crate::client::pool::admission::{DemandSnapshot, ProtocolRequirement, SnapshotVersion};
     use crate::client::pool::partition::PartitionId;
-1316
     fn cell(index: usize) -> PartitionId {
         PartitionId::from_index(index)
-1319
+    }
-1320
     fn schedule(requesting_partition: PartitionId, group: EligibilityGroup) -> DemandSchedule {
         let mut schedule = DemandSchedule::default();
         schedule.publish(
             requesting_partition,

             DemandSnapshot::active(
                 DemandId::from_u64(1),
                 SnapshotVersion::INITIAL,
                 ProtocolRequirement::H1Compatible,
                 group,
             ),
         );
         schedule
-1333
+    }
-1334
     fn availability_snapshot(
         revision: u64,
         advertised: bool,
         blocked: bool,
     ) -> H1AvailabilitySnapshot {
         H1AvailabilitySnapshot::new(
             revision,
             H1Availability {
                 advertised,
                 blocked,

             );
-1361
+        }
-1362
         assert_eq!(1, coordination.connection_partitions.len());
         assert_eq!(1, coordination.origin_order.len());
         assert_eq!(
 ,
             coordination
                 .group_orders
                 .get(&group)
                 .expect("connection cell group was not published")
                 .len()
         );
-1373
         coordination.update_availability(
             connection_partition,
             group.clone(),
             availability_snapshot(501, false, false),
         );
         assert_eq!(0, coordination.origin_order.len());

         );
         coordination.update_availability(
             peer,
             group.clone(),
             availability_snapshot(1, true, false),
         );
-1438
         let reuse_id = coordination
             .prepare_reuse(&schedule(requesting_partition, group))
             .expect("peer connection cell was not selected");
         assert_eq!(peer, reuse_id.connection_partition);
         assert_ne!(requesting_partition, reuse_id.connection_partition);
-1444
+    }
-1445
     #[test]
     fn expired_connection_cell_terminates_its_reuse_without_republication() {
         let connection_partition = cell(1);
         let requesting_partition = cell(2);
         let group = EligibilityGroup::Pool;
         let schedule = schedule(requesting_partition, group.clone());

  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Construction and validation for [`ConnectionPool`].
-7
 use super::establish::{self, TransportFactory};
 use super::maintenance::MaintenanceConfig;
 use super::registry::{PartitionRegistry, PartitionRegistryError};
 use super::{ConnectionPool, ConnectionReuseScope, Partition, PoolConfig, PoolInner};
 use crate::client::TlsUnset;
 use crate::client::{TlsProviderSelected, TlsUnset};
 use crate::sync::Arc;
 use crate::tls::{self, TlsContext};
 use aws_smithy_async::rt::sleep::{default_async_sleep, AsyncSleep, SharedAsyncSleep};
 use aws_smithy_async::time::{SharedTimeSource, TimeSource};
 #[cfg(any(
     all(test, feature = "rt-tokio"),
     all(feature = "test-util", aws_sdk_unstable)
 ))]
 use aws_smithy_runtime_api::box_error::BoxError;
 #[cfg(any(
     all(test, feature = "rt-tokio"),
     all(feature = "test-util", aws_sdk_unstable)

 ))]
 use http_1x::Uri;
 use hyper_util::client::legacy::connect::dns::GaiResolver;
 use hyper_util::client::legacy::connect::HttpConnector;
 use std::error::Error;
 use std::fmt;
 use std::num::NonZeroUsize;
 use std::sync::atomic::AtomicU64;
 use std::time::Duration;
 #[cfg(any(
     all(test, feature = "rt-tokio"),
     all(feature = "test-util", aws_sdk_unstable)
 ))]
 use tower::Service;
-39
 /// Default duration for retaining an idle reusable connection.
 const DEFAULT_IDLE_TIMEOUT: Duration = Duration::from_secs(90);
-42
 #[derive(Clone, Debug, Default, Eq, PartialEq)]
 enum TriStateOption<T> {

     pub fn set_max_connections_per_host(&mut self, limit: Option<usize>) -> &mut Self {
         self.max_connections_per_host = limit;
         self
-217
+    }
-218
     /// Selects which partitions may reuse each other's connections.
     pub fn connection_reuse_scope(mut self, scope: ConnectionReuseScope) -> Self {
         self.reuse_scope = scope;
         self
-223
+    }
-224
     /// Mutably selects the connection reuse scope.
     pub fn set_connection_reuse_scope(&mut self, scope: ConnectionReuseScope) -> &mut Self {
         self.reuse_scope = scope;
         self
-229
+    }
-230
     /// Declares the complete explicit partition set.
     pub fn partitions(mut self, partitions: impl IntoIterator<Item = Partition>) -> Self {
         self.partitions = Some(partitions.into_iter().collect());

         self
-235
+    }
-236
     /// Mutably declares or clears the explicit partition set.
     pub fn set_partitions(&mut self, partitions: Option<Vec<Partition>>) -> &mut Self {
         self.partitions = partitions;
         self
-241
+    }
-242
+}
-243
 impl Builder<TlsUnset> {
     /// Selects the TLS provider used by HTTPS pool connections.
     pub fn tls_provider(self, provider: tls::Provider) -> Builder<TlsProviderSelected> {
         Builder {
             idle_timeout: self.idle_timeout,
             time_source: self.time_source,
             sleep_impl: self.sleep_impl,
             tcp_nodelay: self.tcp_nodelay,
             tcp_keepalive: self.tcp_keepalive,
             max_connections_per_host: self.max_connections_per_host,
             reuse_scope: self.reuse_scope,
             partitions: self.partitions,
             tls: TlsProviderSelected {
                 provider,
                 context: TlsContext::default(),
             },
-+
+        }
-+
+    }
-+
+}
-+
 crate::cfg::cfg_tls! {
     impl Builder<TlsProviderSelected> {
         /// Sets provider-specific TLS configuration for HTTPS connections.
         pub fn tls_context(mut self, context: TlsContext) -> Self {
             self.tls.context = context;
             self
-+
+        }
-+
         /// Mutably sets provider-specific TLS configuration.
         pub fn set_tls_context(&mut self, context: TlsContext) -> &mut Self {
             self.tls.context = context;
             self
-+
+        }
-+
         /// Builds a pool whose connector performs TLS and ALPN negotiation.
         ///
         /// This builder currently creates direct connections. Proxy routing
         /// remains available through the existing `Connector` API.
         pub fn build_https(self) -> Result<ConnectionPool, BuildError> {
             validate_default_connector_interfaces(self.partitions.as_deref())?;
             let provider = self.tls.provider.clone();
             let context = self.tls.context.clone();
             match provider {
                 #[cfg(feature = "__rustls")]
                 tls::Provider::Rustls(crypto_mode) => {
                     let tcp_nodelay = self.tcp_nodelay;
                     let tcp_keepalive = self.tcp_keepalive.clone().resolve(None);
                     let transport = establish::cached_transport_factory_for_interface(
                         move |interface, alpn_protocols| {
                             let mut connector =
                                 HttpConnector::new_with_resolver(GaiResolver::new());
                             connector.set_nodelay(tcp_nodelay);
                             connector.set_keepalive(tcp_keepalive);
                             set_default_connector_interface(&mut connector, interface);
                             tls::rustls_provider::build_connector::wrap_connector_with_alpn(
                                 connector,
                                 crypto_mode.clone(),
                                 &context,
                                 crate::proxy::ProxyConfig::disabled(),
                                 alpn_protocols,
-+
+                            )
                         },
                         true,
                     );
                     self.build_with_transport(transport)
-+
+                }
                 #[cfg(feature = "s2n-tls")]
                 tls::Provider::S2nTls => {
                     let tcp_nodelay = self.tcp_nodelay;
                     let tcp_keepalive = self.tcp_keepalive.clone().resolve(None);
                     let transport = establish::cached_transport_factory_for_interface(
                         move |interface, _alpn_protocols| {
                             let mut connector =
                                 HttpConnector::new_with_resolver(GaiResolver::new());
                             connector.set_nodelay(tcp_nodelay);
                             connector.set_keepalive(tcp_keepalive);
                             set_default_connector_interface(&mut connector, interface);
                             tls::s2n_tls_provider::build_connector::wrap_connector(
                                 connector,
                                 &context,
                                 crate::proxy::ProxyConfig::disabled(),
-+
+                            )
                         },
                         false,
                     );
                     self.build_with_transport(transport)
-+
+                }
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 impl Builder<TlsUnset> {
     /// Builds a pool for cleartext HTTP connections.
     ///
     /// This method remains public for explicit cleartext endpoints and tests,
     /// but is hidden from generated documentation because HTTPS with automatic
     /// ALPN negotiation is the intended production entry point.
     #[doc(hidden)]
     pub fn build_http(self) -> Result<ConnectionPool, BuildError> {
         validate_default_connector_interfaces(self.partitions.as_deref())?;
         let mut connector = HttpConnector::new_with_resolver(GaiResolver::new());
         connector.set_nodelay(self.tcp_nodelay);
         connector.set_keepalive(self.tcp_keepalive.clone().resolve(None));
         let transport = establish::transport_factory_for_interface(move |interface| {
             let mut connector = connector.clone();
             set_default_connector_interface(&mut connector, interface);
             connector
         });
         self.build_with_transport(transport)
-354
+    }
-355
     /// Builds a pool around a test transport connector.
     #[cfg(all(feature = "test-util", aws_sdk_unstable))]
     #[doc(hidden)]
     pub fn build_http_with_tcp_connector<C, IO>(
         self,
         connector: C,
     ) -> Result<ConnectionPool, BuildError>
     where
         C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
         C::Error: Into<BoxError>,
         C::Future: Send + 'static,
         IO: hyper::rt::Read
             + hyper::rt::Write
             + hyper_util::client::legacy::connect::Connection
             + Send
             + Sync
             + Unpin
             + 'static,
-374
+    {
         self.build_with_transport(establish::transport_factory(connector))
-376
+    }
-377
     #[cfg(all(test, feature = "rt-tokio"))]
     pub(super) fn build_with_connector<C, IO>(
         self,
         connector: C,
     ) -> Result<ConnectionPool, BuildError>
     where
         C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
         C::Error: Into<BoxError>,
         C::Future: Send + 'static,
         IO: hyper::rt::Read
             + hyper::rt::Write
             + hyper_util::client::legacy::connect::Connection
             + Send
             + Sync
             + Unpin
             + 'static,
-394
+    {
         self.build_with_transport(establish::transport_factory(connector))
-396
+    }
-+
+}
-398
 impl<Tls> Builder<Tls> {
     /// Validates pool policy and installs the type-erased transport factory.
     fn build_with_transport(
         self,
         transport: std::sync::Arc<dyn TransportFactory>,
     ) -> Result<ConnectionPool, BuildError> {
         let max_connections_per_host = match self.max_connections_per_host {
             Some(0) => {
                 return Err(BuildError::new(
                     "max_connections_per_host must be greater than zero",
                 ))
-410
+            }
             Some(limit) => NonZeroUsize::new(limit),
             None => None,
         };
         let idle_timeout = self.idle_timeout.resolve(Some(DEFAULT_IDLE_TIMEOUT));
         let time_source = self.time_source.unwrap_or_default();
         let sleep_impl = self.sleep_impl.or_else(default_async_sleep);
         if idle_timeout.is_some() && sleep_impl.is_none() {
             return Err(BuildError::new(
                 "idle_timeout requires an async sleep implementation",
             ));
-421
+        }
         let maintenance = MaintenanceConfig {
             idle_timeout,
             time_source: time_source.clone(),
             sleep: sleep_impl.clone(),
         };
         let allow_h2_reclaim_for_h1 = transport.guarantees_http1();
         let registry = PartitionRegistry::new(
             self.partitions,
             self.reuse_scope,
             max_connections_per_host,
             allow_h2_reclaim_for_h1,
             maintenance,
-434
+        )
         .map_err(BuildError::from)?;
         let config = PoolConfig {
             idle_timeout,
             max_connections_per_host,
             reuse_scope: self.reuse_scope,
         };
         Ok(ConnectionPool {
             inner: Arc::new(PoolInner {

                 config,
                 registry,
                 transport,
                 next_connection_id: AtomicU64::new(0),
             }),
         })
-449
+    }
-450
+}
-451
 /// Error returned when a pool configuration is internally inconsistent.
 #[derive(Debug)]
 pub struct BuildError {
     /// Stable configuration diagnostic.
     message: String,
-457
+}
-458
 impl BuildError {
     fn new(message: impl Into<String>) -> Self {
         Self {
             message: message.into(),