AWS SDK

AWS SDK

rev. 01140188efe3be26ec6653bc30094c80ed1a1b99

Files changed:

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

@@ -1,1 +536,575 @@

↥ Expand context

  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! One-to-one acquisition payload delivery across admission and cell locks.
 //!
 //! Admission reserves one demand generation and moves either bounded capacity
 //! or a provisional HTTP/1 sender into a [`DeliveryGuard`]. The guard
 //! commits any connection cell transition before taking the requesting
 //! cell lock, then owns the payload until that cell accepts it. Its
 //! acknowledgement keeps the demand fence installed until requesting-cell
 //! settlement keeps the demand assignment installed until requesting-cell
 //! state is authoritative.
-14
 use super::reuse::{H1AvailabilityOutcome, ReuseCandidate, ReuseId};
 use super::h1::{H1Candidate, H1MatchId, H1SupplyOutcome};
 use super::{
     AdmissionAction, DeliveryAckResult, DeliveryId, DemandId, DemandSnapshot, OriginAdmission,
     Permit,
     AdmissionAction, CapacityLease, CapacityPermit, DemandAssignment, DemandAssignmentOutcome,
     DemandId, DemandSnapshot, OriginAdmission,
 };
 use crate::client::pool::cell::h1::H1Selection;
 use crate::client::pool::cell::{
     AcquisitionEvent, AcquisitionResult, EstablishmentPermit, OriginCell,
     AcquisitionOutcome, AcquisitionStep, EstablishmentPermit, OriginCell,
 };
 use crate::client::pool::partition::PartitionId;
 use crate::sync::Arc;
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use std::fmt;
-27
 /// Capacity or a borrowed HTTP/1 sender crossing to waiting demand.
 enum AcquisitionPayload {
 enum DeliveryPayload {
     /// Permit removed from admission but not yet represented by a lease.
     Capacity(Permit),
     Capacity(CapacityPermit),
     /// Provisional sender whose owning-cell reservation must revalidate.
     BorrowedH1 {
         /// Reuse operation that selected this sender.
         reuse_id: ReuseId,
         /// Retained H1 match that selected this sender.
         match_id: H1MatchId,
         /// Cell that owns the sender and local reservation.
         connection_partition: PartitionId,
         supplier: PartitionId,
         /// Sender owner with cancellation fallback.
         candidate: ReuseCandidate,
         candidate: H1Candidate,
     },
-41
+}
-42
 /// Payload after every connection cell-side fallible transition has completed.
 enum MaterializedPayload {
     /// Establishment authority ready to move into the requesting cell.
     Capacity(EstablishmentPermit),
     /// Selected sender ready to move into the requesting cell.
     BorrowedH1 {
         /// Reuse operation that selected this sender.
         reuse_id: ReuseId,
         /// Cell that owns the installed connection record.
         connection_partition: PartitionId,
         /// Dispatch-ready sender ownership.
         selection: H1Selection,
     },
--
+}
--
 /// One acquisition payload and its admission-owned delivery fence.
 /// One acquisition payload and its admission-owned demand assignment.
 ///
 /// Dropping an undelivered guard refunnels its payload before making the
 /// requesting cell's demand schedulable again. Once committed, [`DeliveryAck`]
 /// owns fence completion and the requesting cell owns the acquisition event.
 /// requesting cell's demand schedulable again. Once committed,
 /// [`DeliverySettlement`] owns assignment settlement and the requesting cell
 /// owns the acquisition step.
 pub(in crate::client::pool) struct DeliveryGuard {
     /// Admission authority that owns the delivery fence and fallback.
     origin: Arc<OriginAdmission>,
     /// Never-reused identity of this one crossing.
     delivery: DeliveryId,
     /// Cell selected while admission was locked.
     requesting_partition: PartitionId,
     /// Exact requesting-cell demand generation that must revalidate the
     /// payload.
     demand: DemandId,
     /// Payload ownership before materialization, after materialization, or after transfer.
     state: DeliveryGuardState,
     /// Admission authority that owns the assignment and fallback.
     admission: Arc<OriginAdmission>,
     /// Exact demand excluded from competing admission while the payload crosses.
     assignment: DemandAssignment,
     /// Payload ownership before resolution, after resolution, or after transfer.
     state: DeliveryState,
-56
+}
-57
 /// Ownership phase of one admission-to-cell delivery.
 /// Ownership phase of one admission-to-cell delivery crossing.
 ///
 /// ```text
 /// Undelivered -- materialize connection-cell work ----------> Materialized
 /// Undelivered -- peer sender rejection or drop ------------------> admission fallback
 /// Materialized -- commit to requesting cell ----------------> Disarmed + DeliveryAck
 /// Materialized -- drop -------------------------------------> admission fallback
 /// Pending -- resolve connection-cell work ------------> Ready
 /// Pending -- peer sender rejection or drop -----------> admission fallback
 /// Ready   -- commit to requesting cell ---------------> Disarmed + DeliverySettlement
 /// Ready   -- drop ------------------------------------> admission fallback
 /// ```
 ///
 /// `Disarmed` owns no payload. After commit, `DeliveryAck` owns fence and
 /// connection-cell completion.
 enum DeliveryGuardState {
     /// Admission removed the payload, but connection cell-side work may still fail.
     Undelivered {
         /// Payload still owned by this fallback.
         payload: Option<AcquisitionPayload>,
         /// Demand acknowledgement used if the guard is dropped.
         on_drop: DeliveryAckResult,
     },
     /// Connection-owning-cell work completed; the requesting cell may reserve
     /// its waiter.
     Materialized {
         /// Requesting-cell-ready payload still owned by this fallback.
         payload: Option<MaterializedPayload>,
         /// Demand acknowledgement used if the guard is dropped.
         on_drop: DeliveryAckResult,
 /// `Disarmed` owns no payload. After commit, `DeliverySettlement` owns
 /// assignment and connection-cell completion.
 enum DeliveryState {
     /// Admission selected the payload, but connection cell-side work may fail.
     Pending(DeliveryPayload),
     /// Connection-owning-cell work completed; the exact acquisition step may
     /// now cross to the requesting cell.
     Ready {
         step: AcquisitionStep,
         settlement: DeliverySettlementKind,
     },
     /// Payload and fallback responsibility moved to requesting cell-owned state.
     Disarmed,
-80
+}
-81
 impl DeliveryGuard {
     /// Creates a delivery for one permit removed from admission.
     pub(super) fn capacity(
         origin: Arc<OriginAdmission>,
         delivery: DeliveryId,
         requesting_partition: PartitionId,
         demand: DemandId,
         permit: Permit,
         admission: Arc<OriginAdmission>,
         assignment: DemandAssignment,
         permit: CapacityPermit,
     ) -> Self {
         Self {
             origin,
             delivery,
             requesting_partition,
             demand,
             state: DeliveryGuardState::Undelivered {
                 payload: Some(AcquisitionPayload::Capacity(permit)),
                 on_drop: DeliveryAckResult::RetrySameResidence,
             },
             admission,
             assignment,
             state: DeliveryState::Pending(DeliveryPayload::Capacity(permit)),
-93
+        }
-94
+    }
-95
     /// Creates a delivery for one provisional sender selected for borrowing.
     pub(super) fn borrowed_h1(
         origin: Arc<OriginAdmission>,
         delivery: DeliveryId,
         requesting_partition: PartitionId,
         demand: DemandId,
         reuse_id: ReuseId,
         connection_partition: PartitionId,
         candidate: ReuseCandidate,
         admission: Arc<OriginAdmission>,
         assignment: DemandAssignment,
         match_id: H1MatchId,
         supplier: PartitionId,
         candidate: H1Candidate,
     ) -> Self {
         Self {
             origin,
             delivery,
             requesting_partition,
             demand,
             state: DeliveryGuardState::Undelivered {
                 payload: Some(AcquisitionPayload::BorrowedH1 {
                     reuse_id,
                     connection_partition,
                     candidate,
                 }),
                 on_drop: DeliveryAckResult::RetrySameResidence,
             },
             admission,
             assignment,
             state: DeliveryState::Pending(DeliveryPayload::BorrowedH1 {
                 match_id,
                 supplier,
                 candidate,
             }),
-112
+        }
-113
+    }
-114
     /// Returns the demand generation fenced by this delivery.
     /// Returns the demand generation owned by this delivery assignment.
     pub(in crate::client::pool) fn demand(&self) -> DemandId {
         self.demand
         self.assignment.demand
-118
+    }
-119
     /// Returns whether admission still recognizes this delivery fence.
     /// Returns whether admission still recognizes this delivery assignment.
     #[cfg(test)]
     pub(in crate::client::pool) fn is_current(&self) -> bool {
         self.origin
             .delivery_is_current(self.delivery, &self.requesting_partition, self.demand)
         self.admission.assignment_is_current(&self.assignment)
-124
+    }
-125
     /// Materializes owning-cell state and attempts one requesting-cell delivery.
     pub(super) fn deliver_once(mut self) -> Option<AdmissionAction> {
         if !self.materialize() {
     pub(super) fn deliver(mut self) -> Option<AdmissionAction> {
         if !self.resolve_payload() {
             return None;
-130
+        }
         match self.origin.cell(&self.requesting_partition) {
         match self.admission.cell(&self.assignment.requester) {
             Some(requesting_cell) => OriginCell::receive_delivery(&requesting_cell, self),
             None => self.reject(None),
             None => self.refuse(None),
-134
+        }
-135
+    }
-136
     /// Completes fallible owning-cell work before reserving requesting-cell state.
     fn materialize(&mut self) -> bool {
         let state = std::mem::replace(&mut self.state, DeliveryGuardState::Disarmed);
         let DeliveryGuardState::Undelivered {
             mut payload,
             on_drop,
         } = state
         else {
             unreachable!("delivery payload materialized more than once");
     fn resolve_payload(&mut self) -> bool {
         let state = std::mem::replace(&mut self.state, DeliveryState::Disarmed);
         let DeliveryState::Pending(payload) = state else {
             unreachable!("delivery payload resolved more than once");
         };
         let payload = payload
             .take()
             .expect("undelivered acquisition payload disappeared");
         let materialized = match payload {
             AcquisitionPayload::Capacity(permit) => {
                 MaterializedPayload::Capacity(EstablishmentPermit::bounded(
                     super::CapacityLease::new(self.origin.clone(), permit),
                 ))
--
+            }
             AcquisitionPayload::BorrowedH1 {
                 reuse_id,
                 connection_partition,
         let (step, settlement) = match payload {
             DeliveryPayload::Capacity(permit) => (
                 AcquisitionStep::StartEstablishment(EstablishmentPermit::bounded(
                     CapacityLease::new(self.admission.clone(), permit),
                 )),
                 DeliverySettlementKind::Capacity,
             ),
             DeliveryPayload::BorrowedH1 {
                 match_id,
                 supplier,
                 candidate,
             } => match candidate.commit() {
                 Ok(selection) => MaterializedPayload::BorrowedH1 {
                     reuse_id,
                     connection_partition,
                     selection,
                 },
                 Ok(selection) => {
                     let connection_id = selection.connection_id();
-+
+                    (
                         AcquisitionStep::Resolved(AcquisitionOutcome::H1(selection)),
                         DeliverySettlementKind::BorrowedH1 {
                             connection_id,
                             match_id,
                             supplier,
                         },
-+
+                    )
-+
+                }
                 Err(candidate) => {
                     drop(candidate);
                     let next = OriginAdmission::finish_delivery(
                         &self.origin,
                         self.delivery,
                         &self.requesting_partition,
                     let outcome = candidate.reject();
                     let next = OriginAdmission::settle_borrow_delivery(
                         &self.admission,
                         match_id,
                         &self.assignment,
                         DemandAssignmentOutcome::RetrySamePosition,
                         None,
                         on_drop,
                         Some(outcome),
                     );
                     OriginAdmission::drive(next);
                     OriginAdmission::run_action_chain(next);
                     return false;
-178
+                }
             },
         };
         self.state = DeliveryGuardState::Materialized {
             payload: Some(materialized),
             on_drop,
         };
         self.state = DeliveryState::Ready { step, settlement };
         true
-183
+    }
-184
     /// Materializes this guard before tests manually split the requesting cell
     /// reservation and installation transitions.
     #[cfg(test)]
     pub(in crate::client::pool) fn materialize_for_test(&mut self) -> bool {
         self.materialize()
     pub(in crate::client::pool) fn resolve_payload_for_test(&mut self) -> bool {
         self.resolve_payload()
-190
+    }
-191
     /// Moves the materialized payload into a requesting cell-owned event.
     pub(in crate::client::pool) fn commit(
     /// Moves the resolved payload into a requesting cell-owned acquisition step.
     pub(in crate::client::pool) fn into_step(
         mut self,
         successor: Option<DemandSnapshot>,
     ) -> (AcquisitionEvent, DeliveryAck) {
         let DeliveryGuardState::Materialized { payload, .. } = &mut self.state else {
             unreachable!("delivery committed before payload materialization");
     ) -> (AcquisitionStep, DeliverySettlement) {
         let state = std::mem::replace(&mut self.state, DeliveryState::Disarmed);
         let DeliveryState::Ready { step, settlement } = state else {
             unreachable!("delivery committed before payload resolution");
         };
         let payload = payload
             .take()
             .expect("materialized acquisition payload moved more than once");
         let (event, kind) = match payload {
             MaterializedPayload::Capacity(permit) => {
                 (AcquisitionEvent::Establish(permit), DeliveryKind::Capacity)
--
+            }
             MaterializedPayload::BorrowedH1 {
                 reuse_id,
                 connection_partition,
                 selection,
             } => {
                 let connection_id = selection.connection_id();
--
+                (
                     AcquisitionEvent::Complete(AcquisitionResult::H1(selection)),
                     DeliveryKind::BorrowedH1 {
                         connection_id,
                         reuse_id,
                         connection_partition,
                     },
--
+                )
--
+            }
         };
         self.state = DeliveryGuardState::Disarmed;
-201
+        (
             event,
             DeliveryAck {
                 origin: self.origin.clone(),
                 delivery: self.delivery,
                 requesting_partition: self.requesting_partition,
             step,
             DeliverySettlement {
                 admission: self.admission.clone(),
                 assignment: self.assignment.clone(),
                 successor,
                 kind: Some(kind),
                 kind: Some(settlement),
             },
-209
+        )
-210
+    }
-211
     /// Refunnels this delivery after requesting cell revalidation rejects it.
     pub(in crate::client::pool) fn reject(
     pub(in crate::client::pool) fn refuse(
         mut self,
         successor: Option<DemandSnapshot>,
     ) -> Option<AdmissionAction> {
         let state = std::mem::replace(&mut self.state, DeliveryGuardState::Disarmed);
         self.finish_state(state, DeliveryAckResult::Rejected { successor })
         let state = std::mem::replace(&mut self.state, DeliveryState::Disarmed);
         self.settle_state(state, DemandAssignmentOutcome::Refused { successor })
-219
+    }
-220
     /// Resolves payload fallback before closing the admission-owned fence.
     fn finish_state(
     /// Resolves payload fallback before settling the admission-owned assignment.
     fn settle_state(
         &self,
         state: DeliveryGuardState,
         result: DeliveryAckResult,
         state: DeliveryState,
         outcome: DemandAssignmentOutcome,
     ) -> Option<AdmissionAction> {
         match state {
             DeliveryGuardState::Undelivered { mut payload, .. } => {
                 let permit = match payload.take() {
                     Some(AcquisitionPayload::Capacity(permit)) => Some(permit),
                     Some(AcquisitionPayload::BorrowedH1 { candidate, .. }) => {
                         drop(candidate);
                         None
--
+                    }
                     None => None,
                 };
                 OriginAdmission::finish_delivery(
                     &self.origin,
                     self.delivery,
                     &self.requesting_partition,
                     permit,
                     result,
             DeliveryState::Pending(DeliveryPayload::Capacity(permit)) => {
                 OriginAdmission::settle_delivery(
                     &self.admission,
                     &self.assignment,
                     Some(permit),
                     outcome,
-+
+                )
-+
+            }
             DeliveryState::Pending(DeliveryPayload::BorrowedH1 {
                 match_id,
                 candidate,
                 ..
             }) => {
                 let supply = candidate.reject();
                 OriginAdmission::settle_borrow_delivery(
                     &self.admission,
                     match_id,
                     &self.assignment,
                     outcome,
                     None,
                     Some(supply),
-249
+                )
-250
+            }
             DeliveryGuardState::Materialized { mut payload, .. } => match payload.take() {
                 Some(MaterializedPayload::Capacity(permit)) => {
                     drop(permit);
                     OriginAdmission::finish_delivery(
                         &self.origin,
                         self.delivery,
                         &self.requesting_partition,
             DeliveryState::Ready { step, settlement } => match settlement {
                 DeliverySettlementKind::Capacity => {
                     drop(step);
                     OriginAdmission::settle_delivery(
                         &self.admission,
                         &self.assignment,
                         None,
                         result,
                         outcome,
-259
+                    )
-260
+                }
                 Some(MaterializedPayload::BorrowedH1 {
                     reuse_id,
                     connection_partition,
                     selection,
                 }) => {
                     let owning_cell = self.origin.cell(&connection_partition);
                     drop(selection);
                     let outcome = match owning_cell {
                         Some(owning_cell) => H1AvailabilityOutcome::reported(
                             connection_partition,
                             owning_cell.cancel_h1_reuse(reuse_id),
                 DeliverySettlementKind::BorrowedH1 {
                     match_id, supplier, ..
                 } => {
                     let supplier_cell = self.admission.cell(&supplier);
                     drop(step);
                     let supply = match supplier_cell {
                         Some(supplier_cell) => H1SupplyOutcome::supplier_live(
                             supplier,
                             supplier_cell.cancel_h1_reservation(match_id),
                         ),
                         None => H1AvailabilityOutcome::expired(connection_partition),
                         None => H1SupplyOutcome::supplier_expired(supplier),
                     };
                     OriginAdmission::finish_borrow_delivery(
                         &self.origin,
                         reuse_id,
                         self.delivery,
                         &self.requesting_partition,
                         result,
                     OriginAdmission::settle_borrow_delivery(
                         &self.admission,
                         match_id,
                         &self.assignment,
                         outcome,
                         None,
                         Some(outcome),
                         Some(supply),
-280
+                    )
-281
+                }
--
                 None => None,
             },
             DeliveryGuardState::Disarmed => None,
             DeliveryState::Disarmed => None,
-284
+        }
-285
+    }
-286
+}
-287
 impl fmt::Debug for DeliveryGuard {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("DeliveryGuard")
             .field("delivery", &self.delivery)
             .field("requesting_partition", &self.requesting_partition)
             .field("demand", &self.demand)
             .field("assignment", &self.assignment)
             .finish_non_exhaustive()
-293
+    }
-294
+}
-295
 impl Drop for DeliveryGuard {
     fn drop(&mut self) {
         let state = std::mem::replace(&mut self.state, DeliveryGuardState::Disarmed);
         let result = match &state {
             DeliveryGuardState::Undelivered { on_drop, .. }
             | DeliveryGuardState::Materialized { on_drop, .. } => on_drop.clone(),
             DeliveryGuardState::Disarmed => return,
         };
         let next = self.finish_state(state, result);
         OriginAdmission::drive(next);
         let state = std::mem::replace(&mut self.state, DeliveryState::Disarmed);
         if matches!(state, DeliveryState::Disarmed) {
             return;
-+
+        }
         let next = self.settle_state(state, DemandAssignmentOutcome::RetrySamePosition);
         OriginAdmission::run_action_chain(next);
-304
+    }
-305
+}
-306
 /// Acknowledgement after an event becomes owned by the requesting cell.
 /// Assignment settlement after a step becomes owned by the requesting cell.
 ///
 /// Explicit rejection returns the event before closing the delivery fence.
 /// Once the requesting cell owns the event, dropping this value acknowledges
 /// success and completes any borrowed connection's reuse operation.
 pub(in crate::client::pool) struct DeliveryAck {
     /// Admission authority that owns the outstanding fence.
     origin: Arc<OriginAdmission>,
     /// Delivery fence closed by this acknowledgement.
     delivery: DeliveryId,
     /// Cell that now owns or rejected the acquisition event.
     requesting_partition: PartitionId,
 /// Explicit refusal returns the step before settling the assignment. Once the
 /// requesting cell owns the step, dropping this value accepts the assignment
 /// and completes any borrowed connection's H1 match.
 pub(in crate::client::pool) struct DeliverySettlement {
     /// Admission authority that owns the outstanding assignment.
     admission: Arc<OriginAdmission>,
     /// Exact assignment settled by this value.
     assignment: DemandAssignment,
     /// Successor demand to publish after the old generation finishes.
     successor: Option<DemandSnapshot>,
     /// Payload-specific connection cell completion still owed by this guard.
     kind: Option<DeliveryKind>,
     kind: Option<DeliverySettlementKind>,
-321
+}
-322
 /// Terminal work required after the requesting cell takes ownership.
 enum DeliveryKind {
     /// Only the admission delivery fence remains to acknowledge.
 enum DeliverySettlementKind {
     /// Only the admission demand assignment remains to settle.
     Capacity,
     /// The connection cell must learn whether sender transfer succeeded.
     BorrowedH1 {
         /// Connection whose sender crossed to the requesting cell.
         connection_id: ConnectionId,
         /// Reuse operation completed by this acknowledgement.
         reuse_id: ReuseId,
         /// Cell whose local reuse reservation must complete.
         connection_partition: PartitionId,
         /// Retained H1 match completed by this settlement.
         match_id: H1MatchId,
         /// Cell whose local peer reservation must complete.
         supplier: PartitionId,
     },
-336
+}
-337
 impl DeliveryAck {
 impl DeliverySettlement {
     /// Avoids republishing a successor already covered by visible HTTP/2 state.
     ///
     /// Capacity or HTTP/1 delivery may reveal another waiter after the
     /// requesting cell already gained a local generation or peer route. That
     /// visible HTTP/2 state can serve an H2-compatible successor directly. If
     /// it later closes, its close path publishes the cell's current demand.
     pub(in crate::client::pool) fn suppress_h2_successor(&mut self) {
         if self
             .successor
             .as_ref()
             .is_some_and(DemandSnapshot::accepts_h2)
-350
+        {
             self.successor = None;
-352
+        }
-353
+    }
-354
     /// Acknowledges that requesting cell state accepted the acquisition event.
     /// Records that requesting cell state accepted the acquisition step.
     pub(in crate::client::pool) fn accept(mut self) -> Option<AdmissionAction> {
         let kind = self
             .kind
             .take()
             .expect("delivery acknowledgement completed more than once");
             .expect("delivery settlement completed more than once");
         let successor = self.successor.take();
         self.finish(kind, DeliveryAckResult::Accepted { successor }, None)
         self.settle(kind, DemandAssignmentOutcome::Accepted { successor }, None)
-363
+    }
-364
     /// Refunnels events rejected after requesting cell reservation.
     pub(in crate::client::pool) fn reject(
     /// Refunnels steps refused after requesting cell reservation.
     pub(in crate::client::pool) fn refuse(
         mut self,
         returned_events: [Option<AcquisitionEvent>; 2],
         returned_steps: [Option<AcquisitionStep>; 2],
     ) -> Option<AdmissionAction> {
         let kind = self
             .kind
             .take()
             .expect("delivery acknowledgement completed more than once");
             .expect("delivery settlement completed more than once");
         let successor = self.successor.take();
         self.finish(
         self.settle(
             kind,
             DeliveryAckResult::Rejected { successor },
             Some(returned_events),
             DemandAssignmentOutcome::Refused { successor },
             Some(returned_steps),
-379
+        )
-380
+    }
-381
     /// Completes the payload-specific reuse operation and delivery fence.
     fn finish(
     /// Completes payload-specific H1 work and settles the demand assignment.
     fn settle(
         &self,
         kind: DeliveryKind,
         result: DeliveryAckResult,
         returned_events: Option<[Option<AcquisitionEvent>; 2]>,
         kind: DeliverySettlementKind,
         outcome: DemandAssignmentOutcome,
         returned_steps: Option<[Option<AcquisitionStep>; 2]>,
     ) -> Option<AdmissionAction> {
         match kind {
             DeliveryKind::Capacity => {
                 drop(returned_events);
                 OriginAdmission::finish_delivery(
                     &self.origin,
                     self.delivery,
                     &self.requesting_partition,
                     None,
                     result,
--
+                )
             DeliverySettlementKind::Capacity => {
                 drop(returned_steps);
                 OriginAdmission::settle_delivery(&self.admission, &self.assignment, None, outcome)
-393
+            }
             DeliveryKind::BorrowedH1 {
             DeliverySettlementKind::BorrowedH1 {
                 connection_id,
                 reuse_id,
                 connection_partition,
                 match_id,
                 supplier,
             } => {
                 let rejected = returned_events.is_some();
                 let owning_cell = rejected
                     .then(|| self.origin.cell(&connection_partition))
                     .flatten();
                 drop(returned_events);
                 let rejected_outcome = rejected.then(|| match owning_cell {
                     Some(owning_cell) => H1AvailabilityOutcome::reported(
                         connection_partition,
                         owning_cell.cancel_h1_reuse(reuse_id),
                 let refused = returned_steps.is_some();
                 let supplier_cell = refused.then(|| self.admission.cell(&supplier)).flatten();
                 drop(returned_steps);
                 let refused_outcome = refused.then(|| match supplier_cell {
                     Some(supplier_cell) => H1SupplyOutcome::supplier_live(
                         supplier,
                         supplier_cell.cancel_h1_reservation(match_id),
                     ),
                     None => H1AvailabilityOutcome::expired(connection_partition),
                     None => H1SupplyOutcome::supplier_expired(supplier),
                 });
                 let transferred_connection_cell = (!rejected).then_some(connection_partition);
                 let action = OriginAdmission::finish_borrow_delivery(
                     &self.origin,
                     reuse_id,
                     self.delivery,
                     &self.requesting_partition,
                     result,
                     transferred_connection_cell,
                     rejected_outcome,
                 let transferred_supplier = (!refused).then_some(supplier);
                 let action = OriginAdmission::settle_borrow_delivery(
                     &self.admission,
                     match_id,
                     &self.assignment,
                     outcome,
                     transferred_supplier,
                     refused_outcome,
                 );
                 if !rejected {
                 if !refused {
                     tracing::trace!(
                         connection_id = %connection_id,
                         request_partition = ?self.requesting_partition,
                         connection_partition = ?connection_partition,
                         origin_scheme = %self.origin.origin().scheme(),
                         origin_host = self.origin.origin().host(),
                         origin_port = ?self.origin.origin().port(),
                         request_partition = ?self.assignment.requester,
                         connection_partition = ?supplier,
                         origin_scheme = %self.admission.origin().scheme(),
                         origin_host = self.admission.origin().host(),
                         origin_port = ?self.admission.origin().port(),
                         "HTTP/1 connection borrowed for peer demand"
                     );
-428
+                }
                 action
-430
+            }
-431
+        }
-432
+    }
-433
+}
-434
 impl fmt::Debug for DeliveryAck {
 impl fmt::Debug for DeliverySettlement {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("DeliveryAck")
             .field("delivery", &self.delivery)
             .field("requesting_partition", &self.requesting_partition)
         f.debug_struct("DeliverySettlement")
             .field("assignment", &self.assignment)
             .finish_non_exhaustive()
-440
+    }
-441
+}
-442
 impl Drop for DeliveryAck {
 impl Drop for DeliverySettlement {
     fn drop(&mut self) {
         let Some(kind) = self.kind.take() else {
             return;
         };
         let successor = self.successor.take();
         let next = self.finish(kind, DeliveryAckResult::Accepted { successor }, None);
         OriginAdmission::drive(next);
         let next = self.settle(kind, DemandAssignmentOutcome::Accepted { successor }, None);
         OriginAdmission::run_action_chain(next);
-+
+    }
-+
+}
-+
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::super::{ProtocolRequirement, SnapshotVersion};
     use super::*;
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::EligibilityGroup;
     use std::num::NonZeroUsize;
-+
     fn cell(origin: &Arc<OriginAdmission>, partition: usize) -> Arc<OriginCell> {
         let cell = Arc::new(OriginCell::new(
             PartitionId::from_index(partition),
             OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None).unwrap(),
             EligibilityGroup::Pool,
             Some(origin.clone()),
             None,
         ));
         OriginAdmission::register_cell(origin, cell)
-+
+    }
-+
     fn demand(id: u64) -> DemandSnapshot {
         DemandSnapshot::active(
             DemandId::from_u64(id),
             SnapshotVersion::INITIAL,
             ProtocolRequirement::H1Compatible,
             EligibilityGroup::Pool,
-+
+        )
-+
+    }
-+
     #[test]
     fn assignment_currency_includes_the_demand_id() {
         let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let requesting_partition = cell(&origin, 1);
         let delivery = OriginAdmission::submit_without_running(
             &origin,
             requesting_partition.id().partition(),
             demand(1),
-+
+        )
         .unwrap();
         assert!(delivery.is_current());
-+
         origin
             .state
             .lock()
             .apply_demand_snapshot(requesting_partition.id().partition(), demand(2));
         assert!(!delivery.is_current());
         delivery.refuse(None);
-+
+    }
-+
     #[test]
     fn stale_successor_cannot_leave_active_demand_idle() {
         let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let requesting_partition = cell(&origin, 1);
         let delivery = OriginAdmission::submit_without_running(
             &origin,
             requesting_partition.id().partition(),
             demand(1),
-+
+        )
         .unwrap();
         origin.state.lock().apply_demand_snapshot(
             requesting_partition.id().partition(),
             DemandSnapshot::active(
                 DemandId::from_u64(1),
                 SnapshotVersion::INITIAL.next(),
                 ProtocolRequirement::H1Compatible,
                 EligibilityGroup::Pool,
             ),
         );
         delivery.refuse(Some(demand(1)));
-+
         assert_eq!(1, origin.probe().available);
         assert_eq!(0, origin.probe().ordered);
-+
+    }
-+
     #[test]
     fn dropped_delivery_refunnels_capacity_and_preserves_order() {
         let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let first = cell(&origin, 1);
         let second = cell(&origin, 2);
         let (first_waiter, first_demand) =
             first.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
         let (second_waiter, second_demand) =
             second.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
         let delivery =
             OriginAdmission::submit_without_running(&origin, first.id().partition(), first_demand)
                 .unwrap();
-+
+        {
             let mut state = origin.state.lock();
             state.apply_demand_snapshot(second.id().partition(), second_demand);
-+
+        }
         drop(delivery);
-+
         let first_lease = OriginCell::take_ready_lease(&first, first_waiter)
             .expect("dropped delivery did not retry the original head");
         assert!(OriginCell::take_ready_lease(&second, second_waiter).is_none());
         assert_eq!(1, origin.probe().ordered);
-+
         drop(first_lease);
         let second_lease = OriginCell::take_ready_lease(&second, second_waiter)
             .expect("younger demand did not run after the original head");
         drop(second_lease);
-+
+    }
-+
     #[test]
     fn expired_requesting_cell_refunnels_capacity() {
         let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
         let requesting_partition = cell(&origin, 1);
         let requesting_cell_id = requesting_partition.id().partition();
         let (_waiter, snapshot) =
             requesting_partition.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
         let delivery =
             OriginAdmission::submit_without_running(&origin, requesting_cell_id, snapshot).unwrap();
-+
         drop(requesting_partition);
         assert!(origin.cell(&requesting_cell_id).is_none());
         OriginAdmission::run_action_chain(Some(AdmissionAction::Deliver(delivery)));
-+
         let probe = origin.probe();
         assert_eq!(1, probe.available);
         assert_eq!(0, probe.assigned);
         assert_eq!(0, probe.ordered);
-574
+    }
-575
+}

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

@@ -1,1 +1104,1196 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Versioned cross-cell demand scheduling for one bounded origin.
 //!
 //! Each cell publishes at most one active demand: its acquisition queue's

 //! current head. That demand may be satisfied either by origin-wide connection
 //! capacity or by an existing HTTP/2 connection in its eligibility group.
 //! These resources require different orderings.
 //!
 //! For example, the origin order may contain `A(group X), B(group Y)` while
 //! only group Y has a reusable HTTP/2 connection. A remains first for the next
 //! available connection permit, while the group-Y connection can serve B
 //! without consuming that permit or delaying A. The group order finds B
 //! directly instead of scanning every partition.
 //!
 //! The same demand occupies both orders because capacity and HTTP/2 publication
 //! can race to satisfy it. Reserving either position retains a fence in both
 //! orders until the requesting cell accepts or rejects the handoff. The other
 //! resource therefore cannot serve the same demand while the first handoff is
 //! running outside the admission lock.
--
 use super::{
     DeliveryAckResult, DeliveryId, DemandId, DemandSnapshot, DemandState, IntrusiveLinks,
     IntrusiveOrder, Permit, ProtocolRequirement,
 };
 //! The same demand occupies both orders because capacity delivery and HTTP/2
 //! route installation can race to satisfy it. Assigning either position keeps
 //! both positions attached until the requesting cell accepts or refuses the
 //! handoff. The other resource cannot select the same demand while the first
 //! handoff runs outside the admission lock.
-+
 use super::{CapacityPermit, IntrusiveLinks, IntrusiveOrder};
 use crate::client::pool::partition::{EligibilityGroup, PartitionId};
 use std::collections::HashMap;
-28
 /// Protocol capability required by the head waiter in a cell.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(in crate::client::pool) enum ProtocolRequirement {
     /// The waiter requires HTTP/1 wire semantics.
     H1Required,
     /// The waiter may dispatch over HTTP/1 or HTTP/2.
     H1Compatible,
     /// The waiter requires HTTP/2.
     H2Required,
-+
+}
-+
 impl ProtocolRequirement {
     pub(in crate::client::pool) fn accepts_h1(self) -> bool {
         self != Self::H2Required
-+
+    }
-+
     pub(in crate::client::pool) fn accepts_h2(self) -> bool {
         self != Self::H1Required
-+
+    }
-+
+}
-+
 /// Identity of one cell-local demand generation.
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub(in crate::client::pool) struct DemandId(u64);
-+
 impl DemandId {
     pub(in crate::client::pool) const fn from_u64(value: u64) -> Self {
         Self(value)
-+
+    }
-+
+}
-+
 /// Strict ordering of complete snapshots within one [`DemandId`].
 #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
 pub(in crate::client::pool) struct SnapshotVersion(u64);
-+
 impl SnapshotVersion {
     pub(in crate::client::pool) const INITIAL: Self = Self(0);
-+
     pub(in crate::client::pool) fn next(self) -> Self {
         Self(
             self.0
                 .checked_add(1)
                 .expect("demand snapshot version exhausted"),
-+
+        )
-+
+    }
-+
+}
-+
 /// Complete state submitted for one demand identity.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(in crate::client::pool) enum DemandState {
     Active {
         requirement: ProtocolRequirement,
         eligibility_group: EligibilityGroup,
     },
     Inactive,
-+
+}
-+
 /// Versioned replacement state for one cell's current demand generation.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(in crate::client::pool) struct DemandSnapshot {
     pub(super) id: DemandId,
     pub(super) version: SnapshotVersion,
     pub(super) state: DemandState,
-+
+}
-+
 impl DemandSnapshot {
     pub(in crate::client::pool) fn active(
         id: DemandId,
         version: SnapshotVersion,
         requirement: ProtocolRequirement,
         eligibility_group: EligibilityGroup,
     ) -> Self {
         Self {
             id,
             version,
             state: DemandState::Active {
                 requirement,
                 eligibility_group,
             },
-+
+        }
-+
+    }
-+
     pub(in crate::client::pool) fn accepts_h2(&self) -> bool {
         matches!(
             self.state,
             DemandState::Active { requirement, .. } if requirement.accepts_h2()
-+
+        )
-+
+    }
-+
     pub(in crate::client::pool) fn inactive(id: DemandId, version: SnapshotVersion) -> Self {
         Self {
             id,
             version,
             state: DemandState::Inactive,
-+
+        }
-+
+    }
-+
     #[cfg(test)]
     pub(in crate::client::pool) fn id_for_test(&self) -> DemandId {
         self.id
-+
+    }
-+
     pub(in crate::client::pool) fn is_active(&self) -> bool {
         matches!(self.state, DemandState::Active { .. })
-+
+    }
-+
     fn is_newer_than(&self, current: &Self) -> bool {
         self.id > current.id || (self.id == current.id && self.version > current.version)
-+
+    }
-+
+}
-+
 /// Never-reused identity of one resource-to-demand assignment.
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 pub(super) struct DemandAssignmentId(pub(super) u64);
-+
 /// Exact demand temporarily assigned while a resource crosses lock domains.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(super) struct DemandAssignment {
     pub(super) id: DemandAssignmentId,
     pub(super) requester: PartitionId,
     pub(super) demand: DemandId,
-+
+}
-+
 /// Admission's settlement of one detached demand assignment.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(super) enum DemandAssignmentOutcome {
     Accepted { successor: Option<DemandSnapshot> },
     RetrySamePosition,
     Refused { successor: Option<DemandSnapshot> },
-+
+}
-+
 /// Cross-cell demand records and their scheduling orders.
 ///
 /// At every completed transition:
 ///
 /// - `records` owns the newest snapshot for every retained cell.
 /// - the origin order and the record's eligibility-group order contain exactly
 ///   the `Queued` and `Delivering` records;
 /// - links for both views live inside those ordered residence variants;
 /// - an origin delivery fences the origin head; and
 /// - an HTTP/2 publication fences its eligibility-group head.
 ///   the `Queued` and `PendingAssignment` records;
 /// - links for both orders live inside those schedule states;
 /// - an origin assignment owns the origin head; and
 /// - an HTTP/2 route assignment owns its eligibility-group head.
 ///
 /// Admission coordinates capacity extraction with this schedule while holding
 /// the same origin lock.
 #[derive(Debug, Default)]
 pub(super) struct DemandSchedule {
     /// Latest demand and scheduling residence for each retained cell.
     records: HashMap<PartitionId, DemandRecord>,
     /// Origin-wide order used by one-to-one capacity and HTTP/1 reuse.
     origin_order: IntrusiveOrder<PartitionId>,
     /// Reusable-HTTP/2 demand order for each connection-reuse group.
     group_orders: HashMap<EligibilityGroup, IntrusiveOrder<PartitionId>>,
-181
+}
-182
 /// Complete origin-order head used to choose one HTTP/1 reuse action.
 #[derive(Clone, Debug)]
 pub(super) struct QueuedDemand {
     /// Cell whose oldest waiter owns this demand generation.
     pub(super) requesting_partition: PartitionId,
     pub(super) requester: PartitionId,
     /// Demand identity revalidated at each crossing.
     pub(super) demand: DemandId,
     /// Protocol capability required by the requesting cell waiter.
     pub(super) requirement: ProtocolRequirement,
     /// Connection-owning cells whose H1 senders may satisfy this demand.
     pub(super) eligibility_group: EligibilityGroup,
-194
+}
-195
 /// Latest snapshot and scheduling residence for one stable cell.
 #[derive(Debug)]
 struct DemandRecord {
     /// Newest complete publication observed for the cell.
     pub(super) latest: DemandSnapshot,
     /// Stable group retained while an inactive replacement crosses a fence.
     group: Option<EligibilityGroup>,
     /// Newest complete snapshot observed for the cell.
     latest: DemandSnapshot,
     /// Stable group retained while an inactive replacement crosses an assignment.
     eligibility_group: Option<EligibilityGroup>,
     /// Scheduling residence, including links while ordered.
     residence: DemandResidence,
     schedule_state: DemandScheduleState,
-205
+}
-206
 /// Links retained by one demand in both scheduling views.
 #[derive(Clone, Debug)]
 struct DemandLinks {
     /// Position in origin-wide capacity and HTTP/1 order.
     origin: IntrusiveLinks<PartitionId>,
     /// Position in the demand's all-protocol eligibility-group order.
     group: IntrusiveLinks<PartitionId>,
-214
+}
-215
 /// Order whose head is fenced by one delivery crossing.
 /// Order whose head is held by one detached demand assignment.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum DeliveryView {
 enum DemandOrder {
     /// Capacity or HTTP/1 delivery selected from origin order.
     Origin,
     /// HTTP/2 publication selected from eligibility-group order.
     Group,
--
+}
--
 /// Selects one of a demand residence's intrusive link sets.
 #[derive(Clone, Copy)]
 enum DemandOrderView {
     /// Origin-wide capacity and HTTP/1 order.
     Origin,
     /// Eligibility-group all-protocol order.
     /// HTTP/2 route selected from eligibility-group order.
     Group,
-223
+}
-224
 /// Admission residence for one partition's latest demand.
 /// Scheduling state for one partition's latest demand.
 ///
 /// ```text
 /// Idle -- active publication ------------------------------> Queued
 /// Queued -- reserve head ----------------------------------> Delivering
 /// Queued -- inactive or replacement publication ----------> Idle
 /// Delivering -- retry unchanged demand --------------------> Queued
 /// Delivering -- accept, reject, or replacement retry ------> Idle
 /// Idle -- active successor remains ------------------------> Queued
 /// Unscheduled -- active snapshot --------------------------> Queued
 /// Queued -- prepare assignment ----------------------------> PendingAssignment
 /// Queued -- inactive or replacement snapshot -------------> Unscheduled
 /// PendingAssignment -- retry unchanged demand -------------> Queued
 /// PendingAssignment -- accept/refuse/replacement retry ----> Unscheduled
 /// Unscheduled -- active successor remains -----------------> Queued
 /// ```
 #[derive(Clone, Debug)]
 enum DemandResidence {
 enum DemandScheduleState {
     /// The cell has no demand represented in scheduling.
     Idle,
     Unscheduled,
     /// The demand is waiting in origin order.
     Queued {
         /// Demand generation represented by this residence.
         demand: DemandId,
         /// Origin-wide and eligibility-group scheduling links.
         links: DemandLinks,
     },
     /// One delivery or publication guard fences this demand.
     Delivering {
         /// Demand generation fenced at one scheduling head.
         demand: DemandId,
         /// Snapshot version current when admission created the fence.
     /// One delivery or route guard owns this demand assignment.
     PendingAssignment {
         /// Exact assignment allowed to settle this state.
         assignment: DemandAssignment,
         /// Snapshot version current when admission created the assignment.
         version: super::SnapshotVersion,
         /// Delivery allowed to complete this fence.
         delivery: DeliveryId,
         /// Scheduling view whose head owns this crossing.
         /// Scheduling order whose head selected this assignment.
         #[cfg_attr(
             not(debug_assertions),
             allow(
                 dead_code,
                 reason = "the scheduling view is retained for debug invariant checks"
-258
+            )
         )]
         view: DeliveryView,
         selected_order: DemandOrder,
         /// Both scheduling positions retained until acknowledgement.
         links: DemandLinks,
     },
-264
+}
-265
 impl DemandResidence {
 impl DemandScheduleState {
     #[cfg(any(debug_assertions, test))]
     /// Borrows one view's links while this record is ordered.
     fn links(&self, view: DemandOrderView) -> Option<&IntrusiveLinks<PartitionId>> {
     fn links(&self, order: DemandOrder) -> Option<&IntrusiveLinks<PartitionId>> {
         let links = match self {
             Self::Idle => return None,
             Self::Queued { links, .. } | Self::Delivering { links, .. } => links,
             Self::Unscheduled => return None,
             Self::Queued { links, .. } | Self::PendingAssignment { links, .. } => links,
         };
         match view {
             DemandOrderView::Origin => Some(&links.origin),
             DemandOrderView::Group => Some(&links.group),
         match order {
             DemandOrder::Origin => Some(&links.origin),
             DemandOrder::Group => Some(&links.group),
-277
+        }
-278
+    }
-279
     /// Mutably borrows one view's links from an ordered record.
     ///
     /// # Panics
     ///
     /// Panics when called for an idle record.
     fn links_mut(&mut self, view: DemandOrderView) -> &mut IntrusiveLinks<PartitionId> {
     fn links_mut(&mut self, order: DemandOrder) -> &mut IntrusiveLinks<PartitionId> {
         let links = match self {
             Self::Idle => panic!("idle demand has no scheduling links"),
             Self::Queued { links, .. } | Self::Delivering { links, .. } => links,
             Self::Unscheduled => panic!("unscheduled demand has no scheduling links"),
             Self::Queued { links, .. } | Self::PendingAssignment { links, .. } => links,
         };
         match view {
             DemandOrderView::Origin => &mut links.origin,
             DemandOrderView::Group => &mut links.group,
         match order {
             DemandOrder::Origin => &mut links.origin,
             DemandOrder::Group => &mut links.group,
-293
+        }
-294
+    }
-295
     /// Detaches both link sets while moving a record out of scheduling.
     fn into_links(self) -> Option<DemandLinks> {
         match self {
             Self::Idle => None,
             Self::Queued { links, .. } | Self::Delivering { links, .. } => Some(links),
             Self::Unscheduled => None,
             Self::Queued { links, .. } | Self::PendingAssignment { links, .. } => Some(links),
-301
+        }
-302
+    }
-303
+}
-304
 /// Demand reserved at the order head for one capacity delivery.
 pub(super) struct ScheduledDemand {
     /// Selected destination cell.
     pub(super) requesting_partition: PartitionId,
     /// Demand identity current when the crossing was prepared.
     pub(super) demand: DemandId,
--
+}
--
 /// Connection capacity extracted under admission lock before crossing to a cell.
 #[derive(Debug)]
 pub(super) struct PreparedCapacityDelivery {
     /// Permit transferred into the delivery guard.
     pub(super) permit: Permit,
     /// Fence identity allocated for this crossing.
     pub(super) delivery: DeliveryId,
     /// Selected destination cell.
     pub(super) requesting_partition: PartitionId,
     /// Demand identity current when the crossing was prepared.
     pub(super) demand: DemandId,
     pub(super) assignment: DemandAssignment,
     pub(super) permit: CapacityPermit,
-310
+}
-311
 impl DemandSchedule {
     /// Applies a complete snapshot and updates its cell's scheduling residence.
     pub(super) fn publish(&mut self, requesting_partition: PartitionId, snapshot: DemandSnapshot) {
         if let Some(current) = self.records.get(&requesting_partition) {
     /// Applies a complete snapshot and updates its cell's scheduling state.
     pub(super) fn apply_snapshot(&mut self, requester: PartitionId, snapshot: DemandSnapshot) {
         if let Some(current) = self.records.get(&requester) {
             if !snapshot.is_newer_than(&current.latest) {
                 return;
-318
+            }
         } else {
             self.records.insert(
                 requesting_partition,
                 requester,
                 DemandRecord {
                     latest: snapshot.clone(),
                     group: None,
                     residence: DemandResidence::Idle,
                     eligibility_group: None,
                     schedule_state: DemandScheduleState::Unscheduled,
                 },
             );
-328
+        }
-329
         let should_remove = matches!(
             &self
                 .records
                 .get(&requesting_partition)
                 .expect("published demand record disappeared")
                 .residence,
             DemandResidence::Queued { demand, .. }
                 .get(&requester)
                 .expect("demand record disappeared")
                 .schedule_state,
             DemandScheduleState::Queued { demand, .. }
                 if *demand != snapshot.id || !snapshot.is_active()
         );
         if should_remove {
             self.remove_from_order(&requesting_partition);
             self.remove_from_order(&requester);
-341
+        }
-342
         if let DemandState::Active {
             eligibility_group, ..
         } = &snapshot.state
-346
+        {
             self.records
                 .get_mut(&requesting_partition)
                 .expect("published demand record disappeared")
                 .group = Some(eligibility_group.clone());
                 .get_mut(&requester)
                 .expect("demand record disappeared")
                 .eligibility_group = Some(eligibility_group.clone());
-351
+        }
         self.records
             .get_mut(&requesting_partition)
             .expect("published demand record disappeared")
             .get_mut(&requester)
             .expect("demand record disappeared")
             .latest = snapshot;
-356
         let record = self
             .records
             .get(&requesting_partition)
             .expect("published demand record disappeared");
         if record.latest.is_active() && matches!(&record.residence, DemandResidence::Idle) {
             self.enqueue(requesting_partition);
             .get(&requester)
             .expect("demand record disappeared");
         if record.latest.is_active()
             && matches!(&record.schedule_state, DemandScheduleState::Unscheduled)
-+
+        {
             self.enqueue(requester);
-365
+        }
         self.assert_consistent();
-367
+    }
-368
     /// Appends an idle active demand to both scheduling orders.
     fn enqueue(&mut self, requesting_partition: PartitionId) {
     fn enqueue(&mut self, requester: PartitionId) {
         let group = self
             .group_for(&requesting_partition)
             .group_for(&requester)
             .expect("enqueued demand was inactive");
         let origin = self.origin_order.push_back(requesting_partition);
         let origin = self.origin_order.push_back(requester);
         let group_links = self
             .group_orders
             .entry(group)
             .or_default()
             .push_back(requesting_partition);
             .push_back(requester);
         if let Some(previous) = origin.previous {
             self.records
                 .get_mut(&previous)
                 .expect("origin demand tail disappeared")
                 .residence
                 .links_mut(DemandOrderView::Origin)
                 .next = Some(requesting_partition);
                 .schedule_state
                 .links_mut(DemandOrder::Origin)
                 .next = Some(requester);
-387
+        }
         if let Some(previous) = group_links.previous {
             self.records
                 .get_mut(&previous)
                 .expect("group demand tail disappeared")
                 .residence
                 .links_mut(DemandOrderView::Group)
                 .next = Some(requesting_partition);
                 .schedule_state
                 .links_mut(DemandOrder::Group)
                 .next = Some(requester);
-395
+        }
-396
         let record = self
             .records
             .get_mut(&requesting_partition)
             .get_mut(&requester)
             .expect("queued demand record disappeared");
         debug_assert!(matches!(record.residence, DemandResidence::Idle));
         debug_assert!(matches!(
             record.schedule_state,
             DemandScheduleState::Unscheduled
         ));
         let demand = record.latest.id;
         record.residence = DemandResidence::Queued {
         record.schedule_state = DemandScheduleState::Queued {
             demand,
             links: DemandLinks {
                 origin,
                 group: group_links,
             },
         };
-413
+    }
-414
     /// Removes an ordered demand from both views and leaves its record idle.
     fn remove_from_order(&mut self, requesting_partition: &PartitionId) {
     fn remove_from_order(&mut self, requester: &PartitionId) {
         let group = self
             .group_for(requesting_partition)
             .group_for(requester)
             .expect("removed demand had no eligibility group");
         let residence = {
         let schedule_state = {
             let record = self
                 .records
                 .get_mut(requesting_partition)
                 .get_mut(requester)
                 .expect("removed demand record disappeared");
             std::mem::replace(&mut record.residence, DemandResidence::Idle)
             std::mem::replace(&mut record.schedule_state, DemandScheduleState::Unscheduled)
         };
         let links = residence
         let links = schedule_state
             .into_links()
             .expect("removed demand had no scheduling links");
-430
         if let Some(previous) = links.origin.previous {
             self.records
                 .get_mut(&previous)
                 .expect("previous origin demand disappeared")
                 .residence
                 .links_mut(DemandOrderView::Origin)
                 .schedule_state
                 .links_mut(DemandOrder::Origin)
                 .next = links.origin.next;
-438
+        }
         if let Some(next) = links.origin.next {
             self.records
                 .get_mut(&next)
                 .expect("next origin demand disappeared")
                 .residence
                 .links_mut(DemandOrderView::Origin)
                 .schedule_state
                 .links_mut(DemandOrder::Origin)
                 .previous = links.origin.previous;
-446
+        }
         self.origin_order
             .remove(*requesting_partition, links.origin);
         self.origin_order.remove(*requester, links.origin);
-448
         if let Some(previous) = links.group.previous {
             self.records
                 .get_mut(&previous)
                 .expect("previous group demand disappeared")
                 .residence
                 .links_mut(DemandOrderView::Group)
                 .schedule_state
                 .links_mut(DemandOrder::Group)
                 .next = links.group.next;
-456
+        }
         if let Some(next) = links.group.next {
             self.records
                 .get_mut(&next)
                 .expect("next group demand disappeared")
                 .residence
                 .links_mut(DemandOrderView::Group)
                 .schedule_state
                 .links_mut(DemandOrder::Group)
                 .previous = links.group.previous;
-464
+        }
         let order = self
             .group_orders
             .get_mut(&group)
             .expect("ordered demand lost its eligibility-group order");
         order.remove(*requesting_partition, links.group);
         order.remove(*requester, links.group);
         if order.len() == 0 {
             self.group_orders.remove(&group);
-472
+        }
-473
+    }
-474
     /// Returns whether the head can begin a new one-to-one delivery.
     pub(super) fn head_is_queued(&self) -> bool {
         let Some(head) = self.origin_order.head() else {
             return false;
         };
         matches!(
             &self
                 .records
                 .get(&head)
                 .expect("order head disappeared")
                 .residence,
             DemandResidence::Queued { .. }
                 .schedule_state,
             DemandScheduleState::Queued { .. }
-487
+        )
-488
+    }
-489
     /// Returns the complete origin-order head when it may begin reuse.
     pub(super) fn queued_head(&self) -> Option<QueuedDemand> {
         let head = self.origin_order.head()?;
         let record = self.records.get(&head).expect("order head disappeared");
         let DemandResidence::Queued { demand, .. } = &record.residence else {
         let DemandScheduleState::Queued { demand, .. } = &record.schedule_state else {
             return None;
         };
         let DemandState::Active {
             head: requirement,
             requirement,
             eligibility_group,
         } = &record.latest.state
         else {
             unreachable!("queued demand became inactive");
         };
         Some(QueuedDemand {
             requesting_partition: head,
             requester: head,
             demand: *demand,
             requirement: *requirement,
             eligibility_group: eligibility_group.clone(),
         })
-510
+    }
-511
     /// Returns one eligibility-group head when it may receive H2 visibility.
     pub(super) fn queued_group_head(&self, group: &EligibilityGroup) -> Option<QueuedDemand> {
         let head = self.group_orders.get(group)?.head()?;
         let record = self
             .records
             .get(&head)
             .expect("group demand head disappeared");
         let DemandResidence::Queued { demand, .. } = &record.residence else {
         let DemandScheduleState::Queued { demand, .. } = &record.schedule_state else {
             return None;
         };
         let DemandState::Active {
             head: requirement,
             requirement,
             eligibility_group,
         } = &record.latest.state
         else {
             unreachable!("queued group demand became inactive");
         };
         debug_assert_eq!(eligibility_group, group);
         Some(QueuedDemand {
             requesting_partition: head,
             requester: head,
             demand: *demand,
             requirement: *requirement,
             eligibility_group: eligibility_group.clone(),
         })
-536
+    }
-537
     /// Returns the latest eligibility group retained for one cell.
     pub(super) fn group_for(&self, requesting_partition: &PartitionId) -> Option<EligibilityGroup> {
     pub(super) fn group_for(&self, requester: &PartitionId) -> Option<EligibilityGroup> {
         self.records
             .get(requesting_partition)
             .and_then(|record| record.group.clone())
             .get(requester)
             .and_then(|record| record.eligibility_group.clone())
-543
+    }
-544
     /// Returns whether `requesting_partition` still has this demand queued for a new action.
     pub(super) fn is_current_queued(
         &self,
         requesting_partition: &PartitionId,
         demand: DemandId,
     ) -> bool {
         self.records
             .get(requesting_partition)
             .is_some_and(|record| {
                 record.latest.id == demand
                     && record.latest.is_active()
                     && matches!(
                         record.residence,
                         DemandResidence::Queued {
                             demand: current,
                             ..
                         } if current == demand
--
+                    )
             })
     pub(super) fn is_current_queued(&self, requester: &PartitionId, demand: DemandId) -> bool {
         self.records.get(requester).is_some_and(|record| {
             record.latest.id == demand
                 && record.latest.is_active()
                 && matches!(
                     record.schedule_state,
                     DemandScheduleState::Queued {
                         demand: current,
                         ..
                     } if current == demand
-+
+                )
         })
-558
+    }
-559
     /// Fences one reuse operation's demand while it remains the origin-order head.
     pub(super) fn reserve_reuse_demand(
     /// Assigns exact origin-head demand after an H1 match resolves.
     pub(super) fn prepare_h1_assignment(
         &mut self,
         requesting_partition: &PartitionId,
         requester: &PartitionId,
         demand: DemandId,
         delivery: DeliveryId,
     ) -> Option<ScheduledDemand> {
         if !self.is_current_queued(requesting_partition, demand) {
         assignment_id: DemandAssignmentId,
     ) -> Option<DemandAssignment> {
         if !self.is_current_queued(requester, demand) || self.origin_order.head()? != *requester {
             return None;
-569
+        }
         let head = self.origin_order.head()?;
         if head != *requesting_partition {
             return None;
--
+        }
         self.reserve_origin_head(delivery)
         self.prepare_assignment(*requester, assignment_id, DemandOrder::Origin)
-571
+    }
-572
     /// Fences one eligibility-group head for an H2 publication.
     pub(super) fn reserve_group_head(
     /// Assigns one exact eligibility-group head to an H2 route.
     pub(super) fn prepare_group_assignment(
         &mut self,
         group: &EligibilityGroup,
         requesting_partition: &PartitionId,
         eligibility_group: &EligibilityGroup,
         requester: &PartitionId,
         demand: DemandId,
         delivery: DeliveryId,
     ) -> Option<ScheduledDemand> {
         if !self.is_current_queued(requesting_partition, demand)
             || self.group_orders.get(group)?.head() != Some(*requesting_partition)
         assignment_id: DemandAssignmentId,
     ) -> Option<DemandAssignment> {
         if !self.is_current_queued(requester, demand)
             || self.group_orders.get(eligibility_group)?.head() != Some(*requester)
-583
+        {
             return None;
-585
+        }
         self.reserve(*requesting_partition, delivery, DeliveryView::Group)
         self.prepare_assignment(*requester, assignment_id, DemandOrder::Group)
-587
+    }
-588
     /// Changes the origin head into a delivery fence at the same positions.
     pub(super) fn reserve_origin_head(&mut self, delivery: DeliveryId) -> Option<ScheduledDemand> {
         let head = self.origin_order.head()?;
         self.reserve(head, delivery, DeliveryView::Origin)
     /// Assigns the origin-order head to one capacity delivery.
     pub(super) fn prepare_origin_assignment(
         &mut self,
         assignment_id: DemandAssignmentId,
     ) -> Option<DemandAssignment> {
         let requester = self.origin_order.head()?;
         self.prepare_assignment(requester, assignment_id, DemandOrder::Origin)
-596
+    }
-597
     /// Changes one queued demand into a delivery fence.
     fn reserve(
     fn prepare_assignment(
         &mut self,
         requesting_partition: PartitionId,
         delivery: DeliveryId,
         view: DeliveryView,
     ) -> Option<ScheduledDemand> {
         requester: PartitionId,
         assignment_id: DemandAssignmentId,
         selected_order: DemandOrder,
     ) -> Option<DemandAssignment> {
         let record = self
             .records
             .get_mut(&requesting_partition)
             .get_mut(&requester)
             .expect("order head disappeared");
         let residence = std::mem::replace(&mut record.residence, DemandResidence::Idle);
         match residence {
             DemandResidence::Queued { demand, links } => {
         let schedule_state =
             std::mem::replace(&mut record.schedule_state, DemandScheduleState::Unscheduled);
         match schedule_state {
             DemandScheduleState::Queued { demand, links } => {
                 debug_assert_eq!(record.latest.id, demand);
                 debug_assert!(record.latest.is_active());
                 record.residence = DemandResidence::Delivering {
                 let assignment = DemandAssignment {
                     id: assignment_id,
                     requester,
                     demand,
                 };
                 record.schedule_state = DemandScheduleState::PendingAssignment {
                     assignment: assignment.clone(),
                     version: record.latest.version,
                     delivery,
                     view,
                     selected_order,
                     links,
                 };
                 self.assert_consistent();
                 Some(ScheduledDemand {
                     requesting_partition,
                     demand,
                 })
                 Some(assignment)
-627
+            }
             residence => {
                 record.residence = residence;
             schedule_state => {
                 record.schedule_state = schedule_state;
                 None
-631
+            }
-632
+        }
-633
+    }
-634
     /// Returns whether the test-observed delivery fence still names this demand.
     ///
     /// Production reservation revalidates the demand identity through
     /// `AcquisitionQueue::reserve_delivery_waiter`; this helper only exposes
     /// admission-side fence state to focused tests.
     #[cfg(test)]
     pub(super) fn delivery_is_current(
         &self,
         delivery: DeliveryId,
         requesting_partition: &PartitionId,
         demand: DemandId,
     ) -> bool {
         let Some(record) = self.records.get(requesting_partition) else {
     pub(super) fn assignment_is_current(&self, assignment: &DemandAssignment) -> bool {
         let Some(record) = self.records.get(&assignment.requester) else {
             return false;
         };
         matches!(
             &record.residence,
             DemandResidence::Delivering {
                 demand: current_demand,
                 delivery: current_delivery,
                 ..
             } if *current_delivery == delivery
                 && *current_demand == demand
                 && record.latest.id == demand
                 && record.latest.is_active()
             &record.schedule_state,
             DemandScheduleState::PendingAssignment { assignment: current, .. }
                 if current == assignment
                     && record.latest.id == assignment.demand
                     && record.latest.is_active()
-646
+        )
-647
+    }
-648
     /// Closes one delivery fence and resolves the demand's next residence.
     ///
     /// `Accepted` consumes the delivered generation and installs a newer
     /// successor when the requesting cell supplied one. `RetrySameResidence`
     /// restores the same generation at its existing order position when it
     /// remains live; if it was replaced, the latest active generation is
     /// appended as new demand. `Rejected` removes the fenced generation after
     /// its permit has been refunnelled and applies the same successor
     /// arbitration as acceptance.
     pub(super) fn finish_delivery(
     /// Settles one detached assignment and resolves the demand's next state.
     pub(super) fn settle_assignment(
         &mut self,
         delivery: DeliveryId,
         requesting_partition: &PartitionId,
         result: DeliveryAckResult,
         assignment: &DemandAssignment,
         outcome: DemandAssignmentOutcome,
     ) {
         let Some(record) = self.records.get(requesting_partition) else {
         let requester = &assignment.requester;
         let Some(record) = self.records.get(requester) else {
             return;
         };
         let delivered_demand = match &record.residence {
             DemandResidence::Delivering {
                 demand,
         let assigned = match &record.schedule_state {
             DemandScheduleState::PendingAssignment {
                 assignment: current,
                 version,
                 delivery: current,
                 ..
             } if *current == delivery => (*demand, *version),
             } if current == assignment => (current.demand, *version),
             _ => return,
         };
         let (delivered_demand, delivered_version) = delivered_demand;
         let (assigned_demand, assigned_version) = assigned;
-668
         match result {
             DeliveryAckResult::RetrySameResidence => {
         match outcome {
             DemandAssignmentOutcome::RetrySamePosition => {
                 let record = self
                     .records
                     .get(requesting_partition)
                     .expect("delivery demand record disappeared");
                 if record.latest.id == delivered_demand && record.latest.is_active() {
                     .get(requester)
                     .expect("assigned demand record disappeared");
                 if record.latest.id == assigned_demand && record.latest.is_active() {
                     let record = self
                         .records
                         .get_mut(requesting_partition)
                         .expect("delivery demand record disappeared");
                     let residence = std::mem::replace(&mut record.residence, DemandResidence::Idle);
                     let DemandResidence::Delivering {
                         demand,
                         version: _,
                         delivery: current,
                         view: _,
                         .get_mut(requester)
                         .expect("assigned demand record disappeared");
                     let schedule_state = std::mem::replace(
                         &mut record.schedule_state,
                         DemandScheduleState::Unscheduled,
                     );
                     let DemandScheduleState::PendingAssignment {
                         assignment: current,
                         links,
                     } = residence
                         ..
                     } = schedule_state
                     else {
                         unreachable!("delivery fence disappeared");
                         unreachable!("demand assignment disappeared");
                     };
                     debug_assert_eq!(current, *assignment);
                     record.schedule_state = DemandScheduleState::Queued {
                         demand: current.demand,
                         links,
                     };
                     debug_assert_eq!(current, delivery);
                     record.residence = DemandResidence::Queued { demand, links };
                     self.assert_consistent();
                     return;
-699
+                }
-700
                 self.remove_from_order(requesting_partition);
                 self.remove_from_order(requester);
                 if self
                     .records
                     .get(requesting_partition)
                     .expect("delivery demand record disappeared")
                     .get(requester)
                     .expect("assigned demand record disappeared")
                     .latest
                     .is_active()
-708
+                {
                     self.enqueue(*requesting_partition);
                     self.enqueue(*requester);
-710
+                }
-711
+            }
             DeliveryAckResult::Accepted { successor }
             | DeliveryAckResult::Rejected { successor } => {
                 self.remove_from_order(requesting_partition);
             DemandAssignmentOutcome::Accepted { successor }
             | DemandAssignmentOutcome::Refused { successor } => {
                 self.remove_from_order(requester);
-715
                 let install_successor = successor.as_ref().is_some_and(|successor| {
                     successor.id > delivered_demand
                     successor.id > assigned_demand
                         && successor.is_newer_than(
                             &self
                                 .records
                                 .get(requesting_partition)
                                 .expect("delivery demand record disappeared")
                                 .get(requester)
                                 .expect("assigned demand record disappeared")
                                 .latest,
-724
+                        )
                 });
                 if install_successor {
                     self.records
                         .get_mut(requesting_partition)
                         .expect("delivery demand record disappeared")
                         .get_mut(requester)
                         .expect("assigned demand record disappeared")
                         .latest = successor.expect("validated successor disappeared");
                 } else {
                     let retirement =
                         DemandSnapshot::inactive(delivered_demand, delivered_version.next());
                         DemandSnapshot::inactive(assigned_demand, assigned_version.next());
                     let record = self
                         .records
                         .get_mut(requesting_partition)
                         .expect("delivery demand record disappeared");
                         .get_mut(requester)
                         .expect("assigned demand record disappeared");
                     if retirement.is_newer_than(&record.latest) {
                         record.latest = retirement;
-740
+                    }
-741
+                }
-742
                 if self
                     .records
                     .get(requesting_partition)
                     .expect("delivery demand record disappeared")
                     .get(requester)
                     .expect("assigned demand record disappeared")
                     .latest
                     .is_active()
-749
+                {
                     self.enqueue(*requesting_partition);
                     self.enqueue(*requester);
-751
+                }
-752
+            }
-753
+        }
         self.assert_consistent();
-755
+    }
-756
     /// Returns the latest complete snapshot retained for `requesting_partition`.
     /// Returns the latest complete snapshot retained for `requester`.
     #[cfg(test)]
     pub(super) fn latest_for_test(
         &self,
         requesting_partition: &PartitionId,
     ) -> Option<&DemandSnapshot> {
         self.records
             .get(requesting_partition)
             .map(|record| &record.latest)
     pub(super) fn latest_for_test(&self, requester: &PartitionId) -> Option<&DemandSnapshot> {
         self.records.get(requester).map(|record| &record.latest)
-761
+    }
-762
     #[cfg(test)]
     pub(super) fn len(&self) -> usize {
         self.origin_order.len()
-766
+    }
-767
     #[cfg(test)]
     pub(super) fn queued_len(&self) -> usize {
         self.records
             .values()
             .filter(|record| matches!(&record.residence, DemandResidence::Queued { .. }))
             .filter(|record| matches!(&record.schedule_state, DemandScheduleState::Queued { .. }))
             .count()
-774
+    }
-775
     #[cfg(test)]
     pub(super) fn delivering_len(&self) -> usize {
     pub(super) fn pending_assignment_count(&self) -> usize {
         self.records
             .values()
             .filter(|record| matches!(&record.residence, DemandResidence::Delivering { .. }))
             .filter(|record| {
                 matches!(
                     &record.schedule_state,
                     DemandScheduleState::PendingAssignment { .. }
-+
+                )
             })
             .count()
-787
+    }
-788
     /// Checks residence, link, length, group, and fence relationships.
     /// Checks residence, link, length, group, and assignment relationships.
     fn assert_consistent(&self) {
         #[cfg(any(debug_assertions, test))]
-792
+        {
             if std::thread::panicking() {
                 return;
-795
+            }
             self.assert_consistent_debug();
-797
+        }
-798
+    }
-799
     #[cfg(any(debug_assertions, test))]
     fn assert_consistent_debug(&self) {
         let ordered_records = self
             .records
             .values()
             .filter(|record| record.residence.links(DemandOrderView::Origin).is_some())
             .filter(|record| record.schedule_state.links(DemandOrder::Origin).is_some())
             .count();
         let origin_deliveries = self
         let origin_assignments = self
             .records
             .values()
             .filter(|record| {
                 matches!(
                     record.residence,
                     DemandResidence::Delivering {
                         view: DeliveryView::Origin,
                     record.schedule_state,
                     DemandScheduleState::PendingAssignment {
                         selected_order: DemandOrder::Origin,
                         ..
-816
+                    }
-817
+                )
             })
             .count();
         assert!(
             origin_deliveries <= 1,
             "more than one origin delivery fence was active"
             origin_assignments <= 1,
             "more than one origin demand assignment was active"
         );
         self.origin_order.assert_consistent(
             ordered_records,
             self.records.len(),
             "origin demand order",
             |requesting_partition| {
                 let record = self
                     .records
                     .get(&requesting_partition)
                     .expect("origin-ordered demand disappeared");
                 match &record.residence {
                     DemandResidence::Idle => {
                         unreachable!("origin-ordered demand became idle")
                 match &record.schedule_state {
                     DemandScheduleState::Unscheduled => {
                         unreachable!("origin-ordered demand became unscheduled")
-836
+                    }
                     DemandResidence::Queued { demand, .. } => {
                     DemandScheduleState::Queued { demand, .. } => {
                         assert!(record.latest.is_active(), "queued demand became inactive");
                         assert_eq!(
                             record.latest.id, *demand,
                             "queued residence did not match its latest demand"
                             "queued schedule state did not match its latest demand"
                         );
-843
+                    }
                     DemandResidence::Delivering { demand, view, .. } => {
                         if *view == DeliveryView::Origin {
                     DemandScheduleState::PendingAssignment {
                         assignment,
                         selected_order,
                         ..
                     } => {
                         if *selected_order == DemandOrder::Origin {
                             assert_eq!(
                                 Some(requesting_partition),
                                 self.origin_order.head(),
                                 "origin delivery fence moved away from its head"
                                 "origin demand assignment moved away from its head"
                             );
-855
+                        }
                         assert!(
                             record.latest.id >= *demand,
                             "delivery fence named a future demand"
                             record.latest.id >= assignment.demand,
                             "demand assignment named a future demand"
                         );
-860
+                    }
-861
+                }
                 *record
                     .residence
                     .links(DemandOrderView::Origin)
                     .schedule_state
                     .links(DemandOrder::Origin)
                     .expect("origin-ordered demand lost its links")
             },
         );
-868
         let group_ordered_records = self
             .records
             .values()
             .filter(|record| record.residence.links(DemandOrderView::Group).is_some())
             .filter(|record| record.schedule_state.links(DemandOrder::Group).is_some())
             .count();
         for record in self
             .records
             .values()
             .filter(|record| record.residence.links(DemandOrderView::Group).is_some())
             .filter(|record| record.schedule_state.links(DemandOrder::Group).is_some())
-878
+        {
             let group = record
                 .group
                 .eligibility_group
                 .as_ref()
                 .expect("group-ordered demand lost its eligibility group");
             assert!(
                 self.group_orders.contains_key(group),
                 "ordered demand lost its eligibility-group order"
             );
-887
+        }
         assert_eq!(
             group_ordered_records,
             self.group_orders
                 .values()
                 .map(IntrusiveOrder::len)
                 .sum::<usize>(),
             "eligibility-group orders did not contain every ordered demand"
         );
-896
         for (group, order) in &self.group_orders {
             let group_deliveries = self
             let group_assignments = self
                 .records
                 .values()
                 .filter(|record| {
                     record.group.as_ref() == Some(group)
                     record.eligibility_group.as_ref() == Some(group)
                         && matches!(
                             record.residence,
                             DemandResidence::Delivering {
                                 view: DeliveryView::Group,
                             record.schedule_state,
                             DemandScheduleState::PendingAssignment {
                                 selected_order: DemandOrder::Group,
                                 ..
-908
+                            }
-909
+                        )
                 })
                 .count();
             assert!(
                 group_deliveries <= 1,
                 "more than one group delivery fence was active"
                 group_assignments <= 1,
                 "more than one group demand assignment was active"
             );
             let expected = self
                 .records
                 .values()
                 .filter(|record| {
                     record.residence.links(DemandOrderView::Group).is_some()
                         && record.group.as_ref() == Some(group)
                     record.schedule_state.links(DemandOrder::Group).is_some()
                         && record.eligibility_group.as_ref() == Some(group)
                 })
                 .count();
             order.assert_consistent(
                 expected,
                 self.records.len(),
                 "eligibility-group demand order",
                 |requesting_partition| {
                     let record = self
                         .records
                         .get(&requesting_partition)
                         .expect("group-ordered demand disappeared");
                     assert_eq!(
                         record.group.as_ref(),
                         record.eligibility_group.as_ref(),
                         Some(group),
                         "demand occupied the wrong eligibility-group order"
                     );
                     if matches!(
                         record.residence,
                         DemandResidence::Delivering {
                             view: DeliveryView::Group,
                         record.schedule_state,
                         DemandScheduleState::PendingAssignment {
                             selected_order: DemandOrder::Group,
                             ..
-943
+                        }
                     ) {
                         assert_eq!(
                             Some(requesting_partition),
                             order.head(),
                             "group publication fence moved away from its head"
                             "group route assignment moved away from its head"
                         );
-950
+                    }
                     *record
                         .residence
                         .links(DemandOrderView::Group)
                         .schedule_state
                         .links(DemandOrder::Group)
                         .expect("group-ordered demand lost its links")
                 },
             );
-957
+        }
-958
+    }
-959
+}
-960
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
-964
     fn partition(index: usize) -> PartitionId {
         PartitionId::from_index(index)
-967
+    }
-968
     fn active(id: u64, group: EligibilityGroup) -> DemandSnapshot {
         DemandSnapshot::active(
             DemandId::from_u64(id),
             super::super::SnapshotVersion::INITIAL,
             ProtocolRequirement::H1Compatible,
             group,
-975
+        )
-976
+    }
-977
     #[test]
     fn origin_and_group_orders_repair_independently() {
         let pool = EligibilityGroup::Pool;
         let isolated = EligibilityGroup::Partition(partition(2));
         let mut schedule = DemandSchedule::default();
         schedule.publish(partition(1), active(1, pool.clone()));
         schedule.publish(partition(2), active(2, isolated.clone()));
         schedule.publish(partition(3), active(3, pool.clone()));
         schedule.apply_snapshot(partition(1), active(1, pool.clone()));
         schedule.apply_snapshot(partition(2), active(2, isolated.clone()));
         schedule.apply_snapshot(partition(3), active(3, pool.clone()));
-986
         assert_eq!(
             Some(partition(1)),
             schedule.queued_head().map(|head| head.requesting_partition)
             schedule.queued_head().map(|head| head.requester)
         );
         assert_eq!(
             Some(partition(1)),
             schedule
                 .queued_group_head(&pool)
                 .map(|head| head.requesting_partition)
             schedule.queued_group_head(&pool).map(|head| head.requester)
         );
         assert_eq!(
             Some(partition(2)),
             schedule
                 .queued_group_head(&isolated)
                 .map(|head| head.requesting_partition)
                 .map(|head| head.requester)
         );
-1001
         let delivery = DeliveryId(7);
         schedule
             .reserve_group_head(&pool, &partition(1), DemandId::from_u64(1), delivery)
         let assignment = schedule
             .prepare_group_assignment(
                 &pool,
                 &partition(1),
                 DemandId::from_u64(1),
                 DemandAssignmentId(7),
-+
+            )
             .expect("pool group head should reserve");
         assert!(schedule.queued_group_head(&pool).is_none());
         assert_eq!(
             Some(partition(2)),
             schedule
                 .queued_group_head(&isolated)
                 .map(|head| head.requesting_partition)
                 .map(|head| head.requester)
         );
         assert!(schedule.queued_head().is_none());
-1018
         schedule.finish_delivery(
             delivery,
             &partition(1),
             DeliveryAckResult::Accepted { successor: None },
         schedule.settle_assignment(
             &assignment,
             DemandAssignmentOutcome::Accepted { successor: None },
         );
         assert_eq!(
             Some(partition(2)),
             schedule.queued_head().map(|head| head.requesting_partition)
             schedule.queued_head().map(|head| head.requester)
         );
         assert_eq!(
             Some(partition(3)),
             schedule
                 .queued_group_head(&pool)
                 .map(|head| head.requesting_partition)
             schedule.queued_group_head(&pool).map(|head| head.requester)
         );
-1031
+    }
-1032
     #[test]
     fn retry_preserves_both_order_positions() {
         let group = EligibilityGroup::Pool;
         let mut schedule = DemandSchedule::default();
         schedule.publish(partition(1), active(1, group.clone()));
         schedule.publish(partition(2), active(2, group.clone()));
         schedule.apply_snapshot(partition(1), active(1, group.clone()));
         schedule.apply_snapshot(partition(2), active(2, group.clone()));
-1039
         let delivery = DeliveryId(9);
         schedule
             .reserve_group_head(&group, &partition(1), DemandId::from_u64(1), delivery)
         let assignment = schedule
             .prepare_group_assignment(
                 &group,
                 &partition(1),
                 DemandId::from_u64(1),
                 DemandAssignmentId(9),
-+
+            )
             .expect("group head should reserve");
         schedule.finish_delivery(
             delivery,
             &partition(1),
             DeliveryAckResult::RetrySameResidence,
         );
         schedule.settle_assignment(&assignment, DemandAssignmentOutcome::RetrySamePosition);
-1049
         assert_eq!(
             Some(partition(1)),
             schedule.queued_head().map(|head| head.requesting_partition)
             schedule.queued_head().map(|head| head.requester)
         );
         assert_eq!(
             Some(partition(1)),
             schedule
                 .queued_group_head(&group)
                 .map(|head| head.requesting_partition)
                 .map(|head| head.requester)
         );
-1060
+    }
-1061
     #[test]
     fn inactive_snapshot_during_group_fence_retires_without_losing_group() {
     fn inactive_snapshot_during_group_assignment_retires_without_losing_group() {
         let group = EligibilityGroup::Pool;
         let mut schedule = DemandSchedule::default();
         schedule.publish(partition(1), active(1, group.clone()));
         let delivery = DeliveryId(11);
         schedule
             .reserve_group_head(&group, &partition(1), DemandId::from_u64(1), delivery)
         schedule.apply_snapshot(partition(1), active(1, group.clone()));
         let assignment = schedule
             .prepare_group_assignment(
                 &group,
                 &partition(1),
                 DemandId::from_u64(1),
                 DemandAssignmentId(11),
-+
+            )
             .expect("group head should reserve");
         schedule.publish(
         schedule.apply_snapshot(
             partition(1),
             DemandSnapshot::inactive(
                 DemandId::from_u64(1),
                 super::super::SnapshotVersion::INITIAL.next(),
             ),
         );
-1082
         schedule.finish_delivery(
             delivery,
             &partition(1),
             DeliveryAckResult::RetrySameResidence,
         );
         schedule.settle_assignment(&assignment, DemandAssignmentOutcome::RetrySamePosition);
         assert_eq!(0, schedule.len());
         assert!(schedule.queued_group_head(&group).is_none());
-1086
+    }
-1087
     #[test]
     fn accepted_fence_does_not_retire_a_newer_active_publication() {
     fn accepted_assignment_does_not_retire_a_newer_active_snapshot() {
         let group = EligibilityGroup::Pool;
         let partition = partition(1);
         let demand = DemandId::from_u64(1);
         let initial = super::super::SnapshotVersion::INITIAL;
         let mut schedule = DemandSchedule::default();
         schedule.publish(
         schedule.apply_snapshot(
             partition,
             DemandSnapshot::active(
                 demand,
                 initial,
                 ProtocolRequirement::H2Required,
                 group.clone(),
             ),
         );
         let delivery = DeliveryId(12);
         schedule
             .reserve_group_head(&group, &partition, demand, delivery)
         let assignment = schedule
             .prepare_group_assignment(&group, &partition, demand, DemandAssignmentId(12))
             .expect("group head should reserve");
         let republished = DemandSnapshot::active(
             demand,
             initial.next().next(),
             ProtocolRequirement::H2Required,
             group.clone(),
         );
         schedule.publish(partition, republished.clone());
         schedule.apply_snapshot(partition, republished.clone());
-1114
         schedule.finish_delivery(
             delivery,
             &partition,
             DeliveryAckResult::Accepted { successor: None },
         schedule.settle_assignment(
             &assignment,
             DemandAssignmentOutcome::Accepted { successor: None },
         );
-1119
         assert_eq!(Some(&republished), schedule.latest_for_test(&partition));
         assert_eq!(
             Some(partition),
             schedule.queued_head().map(|head| head.requesting_partition)
             schedule.queued_head().map(|head| head.requester)
         );
         assert_eq!(
             Some(partition),
             schedule
                 .queued_group_head(&group)
                 .map(|head| head.requesting_partition)
                 .map(|head| head.requester)
         );
-1131
+    }
-1132
     #[test]
     fn group_reservation_rejects_a_stale_demand_identity() {
     fn group_assignment_rejects_a_stale_demand_identity() {
         let group = EligibilityGroup::Pool;
         let partition = partition(1);
         let mut schedule = DemandSchedule::default();
         schedule.publish(partition, active(1, group.clone()));
         schedule.publish(partition, active(2, group.clone()));
         schedule.apply_snapshot(partition, active(1, group.clone()));
         schedule.apply_snapshot(partition, active(2, group.clone()));
-1140
         assert!(schedule
             .reserve_group_head(&group, &partition, DemandId::from_u64(1), DeliveryId(13),)
             .prepare_group_assignment(
                 &group,
                 &partition,
                 DemandId::from_u64(1),
                 DemandAssignmentId(13),
-+
+            )
             .is_none());
         assert_eq!(
             Some(DemandId::from_u64(2)),
             schedule
                 .queued_group_head(&group)
                 .map(|queued| queued.demand)
         );
-1155
+    }
-1156
     #[test]
     fn stale_acknowledgement_does_not_close_a_newer_delivery_fence() {
     fn stale_settlement_does_not_close_a_newer_assignment() {
         let group = EligibilityGroup::Pool;
         let partition = partition(1);
         let demand = DemandId::from_u64(1);
         let mut schedule = DemandSchedule::default();
         schedule.publish(partition, active(1, group.clone()));
         schedule
             .reserve_group_head(&group, &partition, demand, DeliveryId(14))
         schedule.apply_snapshot(partition, active(1, group.clone()));
         let assignment = schedule
             .prepare_group_assignment(&group, &partition, demand, DemandAssignmentId(14))
             .expect("group head should reserve");
         let stale = DemandAssignment {
             id: DemandAssignmentId(15),
             requester: partition,
             demand,
         };
-1172
         schedule.finish_delivery(
             DeliveryId(15),
             &partition,
             DeliveryAckResult::Accepted { successor: None },
         schedule.settle_assignment(
             &stale,
             DemandAssignmentOutcome::Accepted { successor: None },
         );
-1177
         assert!(schedule.delivery_is_current(DeliveryId(14), &partition, demand));
         assert!(!schedule.delivery_is_current(DeliveryId(15), &partition, demand));
         assert!(schedule.assignment_is_current(&assignment));
         assert!(!schedule.assignment_is_current(&stale));
         assert_eq!(
             Some(&active(1, group)),
             schedule.latest_for_test(&partition)
         );
-1184
+    }
-1185
     #[test]
     #[should_panic(expected = "ordered demand lost its eligibility-group order")]
     fn consistency_check_rejects_an_orphaned_group_link() {
         let group = EligibilityGroup::Pool;
         let mut schedule = DemandSchedule::default();
         schedule.publish(partition(1), active(1, group.clone()));
         schedule.apply_snapshot(partition(1), active(1, group.clone()));
-1192
         schedule.group_orders.remove(&group);
         schedule.assert_consistent();
-1195
+    }
-1196
+}

Pages: 1 2 3 4 5 6 7 8 9 10 11