AWS SDK

AWS SDK

rev. 68ed678eb244336c4e4a0f89bed843ed63aa9548 (ignoring whitespace)

Files changed:

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

@@ -0,1 +0,1419 @@
           1  +
/*
           2  +
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
           3  +
 * SPDX-License-Identifier: Apache-2.0
           4  +
 */
           5  +
           6  +
//! Admission's view of HTTP/1 connection supply for one bounded origin.
           7  +
//!
           8  +
//! [`H1Supply`] indexes cells with selectable HTTP/1 senders and pairs the oldest
           9  +
//! origin demand with one supplier. Compatible demand borrows the sender.
          10  +
//! Incompatible demand closes the selected connection to recover capacity.
          11  +
//!
          12  +
//! Admission retains an [`H1Match`] while supplier reservation crosses locks or
          13  +
//! waits for a sender return. The match owns identities and phase flags only.
          14  +
//! Detached values own resources and their drop fallback.
          15  +
//!
          16  +
//! Admission advances each match through these phases:
          17  +
//!
          18  +
//! ```text
          19  +
//! Reserving -- reservation installed -----------------------> WaitingForSender
          20  +
//! Reserving -- sender extracted ----------------------------> Resolving
          21  +
//! Reserving(cancelled) -- reservation installed ------------> Cancelling
          22  +
//! Reserving(cancelled) -- sender extracted -----------------> Resolving
          23  +
//! Reserving -- reservation rejected or cell expired --------> removed
          24  +
//! WaitingForSender -- sender return intercepted ------------> Resolving
          25  +
//! WaitingForSender -- demand cancelled ---------------------> Cancelling
          26  +
//! WaitingForSender -- terminal supplier outcome ------------> removed
          27  +
//! Cancelling -- sender return intercepted ------------------> Resolving
          28  +
//! Cancelling -- reservation cleared -------------------------> removed
          29  +
//! Resolving -- repeated candidate handoff ------------------> Resolving
          30  +
//! Resolving -- borrow or reclaim completes ------------------> removed
          31  +
//! ```
          32  +
//!
          33  +
//! A sender extracted during installation takes the `Resolving` path even if
          34  +
//! cancellation raced with the install acknowledgement. The extracted sender
          35  +
//! must complete its terminal transition before cancellation can finish.
          36  +
          37  +
use super::{
          38  +
    AdmissionAction, DeliveryGuard, DemandAssignmentOutcome, DemandId, DemandSchedule,
          39  +
    IntrusiveLinks, IntrusiveOrder, OriginAdmission, SupplyRevision,
          40  +
};
          41  +
use crate::client::pool::cell::h1::{H1Selection, ProvisionalH1};
          42  +
use crate::client::pool::cell::{H1ReservationDecision, OriginCell};
          43  +
use crate::client::pool::partition::{EligibilityGroup, PartitionId};
          44  +
use crate::sync::Arc;
          45  +
use aws_smithy_runtime_api::client::connection::ConnectionId;
          46  +
use std::collections::{HashMap, VecDeque};
          47  +
use std::fmt;
          48  +
          49  +
/// Identity of one retained HTTP/1 demand-to-supplier match.
          50  +
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
          51  +
pub(in crate::client::pool) struct H1MatchId(u64);
          52  +
          53  +
impl H1MatchId {
          54  +
    /// Creates a deterministic identity for focused transition tests.
          55  +
    #[cfg(test)]
          56  +
    pub(in crate::client::pool) const fn for_test(value: u64) -> Self {
          57  +
        Self(value)
          58  +
    }
          59  +
}
          60  +
          61  +
/// How one retained HTTP/1 match satisfies waiting demand.
          62  +
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
          63  +
pub(in crate::client::pool) enum H1MatchKind {
          64  +
    /// Move the selected sender to the requesting cell for dispatch.
          65  +
    BorrowSender,
          66  +
    /// Close the selected connection and deliver its capacity instead.
          67  +
    ReclaimCapacity,
          68  +
}
          69  +
          70  +
/// Origin-locked HTTP/1 supply indexes and retained cross-lock matches.
          71  +
#[derive(Debug, Default)]
          72  +
pub(super) struct H1Supply {
          73  +
    index: H1SupplyIndex,
          74  +
    matches: HashMap<H1MatchId, H1Match>,
          75  +
    by_requester: HashMap<PartitionId, H1MatchId>,
          76  +
    /// Lazy cancellation queue; resolution may leave a tombstone to discard.
          77  +
    cancellations: VecDeque<H1MatchId>,
          78  +
    next_match_id: u64,
          79  +
}
          80  +
          81  +
/// Admission-facing HTTP/1 status derived under one cell lock.
          82  +
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
          83  +
pub(in crate::client::pool) struct H1SupplyStatus {
          84  +
    pub(in crate::client::pool) has_returnable_connection: bool,
          85  +
    pub(in crate::client::pool) peer_use_blocked: bool,
          86  +
}
          87  +
          88  +
/// Terminal HTTP/1 supply observed at a supplier cell.
          89  +
pub(in crate::client::pool) enum H1SupplyOutcome {
          90  +
    SupplierLive {
          91  +
        supplier: PartitionId,
          92  +
        revision: SupplyRevision<H1SupplyStatus>,
          93  +
    },
          94  +
    SupplierExpired {
          95  +
        supplier: PartitionId,
          96  +
    },
          97  +
}
          98  +
          99  +
impl H1SupplyOutcome {
         100  +
    pub(in crate::client::pool) fn supplier_live(
         101  +
        supplier: PartitionId,
         102  +
        revision: SupplyRevision<H1SupplyStatus>,
         103  +
    ) -> Self {
         104  +
        Self::SupplierLive { supplier, revision }
         105  +
    }
         106  +
         107  +
    pub(in crate::client::pool) fn supplier_expired(supplier: PartitionId) -> Self {
         108  +
        Self::SupplierExpired { supplier }
         109  +
    }
         110  +
         111  +
    fn supplier(&self) -> &PartitionId {
         112  +
        match self {
         113  +
            Self::SupplierLive { supplier, .. } | Self::SupplierExpired { supplier } => supplier,
         114  +
        }
         115  +
    }
         116  +
}
         117  +
         118  +
/// Admission's indexed view of supplier cells exposing selectable H1 supply.
         119  +
#[derive(Debug, Default)]
         120  +
struct H1SupplyIndex {
         121  +
    records: HashMap<PartitionId, H1SupplyRecord>,
         122  +
    reclaim_order: IntrusiveOrder<PartitionId>,
         123  +
    suppliers_by_group: HashMap<EligibilityGroup, IntrusiveOrder<PartitionId>>,
         124  +
}
         125  +
         126  +
/// Admission's current view of one HTTP/1 supplier cell.
         127  +
#[derive(Debug)]
         128  +
struct H1SupplyRecord {
         129  +
    /// Reuse group whose peers may borrow this connection cell's sender.
         130  +
    eligibility_group: EligibilityGroup,
         131  +
    /// Newest connection cell report accepted by admission.
         132  +
    revision: u64,
         133  +
    /// Whether the cell has an H1 record that can return or be reclaimed.
         134  +
    status: H1SupplyStatus,
         135  +
    reserved_by: Option<H1MatchId>,
         136  +
    index_state: H1SupplyIndexState,
         137  +
}
         138  +
         139  +
impl H1SupplyRecord {
         140  +
    /// Returns whether this connection cell must occupy both scheduling views.
         141  +
    fn is_selectable(&self) -> bool {
         142  +
        self.status.has_returnable_connection
         143  +
            && !self.status.peer_use_blocked
         144  +
            && self.reserved_by.is_none()
         145  +
    }
         146  +
}
         147  +
         148  +
/// Whether a connection cell is linked in both scheduling views.
         149  +
#[derive(Debug, Default)]
         150  +
enum H1SupplyIndexState {
         151  +
    /// The connection cell is absent from connection cell-selection order.
         152  +
    #[default]
         153  +
    Unlinked,
         154  +
    /// The connection cell is linked once in origin and group order.
         155  +
    Linked {
         156  +
        /// Links in origin-wide reclaim order.
         157  +
        reclaim: IntrusiveLinks<PartitionId>,
         158  +
        /// Links in eligibility-group borrow order.
         159  +
        group: IntrusiveLinks<PartitionId>,
         160  +
    },
         161  +
}
         162  +
         163  +
impl H1SupplyIndexState {
         164  +
    fn is_linked(&self) -> bool {
         165  +
        matches!(self, Self::Linked { .. })
         166  +
    }
         167  +
}
         168  +
         169  +
/// Admission-owned state for one nonterminal demand/supplier match.
         170  +
#[derive(Clone, Debug)]
         171  +
struct H1Match {
         172  +
    /// Cell whose connection is reserved by this match.
         173  +
    supplier: PartitionId,
         174  +
    /// Cell whose demand caused this match.
         175  +
    requester: PartitionId,
         176  +
    /// Exact requesting-cell demand generation that caused this match.
         177  +
    demand: DemandId,
         178  +
    /// Whether resolution borrows a sender or reclaims its capacity.
         179  +
    kind: H1MatchKind,
         180  +
    /// Admission-side progress of the match.
         181  +
    state: H1MatchState,
         182  +
    /// Whether requesting cell demand became stale before connection-cell
         183  +
    /// resolution.
         184  +
    cancelled: bool,
         185  +
}
         186  +
         187  +
/// Origin-side progress of one retained H1 match.
         188  +
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
         189  +
enum H1MatchState {
         190  +
    /// Admission selected a connection, but the cell has not reserved it.
         191  +
    Reserving,
         192  +
    /// The cell will intercept the connection's next reusable return.
         193  +
    WaitingForSender,
         194  +
    /// A provisional sender is resolving outside admission.
         195  +
    Resolving,
         196  +
    /// Request cancellation must clear the installed reservation.
         197  +
    Cancelling,
         198  +
}
         199  +
         200  +
/// Work required to reserve one supplier outside the admission lock.
         201  +
#[derive(Debug)]
         202  +
pub(in crate::client::pool) struct PreparedH1Reservation {
         203  +
    /// Match whose supplier is being reserved.
         204  +
    pub(in crate::client::pool) match_id: H1MatchId,
         205  +
    /// Cell whose connection was selected while admission was locked.
         206  +
    pub(in crate::client::pool) supplier: PartitionId,
         207  +
}
         208  +
         209  +
/// Work required to cancel a reservation outside the admission lock.
         210  +
pub(super) struct PreparedH1Cancellation {
         211  +
    /// Match whose supplier reservation must be cleared.
         212  +
    match_id: H1MatchId,
         213  +
    /// Cell that owns the reservation.
         214  +
    supplier: PartitionId,
         215  +
}
         216  +
         217  +
impl H1SupplyIndex {
         218  +
    fn apply_revision(
         219  +
        &mut self,
         220  +
        supplier: PartitionId,
         221  +
        eligibility_group: EligibilityGroup,
         222  +
        revision: SupplyRevision<H1SupplyStatus>,
         223  +
    ) {
         224  +
        if self
         225  +
            .records
         226  +
            .get(&supplier)
         227  +
            .is_some_and(|record| record.revision >= revision.revision)
         228  +
        {
         229  +
            return;
         230  +
        }
         231  +
        self.unlink_supplier(&supplier);
         232  +
        let record = self
         233  +
            .records
         234  +
            .entry(supplier)
         235  +
            .or_insert_with(|| H1SupplyRecord {
         236  +
                eligibility_group: eligibility_group.clone(),
         237  +
                revision: revision.revision,
         238  +
                status: revision.status,
         239  +
                reserved_by: None,
         240  +
                index_state: H1SupplyIndexState::Unlinked,
         241  +
            });
         242  +
        record.eligibility_group = eligibility_group;
         243  +
        record.revision = revision.revision;
         244  +
        record.status = revision.status;
         245  +
        self.link_supplier_if_selectable(&supplier);
         246  +
    }
         247  +
         248  +
    fn select_borrow_supplier(
         249  +
        &mut self,
         250  +
        eligibility_group: &EligibilityGroup,
         251  +
        requester: PartitionId,
         252  +
    ) -> Option<PartitionId> {
         253  +
        let supplier = {
         254  +
            let order = self.suppliers_by_group.get(eligibility_group)?;
         255  +
            self.first_peer_supplier(order, requester, false)?
         256  +
        };
         257  +
        self.unlink_supplier(&supplier);
         258  +
        Some(supplier)
         259  +
    }
         260  +
         261  +
    fn select_peer_reclaim_supplier(&mut self, requester: PartitionId) -> Option<PartitionId> {
         262  +
        let supplier = self.first_peer_supplier(&self.reclaim_order, requester, true)?;
         263  +
        self.unlink_supplier(&supplier);
         264  +
        Some(supplier)
         265  +
    }
         266  +
         267  +
    fn select_oldest_reclaim_supplier(&mut self) -> Option<PartitionId> {
         268  +
        let supplier = self.reclaim_order.head()?;
         269  +
        self.unlink_supplier(&supplier);
         270  +
        Some(supplier)
         271  +
    }
         272  +
         273  +
    fn first_peer_supplier(
         274  +
        &self,
         275  +
        order: &IntrusiveOrder<PartitionId>,
         276  +
        requester: PartitionId,
         277  +
        reclaim: bool,
         278  +
    ) -> Option<PartitionId> {
         279  +
        let head = order.head()?;
         280  +
        if head != requester {
         281  +
            return Some(head);
         282  +
        }
         283  +
        let record = self
         284  +
            .records
         285  +
            .get(&head)
         286  +
            .expect("H1 supply order head disappeared");
         287  +
        let H1SupplyIndexState::Linked {
         288  +
            reclaim: reclaim_links,
         289  +
            group,
         290  +
        } = &record.index_state
         291  +
        else {
         292  +
            unreachable!("ordered H1 supplier was unlinked");
         293  +
        };
         294  +
        if reclaim {
         295  +
            reclaim_links.next
         296  +
        } else {
         297  +
            group.next
         298  +
        }
         299  +
    }
         300  +
         301  +
    fn link_supplier_if_selectable(&mut self, supplier: &PartitionId) {
         302  +
        let Some(record) = self.records.get(supplier) else {
         303  +
            return;
         304  +
        };
         305  +
        if !record.is_selectable() || record.index_state.is_linked() {
         306  +
            return;
         307  +
        }
         308  +
        let eligibility_group = record.eligibility_group.clone();
         309  +
        let reclaim = self.reclaim_order.push_back(*supplier);
         310  +
        let group = self
         311  +
            .suppliers_by_group
         312  +
            .entry(eligibility_group)
         313  +
            .or_default()
         314  +
            .push_back(*supplier);
         315  +
         316  +
        if let Some(previous) = reclaim.previous {
         317  +
            let H1SupplyIndexState::Linked { reclaim, .. } = &mut self
         318  +
                .records
         319  +
                .get_mut(&previous)
         320  +
                .expect("reclaim order tail disappeared")
         321  +
                .index_state
         322  +
            else {
         323  +
                unreachable!("reclaim order tail was unlinked");
         324  +
            };
         325  +
            reclaim.next = Some(*supplier);
         326  +
        }
         327  +
        if let Some(previous) = group.previous {
         328  +
            let H1SupplyIndexState::Linked { group, .. } = &mut self
         329  +
                .records
         330  +
                .get_mut(&previous)
         331  +
                .expect("group supplier order tail disappeared")
         332  +
                .index_state
         333  +
            else {
         334  +
                unreachable!("group supplier order tail was unlinked");
         335  +
            };
         336  +
            group.next = Some(*supplier);
         337  +
        }
         338  +
        self.records
         339  +
            .get_mut(supplier)
         340  +
            .expect("linked H1 supplier disappeared")
         341  +
            .index_state = H1SupplyIndexState::Linked { reclaim, group };
         342  +
    }
         343  +
         344  +
    fn unlink_supplier(&mut self, supplier: &PartitionId) {
         345  +
        let Some(record) = self.records.get_mut(supplier) else {
         346  +
            return;
         347  +
        };
         348  +
        let index_state = std::mem::take(&mut record.index_state);
         349  +
        let H1SupplyIndexState::Linked { reclaim, group } = index_state else {
         350  +
            return;
         351  +
        };
         352  +
        let eligibility_group = record.eligibility_group.clone();
         353  +
        self.repair_reclaim_links(supplier, &reclaim);
         354  +
        self.reclaim_order.remove(*supplier, reclaim);
         355  +
        self.repair_group_links(supplier, &group);
         356  +
        let order = self
         357  +
            .suppliers_by_group
         358  +
            .get_mut(&eligibility_group)
         359  +
            .expect("linked H1 supplier lost its group order");
         360  +
        order.remove(*supplier, group);
         361  +
    }
         362  +
         363  +
    fn repair_reclaim_links(
         364  +
        &mut self,
         365  +
        supplier: &PartitionId,
         366  +
        links: &IntrusiveLinks<PartitionId>,
         367  +
    ) {
         368  +
        if let Some(previous) = links.previous {
         369  +
            let H1SupplyIndexState::Linked { reclaim, .. } = &mut self
         370  +
                .records
         371  +
                .get_mut(&previous)
         372  +
                .expect("previous reclaim supplier disappeared")
         373  +
                .index_state
         374  +
            else {
         375  +
                unreachable!("previous reclaim supplier was unlinked");
         376  +
            };
         377  +
            reclaim.next = links.next;
         378  +
        }
         379  +
        if let Some(next) = links.next {
         380  +
            let H1SupplyIndexState::Linked { reclaim, .. } = &mut self
         381  +
                .records
         382  +
                .get_mut(&next)
         383  +
                .expect("next reclaim supplier disappeared")
         384  +
                .index_state
         385  +
            else {
         386  +
                unreachable!("next reclaim supplier was unlinked");
         387  +
            };
         388  +
            reclaim.previous = links.previous;
         389  +
        }
         390  +
        debug_assert_ne!(links.previous.as_ref(), Some(supplier));
         391  +
        debug_assert_ne!(links.next.as_ref(), Some(supplier));
         392  +
    }
         393  +
         394  +
    fn repair_group_links(&mut self, supplier: &PartitionId, links: &IntrusiveLinks<PartitionId>) {
         395  +
        if let Some(previous) = links.previous {
         396  +
            let H1SupplyIndexState::Linked { group, .. } = &mut self
         397  +
                .records
         398  +
                .get_mut(&previous)
         399  +
                .expect("previous group supplier disappeared")
         400  +
                .index_state
         401  +
            else {
         402  +
                unreachable!("previous group supplier was unlinked");
         403  +
            };
         404  +
            group.next = links.next;
         405  +
        }
         406  +
        if let Some(next) = links.next {
         407  +
            let H1SupplyIndexState::Linked { group, .. } = &mut self
         408  +
                .records
         409  +
                .get_mut(&next)
         410  +
                .expect("next group supplier disappeared")
         411  +
                .index_state
         412  +
            else {
         413  +
                unreachable!("next group supplier was unlinked");
         414  +
            };
         415  +
            group.previous = links.previous;
         416  +
        }
         417  +
        debug_assert_ne!(links.previous.as_ref(), Some(supplier));
         418  +
        debug_assert_ne!(links.next.as_ref(), Some(supplier));
         419  +
    }
         420  +
         421  +
    #[cfg(any(debug_assertions, test))]
         422  +
    fn assert_consistent(&self) {
         423  +
        for record in self.records.values() {
         424  +
            assert_eq!(
         425  +
                record.index_state.is_linked(),
         426  +
                record.is_selectable(),
         427  +
                "H1 supplier index state did not match selectability"
         428  +
            );
         429  +
        }
         430  +
        self.assert_order(&self.reclaim_order, None, true);
         431  +
        for (group, order) in &self.suppliers_by_group {
         432  +
            self.assert_order(order, Some(group), false);
         433  +
        }
         434  +
    }
         435  +
         436  +
    #[cfg(any(debug_assertions, test))]
         437  +
    fn assert_order(
         438  +
        &self,
         439  +
        order: &IntrusiveOrder<PartitionId>,
         440  +
        expected_group: Option<&EligibilityGroup>,
         441  +
        reclaim_order: bool,
         442  +
    ) {
         443  +
        let expected = self
         444  +
            .records
         445  +
            .values()
         446  +
            .filter(|record| {
         447  +
                record.index_state.is_linked()
         448  +
                    && expected_group.is_none_or(|group| &record.eligibility_group == group)
         449  +
            })
         450  +
            .count();
         451  +
        order.assert_consistent(
         452  +
            expected,
         453  +
            self.records.len(),
         454  +
            "HTTP/1 supplier order",
         455  +
            |supplier| {
         456  +
                let record = self
         457  +
                    .records
         458  +
                    .get(&supplier)
         459  +
                    .expect("ordered H1 supplier disappeared");
         460  +
                if let Some(group) = expected_group {
         461  +
                    assert_eq!(&record.eligibility_group, group);
         462  +
                }
         463  +
                let H1SupplyIndexState::Linked { reclaim, group } = &record.index_state else {
         464  +
                    unreachable!("ordered H1 supplier was unlinked");
         465  +
                };
         466  +
                if reclaim_order {
         467  +
                    *reclaim
         468  +
                } else {
         469  +
                    *group
         470  +
                }
         471  +
            },
         472  +
        );
         473  +
    }
         474  +
}
         475  +
         476  +
impl H1Supply {
         477  +
    fn apply_revision(
         478  +
        &mut self,
         479  +
        supplier: PartitionId,
         480  +
        eligibility_group: EligibilityGroup,
         481  +
        revision: SupplyRevision<H1SupplyStatus>,
         482  +
    ) {
         483  +
        self.index
         484  +
            .apply_revision(supplier, eligibility_group, revision);
         485  +
        self.assert_consistent();
         486  +
    }
         487  +
         488  +
    pub(super) fn reconcile_requester(&mut self, requester: &PartitionId, demand: &DemandSchedule) {
         489  +
        let Some(match_id) = self.by_requester.get(requester).copied() else {
         490  +
            return;
         491  +
        };
         492  +
        let Some(retained) = self.matches.get_mut(&match_id) else {
         493  +
            self.by_requester.remove(requester);
         494  +
            self.assert_consistent();
         495  +
            return;
         496  +
        };
         497  +
        if demand.is_current_queued(&retained.requester, retained.demand) {
         498  +
            return;
         499  +
        }
         500  +
        retained.cancelled = true;
         501  +
        if retained.state == H1MatchState::WaitingForSender {
         502  +
            retained.state = H1MatchState::Cancelling;
         503  +
            self.cancellations.push_back(match_id);
         504  +
        }
         505  +
        self.assert_consistent();
         506  +
    }
         507  +
         508  +
    pub(super) fn prepare_match(
         509  +
        &mut self,
         510  +
        demand: &DemandSchedule,
         511  +
    ) -> Option<PreparedH1Reservation> {
         512  +
        let queued = demand.queued_head()?;
         513  +
        if self.by_requester.contains_key(&queued.requester) {
         514  +
            return None;
         515  +
        }
         516  +
        let (supplier, kind) = if queued.requirement.accepts_h1() {
         517  +
            match self
         518  +
                .index
         519  +
                .select_borrow_supplier(&queued.eligibility_group, queued.requester)
         520  +
            {
         521  +
                Some(supplier) => (supplier, H1MatchKind::BorrowSender),
         522  +
                None => (
         523  +
                    self.index.select_peer_reclaim_supplier(queued.requester)?,
         524  +
                    H1MatchKind::ReclaimCapacity,
         525  +
                ),
         526  +
            }
         527  +
        } else {
         528  +
            (
         529  +
                self.index.select_oldest_reclaim_supplier()?,
         530  +
                H1MatchKind::ReclaimCapacity,
         531  +
            )
         532  +
        };
         533  +
        let match_id = self.take_match_id();
         534  +
        let record = self
         535  +
            .index
         536  +
            .records
         537  +
            .get_mut(&supplier)
         538  +
            .expect("selected H1 supplier disappeared");
         539  +
        debug_assert!(record.reserved_by.is_none());
         540  +
        record.reserved_by = Some(match_id);
         541  +
        self.by_requester.insert(queued.requester, match_id);
         542  +
        self.matches.insert(
         543  +
            match_id,
         544  +
            H1Match {
         545  +
                supplier,
         546  +
                requester: queued.requester,
         547  +
                demand: queued.demand,
         548  +
                kind,
         549  +
                state: H1MatchState::Reserving,
         550  +
                cancelled: false,
         551  +
            },
         552  +
        );
         553  +
        self.assert_consistent();
         554  +
        Some(PreparedH1Reservation { match_id, supplier })
         555  +
    }
         556  +
         557  +
    pub(super) fn prepare_cancellation(&mut self) -> Option<PreparedH1Cancellation> {
         558  +
        while let Some(match_id) = self.cancellations.pop_front() {
         559  +
            let Some(retained) = self.matches.get(&match_id) else {
         560  +
                continue;
         561  +
            };
         562  +
            if retained.state != H1MatchState::Cancelling {
         563  +
                continue;
         564  +
            }
         565  +
            return Some(PreparedH1Cancellation {
         566  +
                match_id,
         567  +
                supplier: retained.supplier,
         568  +
            });
         569  +
        }
         570  +
        None
         571  +
    }
         572  +
         573  +
    fn settle_reservation(&mut self, match_id: H1MatchId, resolved: bool) -> Option<H1Match> {
         574  +
        let retained = self.matches.get_mut(&match_id)?;
         575  +
        if retained.state != H1MatchState::Reserving {
         576  +
            return None;
         577  +
        }
         578  +
        retained.state = if resolved {
         579  +
            H1MatchState::Resolving
         580  +
        } else if retained.cancelled {
         581  +
            H1MatchState::Cancelling
         582  +
        } else {
         583  +
            H1MatchState::WaitingForSender
         584  +
        };
         585  +
        if retained.state == H1MatchState::Cancelling {
         586  +
            self.cancellations.push_back(match_id);
         587  +
        }
         588  +
        let retained = retained.clone();
         589  +
        self.assert_consistent();
         590  +
        Some(retained)
         591  +
    }
         592  +
         593  +
    fn begin_resolution(&mut self, match_id: H1MatchId) -> Option<H1Match> {
         594  +
        let retained = self.matches.get_mut(&match_id)?;
         595  +
        if !matches!(
         596  +
            retained.state,
         597  +
            H1MatchState::WaitingForSender | H1MatchState::Resolving | H1MatchState::Cancelling
         598  +
        ) {
         599  +
            return None;
         600  +
        }
         601  +
        retained.state = H1MatchState::Resolving;
         602  +
        let retained = retained.clone();
         603  +
        self.assert_consistent();
         604  +
        Some(retained)
         605  +
    }
         606  +
         607  +
    fn settle_match(&mut self, match_id: H1MatchId, outcome: H1SupplyOutcome) -> Option<H1Match> {
         608  +
        let outcome_supplier = *outcome.supplier();
         609  +
        self.apply_h1_supply_outcome(outcome);
         610  +
        let retained = self.matches.remove(&match_id);
         611  +
        if let Some(retained) = retained.as_ref() {
         612  +
            debug_assert_eq!(retained.supplier, outcome_supplier);
         613  +
            if self.by_requester.get(&retained.requester) == Some(&match_id) {
         614  +
                self.by_requester.remove(&retained.requester);
         615  +
            }
         616  +
            if let Some(record) = self.index.records.get_mut(&retained.supplier) {
         617  +
                if record.reserved_by == Some(match_id) {
         618  +
                    record.reserved_by = None;
         619  +
                }
         620  +
            }
         621  +
        }
         622  +
        self.index.link_supplier_if_selectable(&outcome_supplier);
         623  +
        self.assert_consistent();
         624  +
        retained
         625  +
    }
         626  +
         627  +
    fn apply_h1_supply_outcome(&mut self, outcome: H1SupplyOutcome) {
         628  +
        match outcome {
         629  +
            H1SupplyOutcome::SupplierLive { supplier, revision } => {
         630  +
                let Some(eligibility_group) = self
         631  +
                    .index
         632  +
                    .records
         633  +
                    .get(&supplier)
         634  +
                    .map(|record| record.eligibility_group.clone())
         635  +
                else {
         636  +
                    return;
         637  +
                };
         638  +
                self.index
         639  +
                    .apply_revision(supplier, eligibility_group, revision);
         640  +
            }
         641  +
            H1SupplyOutcome::SupplierExpired { supplier } => {
         642  +
                self.index.unlink_supplier(&supplier);
         643  +
                self.index.records.remove(&supplier);
         644  +
            }
         645  +
        }
         646  +
    }
         647  +
         648  +
    fn take_match_id(&mut self) -> H1MatchId {
         649  +
        let value = self.next_match_id;
         650  +
        self.next_match_id = value
         651  +
            .checked_add(1)
         652  +
            .expect("HTTP/1 match identity exhausted");
         653  +
        H1MatchId(value)
         654  +
    }
         655  +
         656  +
    fn assert_consistent(&self) {
         657  +
        #[cfg(any(debug_assertions, test))]
         658  +
        {
         659  +
            if std::thread::panicking() {
         660  +
                return;
         661  +
            }
         662  +
            self.index.assert_consistent();
         663  +
            for (match_id, retained) in &self.matches {
         664  +
                assert_eq!(self.by_requester.get(&retained.requester), Some(match_id));
         665  +
                assert_eq!(
         666  +
                    self.index
         667  +
                        .records
         668  +
                        .get(&retained.supplier)
         669  +
                        .and_then(|record| record.reserved_by),
         670  +
                    Some(*match_id),
         671  +
                    "H1 supplier back-pointer did not name its retained match"
         672  +
                );
         673  +
            }
         674  +
            for (requester, match_id) in &self.by_requester {
         675  +
                assert_eq!(
         676  +
                    self.matches
         677  +
                        .get(match_id)
         678  +
                        .map(|retained| &retained.requester),
         679  +
                    Some(requester),
         680  +
                    "H1 requester index named a missing match"
         681  +
                );
         682  +
            }
         683  +
            for (supplier, record) in &self.index.records {
         684  +
                if let Some(match_id) = record.reserved_by {
         685  +
                    assert_eq!(
         686  +
                        self.matches
         687  +
                            .get(&match_id)
         688  +
                            .map(|retained| &retained.supplier),
         689  +
                        Some(supplier),
         690  +
                        "H1 supplier record named a missing match"
         691  +
                    );
         692  +
                }
         693  +
            }
         694  +
            for (index, match_id) in self.cancellations.iter().enumerate() {
         695  +
                if let Some(retained) = self.matches.get(match_id) {
         696  +
                    assert!(
         697  +
                        retained.cancelled
         698  +
                            && matches!(
         699  +
                                retained.state,
         700  +
                                H1MatchState::Cancelling | H1MatchState::Resolving
         701  +
                            ),
         702  +
                        "live H1 cancellation named a match outside cancellation or resolution"
         703  +
                    );
         704  +
                }
         705  +
                assert!(
         706  +
                    !self
         707  +
                        .cancellations
         708  +
                        .iter()
         709  +
                        .skip(index + 1)
         710  +
                        .any(|id| id == match_id),
         711  +
                    "H1 cancellation was queued more than once"
         712  +
                );
         713  +
            }
         714  +
        }
         715  +
    }
         716  +
}
         717  +
         718  +
/// Reservation crossing from admission to one supplier cell.
         719  +
pub(in crate::client::pool) struct H1ReservationAction {
         720  +
    admission: Arc<OriginAdmission>,
         721  +
    prepared: Option<PreparedH1Reservation>,
         722  +
}
         723  +
         724  +
impl H1ReservationAction {
         725  +
    pub(super) fn new(admission: Arc<OriginAdmission>, prepared: PreparedH1Reservation) -> Self {
         726  +
        Self {
         727  +
            admission,
         728  +
            prepared: Some(prepared),
         729  +
        }
         730  +
    }
         731  +
         732  +
    pub(super) fn reserve_supplier(mut self) -> Option<AdmissionAction> {
         733  +
        let prepared = self
         734  +
            .prepared
         735  +
            .take()
         736  +
            .expect("HTTP/1 reservation consumed more than once");
         737  +
        let Some(supplier) = self.admission.cell(&prepared.supplier) else {
         738  +
            return OriginAdmission::settle_h1_match(
         739  +
                &self.admission,
         740  +
                prepared.match_id,
         741  +
                H1SupplyOutcome::supplier_expired(prepared.supplier),
         742  +
            );
         743  +
        };
         744  +
        OriginCell::commit_h1_reservation(&supplier, self.admission.clone(), prepared)
         745  +
    }
         746  +
}
         747  +
         748  +
impl Drop for H1ReservationAction {
         749  +
    fn drop(&mut self) {
         750  +
        if let Some(prepared) = self.prepared.take() {
         751  +
            let outcome = h1_supply_outcome(&self.admission, &prepared.supplier, |supplier| {
         752  +
                supplier.cancel_h1_reservation(prepared.match_id)
         753  +
            });
         754  +
            let next =
         755  +
                OriginAdmission::settle_h1_match(&self.admission, prepared.match_id, outcome);
         756  +
            OriginAdmission::run_action_chain(next);
         757  +
        }
         758  +
    }
         759  +
}
         760  +
         761  +
/// Cancellation crossing for an installed supplier reservation.
         762  +
pub(in crate::client::pool) struct H1CancellationAction {
         763  +
    admission: Arc<OriginAdmission>,
         764  +
    prepared: Option<PreparedH1Cancellation>,
         765  +
}
         766  +
         767  +
impl H1CancellationAction {
         768  +
    pub(super) fn new(admission: Arc<OriginAdmission>, prepared: PreparedH1Cancellation) -> Self {
         769  +
        Self {
         770  +
            admission,
         771  +
            prepared: Some(prepared),
         772  +
        }
         773  +
    }
         774  +
         775  +
    pub(super) fn cancel_reservation(mut self) -> Option<AdmissionAction> {
         776  +
        let prepared = self
         777  +
            .prepared
         778  +
            .take()
         779  +
            .expect("HTTP/1 cancellation consumed more than once");
         780  +
        let outcome = h1_supply_outcome(&self.admission, &prepared.supplier, |supplier| {
         781  +
            supplier.cancel_h1_reservation(prepared.match_id)
         782  +
        });
         783  +
        OriginAdmission::settle_h1_match(&self.admission, prepared.match_id, outcome)
         784  +
    }
         785  +
}
         786  +
         787  +
impl Drop for H1CancellationAction {
         788  +
    fn drop(&mut self) {
         789  +
        if let Some(prepared) = self.prepared.take() {
         790  +
            let outcome = h1_supply_outcome(&self.admission, &prepared.supplier, |supplier| {
         791  +
                supplier.cancel_h1_reservation(prepared.match_id)
         792  +
            });
         793  +
            let next =
         794  +
                OriginAdmission::settle_h1_match(&self.admission, prepared.match_id, outcome);
         795  +
            OriginAdmission::run_action_chain(next);
         796  +
        }
         797  +
    }
         798  +
}
         799  +
         800  +
/// Provisional sender owned by one resolving H1 match.
         801  +
pub(in crate::client::pool) struct H1Candidate {
         802  +
    admission: Arc<OriginAdmission>,
         803  +
    match_id: H1MatchId,
         804  +
    supplier: PartitionId,
         805  +
    provisional: Option<ProvisionalH1>,
         806  +
}
         807  +
         808  +
impl H1Candidate {
         809  +
    pub(in crate::client::pool) fn new(
         810  +
        admission: Arc<OriginAdmission>,
         811  +
        match_id: H1MatchId,
         812  +
        supplier: PartitionId,
         813  +
        provisional: ProvisionalH1,
         814  +
    ) -> Self {
         815  +
        Self {
         816  +
            admission,
         817  +
            match_id,
         818  +
            supplier,
         819  +
            provisional: Some(provisional),
         820  +
        }
         821  +
    }
         822  +
         823  +
    pub(super) fn connection_id(&self) -> ConnectionId {
         824  +
        self.provisional
         825  +
            .as_ref()
         826  +
            .expect("HTTP/1 candidate consumed more than once")
         827  +
            .connection_id()
         828  +
    }
         829  +
         830  +
    pub(in crate::client::pool) fn commit(mut self) -> Result<H1Selection, Self> {
         831  +
        let Some(supplier) = self.admission.cell(&self.supplier) else {
         832  +
            return Err(self);
         833  +
        };
         834  +
        let provisional = self
         835  +
            .provisional
         836  +
            .take()
         837  +
            .expect("HTTP/1 candidate consumed more than once");
         838  +
        match OriginCell::commit_h1_match(&supplier, self.match_id, provisional) {
         839  +
            Ok(selection) => Ok(selection),
         840  +
            Err(provisional) => {
         841  +
                self.provisional = Some(provisional);
         842  +
                Err(self)
         843  +
            }
         844  +
        }
         845  +
    }
         846  +
         847  +
    /// Returns the provisional sender to its supplier without independently
         848  +
    /// running the admission action chain.
         849  +
    pub(super) fn reject(mut self) -> H1SupplyOutcome {
         850  +
        let Some(provisional) = self.provisional.take() else {
         851  +
            unreachable!("HTTP/1 candidate consumed more than once");
         852  +
        };
         853  +
        match self.admission.cell(&self.supplier) {
         854  +
            Some(supplier) => H1SupplyOutcome::supplier_live(
         855  +
                self.supplier,
         856  +
                OriginCell::reject_h1_match(&supplier, self.match_id, provisional),
         857  +
            ),
         858  +
            None => {
         859  +
                drop(provisional);
         860  +
                H1SupplyOutcome::supplier_expired(self.supplier)
         861  +
            }
         862  +
        }
         863  +
    }
         864  +
         865  +
    fn close_for_capacity(mut self) -> (H1SupplyOutcome, bool) {
         866  +
        let Some(supplier) = self.admission.cell(&self.supplier) else {
         867  +
            drop(self.provisional.take());
         868  +
            return (H1SupplyOutcome::supplier_expired(self.supplier), false);
         869  +
        };
         870  +
        let provisional = self
         871  +
            .provisional
         872  +
            .take()
         873  +
            .expect("HTTP/1 candidate consumed more than once");
         874  +
        let (revision, reclaimed) =
         875  +
            match OriginCell::reclaim_h1_candidate(&supplier, self.match_id, provisional) {
         876  +
                Ok(result) => result,
         877  +
                Err(provisional) => (
         878  +
                    OriginCell::reject_h1_match(&supplier, self.match_id, provisional),
         879  +
                    false,
         880  +
                ),
         881  +
            };
         882  +
        (
         883  +
            H1SupplyOutcome::supplier_live(self.supplier, revision),
         884  +
            reclaimed,
         885  +
        )
         886  +
    }
         887  +
}
         888  +
         889  +
impl fmt::Debug for H1Candidate {
         890  +
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         891  +
        f.debug_struct("H1Candidate")
         892  +
            .field("match_id", &self.match_id)
         893  +
            .field("supplier", &self.supplier)
         894  +
            .field("provisional", &self.provisional)
         895  +
            .finish_non_exhaustive()
         896  +
    }
         897  +
}
         898  +
         899  +
impl Drop for H1Candidate {
         900  +
    fn drop(&mut self) {
         901  +
        let Some(provisional) = self.provisional.take() else {
         902  +
            return;
         903  +
        };
         904  +
        let outcome = match self.admission.cell(&self.supplier) {
         905  +
            Some(supplier) => H1SupplyOutcome::supplier_live(
         906  +
                self.supplier,
         907  +
                OriginCell::reject_h1_match(&supplier, self.match_id, provisional),
         908  +
            ),
         909  +
            None => {
         910  +
                drop(provisional);
         911  +
                H1SupplyOutcome::supplier_expired(self.supplier)
         912  +
            }
         913  +
        };
         914  +
        let next = OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome);
         915  +
        OriginAdmission::run_action_chain(next);
         916  +
    }
         917  +
}
         918  +
         919  +
/// Exact provisional H1 sender selected to close for bounded capacity.
         920  +
pub(in crate::client::pool) struct H1CapacityReclaim {
         921  +
    admission: Arc<OriginAdmission>,
         922  +
    requester: PartitionId,
         923  +
    match_id: H1MatchId,
         924  +
    candidate: Option<H1Candidate>,
         925  +
}
         926  +
         927  +
impl H1CapacityReclaim {
         928  +
    pub(super) fn new(
         929  +
        admission: Arc<OriginAdmission>,
         930  +
        requester: PartitionId,
         931  +
        match_id: H1MatchId,
         932  +
        candidate: H1Candidate,
         933  +
    ) -> Self {
         934  +
        Self {
         935  +
            admission,
         936  +
            requester,
         937  +
            match_id,
         938  +
            candidate: Some(candidate),
         939  +
        }
         940  +
    }
         941  +
         942  +
    pub(super) fn reclaim_capacity(mut self) -> Option<AdmissionAction> {
         943  +
        let candidate = self
         944  +
            .candidate
         945  +
            .take()
         946  +
            .expect("HTTP/1 capacity reclaim consumed more than once");
         947  +
        let connection_id = candidate.connection_id();
         948  +
        let supplier = candidate.supplier;
         949  +
        let (outcome, reclaimed) = candidate.close_for_capacity();
         950  +
        if reclaimed {
         951  +
            tracing::trace!(
         952  +
                connection_id = %connection_id,
         953  +
                request_partition = ?self.requester,
         954  +
                connection_partition = ?supplier,
         955  +
                origin_scheme = %self.admission.origin().scheme(),
         956  +
                origin_host = self.admission.origin().host(),
         957  +
                origin_port = ?self.admission.origin().port(),
         958  +
                "HTTP/1 connection reclaimed for peer demand"
         959  +
            );
         960  +
        }
         961  +
        OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome)
         962  +
    }
         963  +
}
         964  +
         965  +
/// Supplier-cell settlement after an irreversible sender transfer.
         966  +
pub(in crate::client::pool) struct H1SupplierSettlement {
         967  +
    admission: Arc<OriginAdmission>,
         968  +
    match_id: H1MatchId,
         969  +
    supplier: PartitionId,
         970  +
    transferred: bool,
         971  +
    active: bool,
         972  +
}
         973  +
         974  +
impl H1SupplierSettlement {
         975  +
    pub(super) fn new(
         976  +
        admission: Arc<OriginAdmission>,
         977  +
        match_id: H1MatchId,
         978  +
        supplier: PartitionId,
         979  +
        transferred: bool,
         980  +
    ) -> Self {
         981  +
        Self {
         982  +
            admission,
         983  +
            match_id,
         984  +
            supplier,
         985  +
            transferred,
         986  +
            active: true,
         987  +
        }
         988  +
    }
         989  +
         990  +
    pub(super) fn settle_supplier(mut self) -> Option<AdmissionAction> {
         991  +
        let outcome = h1_supply_outcome(&self.admission, &self.supplier, |supplier| {
         992  +
            supplier.complete_h1_match(self.match_id, self.transferred)
         993  +
        });
         994  +
        let action = OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome);
         995  +
        self.active = false;
         996  +
        action
         997  +
    }
         998  +
}
         999  +
        1000  +
impl Drop for H1SupplierSettlement {
        1001  +
    fn drop(&mut self) {
        1002  +
        if !self.active {
        1003  +
            return;
        1004  +
        }
        1005  +
        let outcome = h1_supply_outcome(&self.admission, &self.supplier, |supplier| {
        1006  +
            supplier.complete_h1_match(self.match_id, self.transferred)
        1007  +
        });
        1008  +
        let next = OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome);
        1009  +
        OriginAdmission::run_action_chain(next);
        1010  +
    }
        1011  +
}
        1012  +
        1013  +
fn h1_supply_outcome(
        1014  +
    admission: &OriginAdmission,
        1015  +
    supplier: &PartitionId,
        1016  +
    current: impl FnOnce(&Arc<OriginCell>) -> SupplyRevision<H1SupplyStatus>,
        1017  +
) -> H1SupplyOutcome {
        1018  +
    match admission.cell(supplier) {
        1019  +
        Some(cell) => H1SupplyOutcome::supplier_live(*supplier, current(&cell)),
        1020  +
        None => H1SupplyOutcome::supplier_expired(*supplier),
        1021  +
    }
        1022  +
}
        1023  +
        1024  +
pub(super) fn apply_supply_revision(
        1025  +
    admission: &Arc<OriginAdmission>,
        1026  +
    supplier: PartitionId,
        1027  +
    eligibility_group: EligibilityGroup,
        1028  +
    revision: SupplyRevision<H1SupplyStatus>,
        1029  +
) {
        1030  +
    let action = {
        1031  +
        let mut state = admission.state.lock();
        1032  +
        state
        1033  +
            .h1_supply
        1034  +
            .apply_revision(supplier, eligibility_group, revision);
        1035  +
        OriginAdmission::prepare_action(admission, &mut state)
        1036  +
    };
        1037  +
    OriginAdmission::run_action_chain(action);
        1038  +
}
        1039  +
        1040  +
pub(super) fn reject_returned_match(
        1041  +
    admission: &Arc<OriginAdmission>,
        1042  +
    match_id: H1MatchId,
        1043  +
    supplier: PartitionId,
        1044  +
    revision: SupplyRevision<H1SupplyStatus>,
        1045  +
) {
        1046  +
    let action = settle_match(
        1047  +
        admission,
        1048  +
        match_id,
        1049  +
        H1SupplyOutcome::supplier_live(supplier, revision),
        1050  +
    );
        1051  +
    OriginAdmission::run_action_chain(action);
        1052  +
}
        1053  +
        1054  +
pub(super) fn settle_reservation(
        1055  +
    admission: &Arc<OriginAdmission>,
        1056  +
    match_id: H1MatchId,
        1057  +
    supplier: PartitionId,
        1058  +
    decision: H1ReservationDecision<H1Candidate>,
        1059  +
) -> Option<AdmissionAction> {
        1060  +
    match decision {
        1061  +
        H1ReservationDecision::Rejected(revision) => settle_match(
        1062  +
            admission,
        1063  +
            match_id,
        1064  +
            H1SupplyOutcome::supplier_live(supplier, revision),
        1065  +
        ),
        1066  +
        H1ReservationDecision::Installed => {
        1067  +
            let mut state = admission.state.lock();
        1068  +
            state.h1_supply.settle_reservation(match_id, false);
        1069  +
            OriginAdmission::prepare_action(admission, &mut state)
        1070  +
        }
        1071  +
        H1ReservationDecision::Candidate(candidate) => {
        1072  +
            {
        1073  +
                let mut state = admission.state.lock();
        1074  +
                if state.h1_supply.settle_reservation(match_id, true).is_none() {
        1075  +
                    drop(state);
        1076  +
                    return reject_candidate(admission, match_id, candidate);
        1077  +
                }
        1078  +
            }
        1079  +
            resolve_match(admission, match_id, candidate)
        1080  +
        }
        1081  +
    }
        1082  +
}
        1083  +
        1084  +
pub(super) fn resolve_match(
        1085  +
    admission: &Arc<OriginAdmission>,
        1086  +
    match_id: H1MatchId,
        1087  +
    candidate: H1Candidate,
        1088  +
) -> Option<AdmissionAction> {
        1089  +
    let mut state = admission.state.lock();
        1090  +
    let Some(retained) = state.h1_supply.begin_resolution(match_id) else {
        1091  +
        drop(state);
        1092  +
        return reject_candidate(admission, match_id, candidate);
        1093  +
    };
        1094  +
    if retained.cancelled
        1095  +
        || !state
        1096  +
            .demand
        1097  +
            .is_current_queued(&retained.requester, retained.demand)
        1098  +
    {
        1099  +
        drop(state);
        1100  +
        return reject_candidate(admission, match_id, candidate);
        1101  +
    }
        1102  +
    match retained.kind {
        1103  +
        H1MatchKind::BorrowSender => {
        1104  +
            let assignment_id = state.take_assignment_id();
        1105  +
            let old_group = state.demand.group_for(&retained.requester);
        1106  +
            let Some(assignment) = state.demand.prepare_h1_assignment(
        1107  +
                &retained.requester,
        1108  +
                retained.demand,
        1109  +
                assignment_id,
        1110  +
            ) else {
        1111  +
                drop(state);
        1112  +
                return reject_candidate(admission, match_id, candidate);
        1113  +
            };
        1114  +
            state.reconcile_demand_indexes(&assignment.requester, old_group);
        1115  +
            Some(AdmissionAction::Deliver(DeliveryGuard::borrowed_h1(
        1116  +
                admission.clone(),
        1117  +
                assignment,
        1118  +
                match_id,
        1119  +
                retained.supplier,
        1120  +
                candidate,
        1121  +
            )))
        1122  +
        }
        1123  +
        H1MatchKind::ReclaimCapacity => Some(AdmissionAction::ReclaimCapacity(
        1124  +
            super::CapacityReclaim::FromH1(H1CapacityReclaim::new(
        1125  +
                admission.clone(),
        1126  +
                retained.requester,
        1127  +
                match_id,
        1128  +
                candidate,
        1129  +
            )),
        1130  +
        )),
        1131  +
    }
        1132  +
}
        1133  +
        1134  +
pub(super) fn settle_match(
        1135  +
    admission: &Arc<OriginAdmission>,
        1136  +
    match_id: H1MatchId,
        1137  +
    outcome: H1SupplyOutcome,
        1138  +
) -> Option<AdmissionAction> {
        1139  +
    let mut state = admission.state.lock();
        1140  +
    state.h1_supply.settle_match(match_id, outcome);
        1141  +
    OriginAdmission::prepare_action(admission, &mut state)
        1142  +
}
        1143  +
        1144  +
/// Returns a provisional sender and keeps successor work in the caller's
        1145  +
/// action chain instead of re-entering it through `H1Candidate::drop`.
        1146  +
fn reject_candidate(
        1147  +
    admission: &Arc<OriginAdmission>,
        1148  +
    match_id: H1MatchId,
        1149  +
    candidate: H1Candidate,
        1150  +
) -> Option<AdmissionAction> {
        1151  +
    let outcome = candidate.reject();
        1152  +
    settle_match(admission, match_id, outcome)
        1153  +
}
        1154  +
        1155  +
pub(super) fn settle_borrow_delivery(
        1156  +
    admission: &Arc<OriginAdmission>,
        1157  +
    match_id: H1MatchId,
        1158  +
    assignment: &super::DemandAssignment,
        1159  +
    outcome: DemandAssignmentOutcome,
        1160  +
    transferred_supplier: Option<PartitionId>,
        1161  +
    refused_outcome: Option<H1SupplyOutcome>,
        1162  +
) -> Option<AdmissionAction> {
        1163  +
    let mut state = admission.state.lock();
        1164  +
    state.settle_assignment(assignment, outcome);
        1165  +
    if let Some(supplier) = transferred_supplier {
        1166  +
        return Some(AdmissionAction::SettleH1Supplier(
        1167  +
            H1SupplierSettlement::new(admission.clone(), match_id, supplier, true),
        1168  +
        ));
        1169  +
    }
        1170  +
    state.h1_supply.settle_match(
        1171  +
        match_id,
        1172  +
        refused_outcome.expect("refused H1 borrow had no supplier outcome"),
        1173  +
    );
        1174  +
    OriginAdmission::prepare_action(admission, &mut state)
        1175  +
}
        1176  +
        1177  +
#[cfg(all(test, not(smithy_http_client_loom)))]
        1178  +
mod tests {
        1179  +
    use super::*;
        1180  +
    use crate::client::pool::admission::{DemandSnapshot, ProtocolRequirement, SnapshotVersion};
        1181  +
    use crate::client::pool::partition::PartitionId;
        1182  +
        1183  +
    fn cell(index: usize) -> PartitionId {
        1184  +
        PartitionId::from_index(index)
        1185  +
    }
        1186  +
        1187  +
    fn schedule(requesting_partition: PartitionId, group: EligibilityGroup) -> DemandSchedule {
        1188  +
        let mut schedule = DemandSchedule::default();
        1189  +
        schedule.apply_snapshot(
        1190  +
            requesting_partition,
        1191  +
            DemandSnapshot::active(
        1192  +
                DemandId::from_u64(1),
        1193  +
                SnapshotVersion::INITIAL,
        1194  +
                ProtocolRequirement::H1Compatible,
        1195  +
                group,
        1196  +
            ),
        1197  +
        );
        1198  +
        schedule
        1199  +
    }
        1200  +
        1201  +
    fn supply_revision(
        1202  +
        revision: u64,
        1203  +
        has_returnable_connection: bool,
        1204  +
        peer_use_blocked: bool,
        1205  +
    ) -> SupplyRevision<H1SupplyStatus> {
        1206  +
        SupplyRevision::new(
        1207  +
            revision,
        1208  +
            H1SupplyStatus {
        1209  +
                has_returnable_connection,
        1210  +
                peer_use_blocked,
        1211  +
            },
        1212  +
        )
        1213  +
    }
        1214  +
        1215  +
    #[test]
        1216  +
    fn supply_revision_keeps_one_bounded_residence() {
        1217  +
        let supplier = cell(1);
        1218  +
        let group = EligibilityGroup::Pool;
        1219  +
        let mut supply = H1Supply::default();
        1220  +
        1221  +
        for revision in 1..=500 {
        1222  +
            supply.apply_revision(
        1223  +
                supplier,
        1224  +
                group.clone(),
        1225  +
                supply_revision(revision, true, false),
        1226  +
            );
        1227  +
        }
        1228  +
        1229  +
        assert_eq!(1, supply.index.records.len());
        1230  +
        assert_eq!(1, supply.index.reclaim_order.len());
        1231  +
        assert_eq!(
        1232  +
            1,
        1233  +
            supply
        1234  +
                .index
        1235  +
                .suppliers_by_group
        1236  +
                .get(&group)
        1237  +
                .expect("supplier group was not indexed")
        1238  +
                .len()
        1239  +
        );
        1240  +
        1241  +
        supply.apply_revision(supplier, group.clone(), supply_revision(501, false, false));
        1242  +
        assert_eq!(0, supply.index.reclaim_order.len());
        1243  +
        assert_eq!(
        1244  +
            0,
        1245  +
            supply
        1246  +
                .index
        1247  +
                .suppliers_by_group
        1248  +
                .get(&group)
        1249  +
                .expect("supplier group disappeared")
        1250  +
                .len()
        1251  +
        );
        1252  +
    }
        1253  +
        1254  +
    #[test]
        1255  +
    fn h2_required_demand_reclaims_local_h1_capacity() {
        1256  +
        let requesting_partition = cell(1);
        1257  +
        let group = EligibilityGroup::Pool;
        1258  +
        let mut schedule = DemandSchedule::default();
        1259  +
        schedule.apply_snapshot(
        1260  +
            requesting_partition,
        1261  +
            DemandSnapshot::active(
        1262  +
                DemandId::from_u64(1),
        1263  +
                SnapshotVersion::INITIAL,
        1264  +
                ProtocolRequirement::H2Required,
        1265  +
                group.clone(),
        1266  +
            ),
        1267  +
        );
        1268  +
        let mut supply = H1Supply::default();
        1269  +
        supply.apply_revision(requesting_partition, group, supply_revision(1, true, false));
        1270  +
        1271  +
        let prepared = supply
        1272  +
            .prepare_match(&schedule)
        1273  +
            .expect("local HTTP/1 capacity was not selected for reclaim");
        1274  +
        1275  +
        assert_eq!(requesting_partition, prepared.supplier);
        1276  +
        assert_eq!(
        1277  +
            H1MatchKind::ReclaimCapacity,
        1278  +
            supply.matches[&prepared.match_id].kind
        1279  +
        );
        1280  +
    }
        1281  +
        1282  +
    #[test]
        1283  +
    fn borrow_supplier_selection_skips_the_requesting_cell() {
        1284  +
        let requesting_partition = cell(1);
        1285  +
        let peer = cell(2);
        1286  +
        let group = EligibilityGroup::Pool;
        1287  +
        let mut supply = H1Supply::default();
        1288  +
        supply.apply_revision(
        1289  +
            requesting_partition,
        1290  +
            group.clone(),
        1291  +
            supply_revision(1, true, false),
        1292  +
        );
        1293  +
        supply.apply_revision(peer, group.clone(), supply_revision(1, true, false));
        1294  +
        1295  +
        let prepared = supply
        1296  +
            .prepare_match(&schedule(requesting_partition, group))
        1297  +
            .expect("peer connection cell was not selected");
        1298  +
        assert_eq!(peer, prepared.supplier);
        1299  +
        assert_ne!(requesting_partition, prepared.supplier);
        1300  +
    }
        1301  +
        1302  +
    #[test]
        1303  +
    fn resolving_match_discards_its_lazy_cancellation_entry() {
        1304  +
        let supplier = cell(1);
        1305  +
        let requester = cell(2);
        1306  +
        let group = EligibilityGroup::Pool;
        1307  +
        let mut schedule = schedule(requester, group.clone());
        1308  +
        let mut supply = H1Supply::default();
        1309  +
        supply.apply_revision(supplier, group.clone(), supply_revision(1, true, false));
        1310  +
        let prepared = supply
        1311  +
            .prepare_match(&schedule)
        1312  +
            .expect("supplier cell did not produce a retained match");
        1313  +
        supply.settle_reservation(prepared.match_id, false);
        1314  +
        1315  +
        schedule.apply_snapshot(
        1316  +
            requester,
        1317  +
            DemandSnapshot::inactive(DemandId::from_u64(1), SnapshotVersion::INITIAL.next()),
        1318  +
        );
        1319  +
        supply.reconcile_requester(&requester, &schedule);
        1320  +
        assert_eq!(
        1321  +
            H1MatchState::Cancelling,
        1322  +
            supply.matches[&prepared.match_id].state
        1323  +
        );
        1324  +
        1325  +
        supply.begin_resolution(prepared.match_id);
        1326  +
        assert_eq!(
        1327  +
            H1MatchState::Resolving,
        1328  +
            supply.matches[&prepared.match_id].state
        1329  +
        );
        1330  +
        assert!(
        1331  +
            supply.prepare_cancellation().is_none(),
        1332  +
            "resolving match produced stale cancellation work"
        1333  +
        );
        1334  +
        assert!(supply.cancellations.is_empty());
        1335  +
    }
        1336  +
        1337  +
    #[test]
        1338  +
    fn expired_supplier_cell_terminates_its_match_without_resubmission() {
        1339  +
        let connection_partition = cell(1);
        1340  +
        let requesting_partition = cell(2);
        1341  +
        let group = EligibilityGroup::Pool;
        1342  +
        let schedule = schedule(requesting_partition, group.clone());
        1343  +
        let mut supply = H1Supply::default();
        1344  +
        supply.apply_revision(connection_partition, group, supply_revision(1, true, false));
        1345  +
        let prepared = supply
        1346  +
            .prepare_match(&schedule)
        1347  +
            .expect("supplier cell did not produce a retained match");
        1348  +
        1349  +
        supply.settle_match(
        1350  +
            prepared.match_id,
        1351  +
            H1SupplyOutcome::supplier_expired(connection_partition),
        1352  +
        );
        1353  +
        1354  +
        assert!(!supply.index.records.contains_key(&connection_partition));
        1355  +
        assert!(supply.prepare_match(&schedule).is_none());
        1356  +
        assert!(supply.matches.is_empty());
        1357  +
    }
        1358  +
        1359  +
    #[test]
        1360  +
    fn stale_supply_outcome_cannot_hide_newer_status() {
        1361  +
        let connection_partition = cell(1);
        1362  +
        let requesting_partition = cell(2);
        1363  +
        let group = EligibilityGroup::Pool;
        1364  +
        let schedule = schedule(requesting_partition, group.clone());
        1365  +
        let mut supply = H1Supply::default();
        1366  +
        supply.apply_revision(connection_partition, group, supply_revision(1, true, false));
        1367  +
        let prepared = supply
        1368  +
            .prepare_match(&schedule)
        1369  +
            .expect("supplier cell did not produce a retained match");
        1370  +
        1371  +
        supply.settle_match(
        1372  +
            prepared.match_id,
        1373  +
            H1SupplyOutcome::supplier_live(connection_partition, supply_revision(3, true, false)),
        1374  +
        );
        1375  +
        supply.settle_match(
        1376  +
            prepared.match_id,
        1377  +
            H1SupplyOutcome::supplier_live(connection_partition, supply_revision(2, false, false)),
        1378  +
        );
        1379  +
        1380  +
        let supplier = supply
        1381  +
            .index
        1382  +
            .records
        1383  +
            .get(&connection_partition)
        1384  +
            .expect("stale revision removed the supplier cell");
        1385  +
        assert!(supplier.status.has_returnable_connection);
        1386  +
        assert!(supplier.index_state.is_linked());
        1387  +
        assert_eq!(3, supplier.revision);
        1388  +
    }
        1389  +
        1390  +
    #[test]
        1391  +
    fn duplicate_supply_outcome_still_refreshes_status() {
        1392  +
        let connection_partition = cell(1);
        1393  +
        let requesting_partition = cell(2);
        1394  +
        let group = EligibilityGroup::Pool;
        1395  +
        let schedule = schedule(requesting_partition, group.clone());
        1396  +
        let mut supply = H1Supply::default();
        1397  +
        supply.apply_revision(connection_partition, group, supply_revision(1, true, false));
        1398  +
        let prepared = supply
        1399  +
            .prepare_match(&schedule)
        1400  +
            .expect("supplier cell did not produce a retained match");
        1401  +
        1402  +
        supply.settle_match(
        1403  +
            prepared.match_id,
        1404  +
            H1SupplyOutcome::supplier_live(connection_partition, supply_revision(2, true, false)),
        1405  +
        );
        1406  +
        supply.settle_match(
        1407  +
            prepared.match_id,
        1408  +
            H1SupplyOutcome::supplier_live(connection_partition, supply_revision(3, false, false)),
        1409  +
        );
        1410  +
        1411  +
        let supplier = supply
        1412  +
            .index
        1413  +
            .records
        1414  +
            .get(&connection_partition)
        1415  +
            .expect("live supply revision removed the supplier cell");
        1416  +
        assert!(!supplier.status.has_returnable_connection);
        1417  +
        assert!(!supplier.index_state.is_linked());
        1418  +
    }
        1419  +
}