AWS SDK

AWS SDK

rev. 01140188efe3be26ec6653bc30094c80ed1a1b99 (ignoring whitespace)

Files changed:

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

@@ -0,1 +0,1220 @@
           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/2 connection supply for one bounded origin.
           7  +
//!
           8  +
//! A partition may send a request on another partition's HTTP/2 connection
           9  +
//! when both partitions belong to the same eligibility group. The connection
          10  +
//! cell reports the exact accepting generation to admission. Admission pairs
          11  +
//! that report with compatible demand and installs an identity-only route in
          12  +
//! the requesting cell. The connection cell keeps the Hyper request handle,
          13  +
//! protocol driver, transport, and capacity lease.
          14  +
//!
          15  +
//! ```text
          16  +
//! supplier cell A:   submit accepting generation G
          17  +
//! admission:         demand from B + supply(A, G) -> assign demand
          18  +
//! connection cell A: validate that G still accepts requests
          19  +
//! requesting cell B: install route(A, G)
          20  +
//! admission:         settle the assigned demand
          21  +
//! ```
          22  +
//!
          23  +
//! The route carries no connection state. Each use upgrades the connection
          24  +
//! cell reference and revalidates generation `G` before reserving one request
          25  +
//! stream. A stale generation, cancelled requesting cell, or rejected route
          26  +
//! settles the assignment so live demand remains schedulable.
          27  +
//!
          28  +
//! Idle accepting generations have a second use under an origin connection
          29  +
//! limit. H1-required demand may reserve one exact idle generation, validate
          30  +
//! that it is still idle, and close it to return capacity for an HTTP/1
          31  +
//! connection attempt. A busy generation never enters this reclaim order.
          32  +
//!
          33  +
//! Admission, connection-cell, and requesting-cell locks are acquired in
          34  +
//! separate steps. [`H2RouteGuard`] owns the demand assignment while route
          35  +
//! installation crosses those steps and settles it on drop.
          36  +
//! [`H2CapacityReclaim`] provides the fallback for idle-generation reclaim.
          37  +
          38  +
use super::{
          39  +
    AdmissionAction, DemandAssignment, DemandAssignmentId, DemandAssignmentOutcome, DemandId,
          40  +
    DemandSchedule, IntrusiveLinks, IntrusiveOrder, OriginAdmission, SupplyRevision,
          41  +
};
          42  +
use crate::client::pool::cell::h2::{H2GenerationId, H2Route};
          43  +
use crate::client::pool::cell::OriginCell;
          44  +
use crate::client::pool::partition::{EligibilityGroup, PartitionId};
          45  +
use crate::sync::Arc;
          46  +
use std::collections::{BTreeSet, HashMap};
          47  +
use std::ops::Bound::{Excluded, Unbounded};
          48  +
          49  +
/// Admission-facing HTTP/2 status derived under one cell lock.
          50  +
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
          51  +
pub(in crate::client::pool) enum H2SupplyStatus {
          52  +
    Unavailable,
          53  +
    Accepting {
          54  +
        generation: H2GenerationId,
          55  +
        idle: bool,
          56  +
    },
          57  +
}
          58  +
          59  +
/// Admission-owned H2 supply index and exact idle-reclaim reservation.
          60  +
///
          61  +
/// At every completed transition:
          62  +
///
          63  +
/// - each connection cell has at most one retained supply record;
          64  +
/// - each accepting connection cell is linked once in its eligibility-group order;
          65  +
/// - each idle accepting generation is linked once in the origin reclaim
          66  +
///   order unless it owns the active reclaim reservation;
          67  +
/// - each linked identity names the newest reported generation;
          68  +
/// - `ready_groups` contains exactly the groups with H2-compatible queued
          69  +
///   demand and peer supply;
          70  +
/// - at most one reclaim reservation crosses to a connection cell; and
          71  +
/// - supply records, prepared routes, and reclaim reservations own no
          72  +
///   connection capacity or protocol sender.
          73  +
#[derive(Debug, Default)]
          74  +
pub(super) struct H2Supply {
          75  +
    index: H2SupplyIndex,
          76  +
    /// Exact idle generation currently crossing to its connection cell.
          77  +
    reclaiming: Option<PreparedH2Reclaim>,
          78  +
}
          79  +
          80  +
#[derive(Debug, Default)]
          81  +
struct H2SupplyIndex {
          82  +
    records: HashMap<PartitionId, H2SupplyRecord>,
          83  +
    suppliers_by_group: HashMap<EligibilityGroup, IntrusiveOrder<PartitionId>>,
          84  +
    reclaim_order: IntrusiveOrder<PartitionId>,
          85  +
    route_ready_groups: BTreeSet<EligibilityGroup>,
          86  +
    last_route_group: Option<EligibilityGroup>,
          87  +
}
          88  +
          89  +
/// Admission's latest complete report for one connection cell.
          90  +
#[derive(Debug)]
          91  +
struct H2SupplyRecord {
          92  +
    /// Reuse group in which this connection may be advertised.
          93  +
    eligibility_group: EligibilityGroup,
          94  +
    /// Newest connection-cell report retained by admission.
          95  +
    revision: u64,
          96  +
    /// Exact accepting generation named by the report.
          97  +
    status: H2SupplyStatus,
          98  +
    /// Intrusive-order residence for an available generation.
          99  +
    group_index: H2GroupIndexState,
         100  +
    /// Links in the origin-wide idle-generation order.
         101  +
    reclaim_links: Option<IntrusiveLinks<PartitionId>>,
         102  +
}
         103  +
         104  +
/// Whether one connection cell occupies its eligibility-group supplier order.
         105  +
#[derive(Debug, Default)]
         106  +
enum H2GroupIndexState {
         107  +
    /// This connection cell has no generation available to peers.
         108  +
    #[default]
         109  +
    Unavailable,
         110  +
    /// The cell is linked in its eligibility-group supplier order.
         111  +
    Linked {
         112  +
        /// Intrusive links owned by this residence.
         113  +
        group: IntrusiveLinks<PartitionId>,
         114  +
    },
         115  +
}
         116  +
         117  +
impl H2GroupIndexState {
         118  +
    fn links(&self) -> Option<&IntrusiveLinks<PartitionId>> {
         119  +
        match self {
         120  +
            Self::Unavailable => None,
         121  +
            Self::Linked { group } => Some(group),
         122  +
        }
         123  +
    }
         124  +
         125  +
    fn links_mut(&mut self) -> &mut IntrusiveLinks<PartitionId> {
         126  +
        match self {
         127  +
            Self::Unavailable => panic!("unavailable H2 connection had supplier-order links"),
         128  +
            Self::Linked { group } => group,
         129  +
        }
         130  +
    }
         131  +
}
         132  +
         133  +
/// Identity selected before connection and requesting cell validation.
         134  +
#[derive(Clone, Debug)]
         135  +
pub(super) struct PreparedH2Route {
         136  +
    pub(super) assignment: DemandAssignment,
         137  +
    /// Cell that owns the routed connection.
         138  +
    pub(super) supplier: PartitionId,
         139  +
    /// Exact accepting generation selected from retained supply.
         140  +
    pub(super) generation: H2GenerationId,
         141  +
    /// Eligibility group shared by demand and supply.
         142  +
    pub(super) eligibility_group: EligibilityGroup,
         143  +
}
         144  +
         145  +
/// Exact idle generation selected to release bounded origin capacity.
         146  +
#[derive(Clone, Debug, Eq, PartialEq)]
         147  +
pub(super) struct PreparedH2Reclaim {
         148  +
    /// Cell whose H1-required demand caused reclaim.
         149  +
    pub(super) requester: PartitionId,
         150  +
    /// Exact demand generation that caused reclaim.
         151  +
    pub(super) demand: DemandId,
         152  +
    /// Cell that owns the idle H2 generation.
         153  +
    pub(super) supplier: PartitionId,
         154  +
    /// Exact accepting generation observed idle by admission.
         155  +
    pub(super) generation: H2GenerationId,
         156  +
    /// Eligibility group retained for supply-index repair.
         157  +
    pub(super) eligibility_group: EligibilityGroup,
         158  +
}
         159  +
         160  +
/// Candidate selected from stored demand and supplier heads.
         161  +
struct H2RouteMatch {
         162  +
    /// Requesting cell selected from the group demand head.
         163  +
    requester: PartitionId,
         164  +
    /// Demand generation at that head.
         165  +
    demand: DemandId,
         166  +
    /// Peer connection cell selected from the supplier head.
         167  +
    supplier: PartitionId,
         168  +
    /// Routed generation in that connection cell.
         169  +
    generation: H2GenerationId,
         170  +
    /// Eligibility group whose turn was selected.
         171  +
    eligibility_group: EligibilityGroup,
         172  +
}
         173  +
         174  +
impl H2SupplyRecord {
         175  +
    fn generation(&self) -> Option<H2GenerationId> {
         176  +
        match self.status {
         177  +
            H2SupplyStatus::Unavailable => None,
         178  +
            H2SupplyStatus::Accepting { generation, .. } => Some(generation),
         179  +
        }
         180  +
    }
         181  +
         182  +
    fn idle(&self) -> bool {
         183  +
        matches!(self.status, H2SupplyStatus::Accepting { idle: true, .. })
         184  +
    }
         185  +
}
         186  +
         187  +
impl H2SupplyIndex {
         188  +
    fn has_route_ready_group(&self) -> bool {
         189  +
        !self.route_ready_groups.is_empty()
         190  +
    }
         191  +
         192  +
    fn apply_revision(
         193  +
        &mut self,
         194  +
        supplier: PartitionId,
         195  +
        eligibility_group: EligibilityGroup,
         196  +
        revision: SupplyRevision<H2SupplyStatus>,
         197  +
        excluded_reclaim: Option<PartitionId>,
         198  +
        demand: &DemandSchedule,
         199  +
    ) {
         200  +
        if self
         201  +
            .records
         202  +
            .get(&supplier)
         203  +
            .is_some_and(|record| record.revision >= revision.revision)
         204  +
        {
         205  +
            return;
         206  +
        }
         207  +
        let old_group = self
         208  +
            .records
         209  +
            .get(&supplier)
         210  +
            .map(|record| record.eligibility_group.clone());
         211  +
        self.unlink_supplier(&supplier);
         212  +
        let record = self
         213  +
            .records
         214  +
            .entry(supplier)
         215  +
            .or_insert_with(|| H2SupplyRecord {
         216  +
                eligibility_group: eligibility_group.clone(),
         217  +
                revision: revision.revision,
         218  +
                status: revision.status,
         219  +
                group_index: H2GroupIndexState::Unavailable,
         220  +
                reclaim_links: None,
         221  +
            });
         222  +
        record.eligibility_group = eligibility_group.clone();
         223  +
        record.revision = revision.revision;
         224  +
        record.status = revision.status;
         225  +
        if record.generation().is_some() {
         226  +
            self.link_group_supplier(supplier);
         227  +
        }
         228  +
        self.link_reclaim_if_selectable(supplier, excluded_reclaim);
         229  +
        if let Some(old_group) = old_group {
         230  +
            self.reconcile_group(&old_group, demand);
         231  +
        }
         232  +
        self.reconcile_group(&eligibility_group, demand);
         233  +
    }
         234  +
         235  +
    fn remove_exact_generation(
         236  +
        &mut self,
         237  +
        supplier: &PartitionId,
         238  +
        generation: H2GenerationId,
         239  +
        demand: &DemandSchedule,
         240  +
    ) {
         241  +
        let Some(record) = self.records.get(supplier) else {
         242  +
            return;
         243  +
        };
         244  +
        if record.generation() != Some(generation) {
         245  +
            return;
         246  +
        }
         247  +
        let group = record.eligibility_group.clone();
         248  +
        self.unlink_supplier(supplier);
         249  +
        self.records
         250  +
            .get_mut(supplier)
         251  +
            .expect("validated H2 supplier disappeared")
         252  +
            .status = H2SupplyStatus::Unavailable;
         253  +
        self.reconcile_group(&group, demand);
         254  +
    }
         255  +
         256  +
    fn reconcile_group(&mut self, group: &EligibilityGroup, demand: &DemandSchedule) {
         257  +
        let ready = demand
         258  +
            .queued_group_head(group)
         259  +
            .and_then(|queued| {
         260  +
                queued
         261  +
                    .requirement
         262  +
                    .accepts_h2()
         263  +
                    .then(|| self.first_peer_supplier(group, queued.requester))
         264  +
                    .flatten()
         265  +
            })
         266  +
            .is_some();
         267  +
        if ready {
         268  +
            self.route_ready_groups.insert(group.clone());
         269  +
        } else {
         270  +
            self.route_ready_groups.remove(group);
         271  +
        }
         272  +
    }
         273  +
         274  +
    fn select_route_match(&mut self, demand: &DemandSchedule) -> Option<H2RouteMatch> {
         275  +
        let eligibility_group = self
         276  +
            .last_route_group
         277  +
            .as_ref()
         278  +
            .and_then(|last| {
         279  +
                self.route_ready_groups
         280  +
                    .range((Excluded(last), Unbounded))
         281  +
                    .next()
         282  +
            })
         283  +
            .or_else(|| self.route_ready_groups.first())?
         284  +
            .clone();
         285  +
        let Some(queued) = demand.queued_group_head(&eligibility_group) else {
         286  +
            self.route_ready_groups.remove(&eligibility_group);
         287  +
            return None;
         288  +
        };
         289  +
        if !queued.requirement.accepts_h2() {
         290  +
            self.route_ready_groups.remove(&eligibility_group);
         291  +
            return None;
         292  +
        }
         293  +
        let Some(supplier) = self.first_peer_supplier(&eligibility_group, queued.requester) else {
         294  +
            self.route_ready_groups.remove(&eligibility_group);
         295  +
            return None;
         296  +
        };
         297  +
        let generation = self
         298  +
            .records
         299  +
            .get(&supplier)
         300  +
            .expect("selected H2 supplier disappeared")
         301  +
            .generation()
         302  +
            .expect("selected H2 supplier had no generation");
         303  +
        Some(H2RouteMatch {
         304  +
            requester: queued.requester,
         305  +
            demand: queued.demand,
         306  +
            supplier,
         307  +
            generation,
         308  +
            eligibility_group,
         309  +
        })
         310  +
    }
         311  +
         312  +
    fn select_reclaim_generation(
         313  +
        &mut self,
         314  +
    ) -> Option<(PartitionId, H2GenerationId, EligibilityGroup)> {
         315  +
        let supplier = self.reclaim_order.head()?;
         316  +
        let record = self
         317  +
            .records
         318  +
            .get(&supplier)
         319  +
            .expect("reclaimable H2 supplier disappeared");
         320  +
        let generation = record
         321  +
            .generation()
         322  +
            .expect("reclaimable H2 supplier had no generation");
         323  +
        let eligibility_group = record.eligibility_group.clone();
         324  +
        self.unlink_reclaim_supplier(&supplier);
         325  +
        Some((supplier, generation, eligibility_group))
         326  +
    }
         327  +
         328  +
    fn link_group_supplier(&mut self, supplier: PartitionId) {
         329  +
        let eligibility_group = self
         330  +
            .records
         331  +
            .get(&supplier)
         332  +
            .expect("linked H2 supplier disappeared")
         333  +
            .eligibility_group
         334  +
            .clone();
         335  +
        let group = self
         336  +
            .suppliers_by_group
         337  +
            .entry(eligibility_group)
         338  +
            .or_default()
         339  +
            .push_back(supplier);
         340  +
        if let Some(previous) = group.previous {
         341  +
            self.records
         342  +
                .get_mut(&previous)
         343  +
                .expect("previous H2 supplier disappeared")
         344  +
                .group_index
         345  +
                .links_mut()
         346  +
                .next = Some(supplier);
         347  +
        }
         348  +
        self.records
         349  +
            .get_mut(&supplier)
         350  +
            .expect("linked H2 supplier disappeared")
         351  +
            .group_index = H2GroupIndexState::Linked { group };
         352  +
    }
         353  +
         354  +
    fn link_reclaim_if_selectable(
         355  +
        &mut self,
         356  +
        supplier: PartitionId,
         357  +
        excluded_reclaim: Option<PartitionId>,
         358  +
    ) {
         359  +
        let Some(record) = self.records.get(&supplier) else {
         360  +
            return;
         361  +
        };
         362  +
        if !record.idle() || excluded_reclaim == Some(supplier) || record.reclaim_links.is_some() {
         363  +
            return;
         364  +
        }
         365  +
        let links = self.reclaim_order.push_back(supplier);
         366  +
        if let Some(previous) = links.previous {
         367  +
            self.records
         368  +
                .get_mut(&previous)
         369  +
                .expect("previous reclaimable H2 supplier disappeared")
         370  +
                .reclaim_links
         371  +
                .as_mut()
         372  +
                .expect("previous reclaimable H2 supplier lost its links")
         373  +
                .next = Some(supplier);
         374  +
        }
         375  +
        self.records
         376  +
            .get_mut(&supplier)
         377  +
            .expect("reclaimable H2 supplier disappeared")
         378  +
            .reclaim_links = Some(links);
         379  +
    }
         380  +
         381  +
    fn unlink_reclaim_supplier(&mut self, supplier: &PartitionId) {
         382  +
        let Some(record) = self.records.get_mut(supplier) else {
         383  +
            return;
         384  +
        };
         385  +
        let Some(links) = record.reclaim_links.take() else {
         386  +
            return;
         387  +
        };
         388  +
        if let Some(previous) = links.previous {
         389  +
            self.records
         390  +
                .get_mut(&previous)
         391  +
                .expect("previous reclaimable H2 supplier disappeared")
         392  +
                .reclaim_links
         393  +
                .as_mut()
         394  +
                .expect("previous reclaimable H2 supplier lost its links")
         395  +
                .next = links.next;
         396  +
        }
         397  +
        if let Some(next) = links.next {
         398  +
            self.records
         399  +
                .get_mut(&next)
         400  +
                .expect("next reclaimable H2 supplier disappeared")
         401  +
                .reclaim_links
         402  +
                .as_mut()
         403  +
                .expect("next reclaimable H2 supplier lost its links")
         404  +
                .previous = links.previous;
         405  +
        }
         406  +
        self.reclaim_order.remove(*supplier, links);
         407  +
    }
         408  +
         409  +
    fn unlink_supplier(&mut self, supplier: &PartitionId) {
         410  +
        self.unlink_reclaim_supplier(supplier);
         411  +
        let Some(record) = self.records.get_mut(supplier) else {
         412  +
            return;
         413  +
        };
         414  +
        let group_index = std::mem::take(&mut record.group_index);
         415  +
        let H2GroupIndexState::Linked { group } = group_index else {
         416  +
            return;
         417  +
        };
         418  +
        let eligibility_group = record.eligibility_group.clone();
         419  +
        if let Some(previous) = group.previous {
         420  +
            self.records
         421  +
                .get_mut(&previous)
         422  +
                .expect("previous H2 supplier disappeared")
         423  +
                .group_index
         424  +
                .links_mut()
         425  +
                .next = group.next;
         426  +
        }
         427  +
        if let Some(next) = group.next {
         428  +
            self.records
         429  +
                .get_mut(&next)
         430  +
                .expect("next H2 supplier disappeared")
         431  +
                .group_index
         432  +
                .links_mut()
         433  +
                .previous = group.previous;
         434  +
        }
         435  +
        let order = self
         436  +
            .suppliers_by_group
         437  +
            .get_mut(&eligibility_group)
         438  +
            .expect("linked H2 supplier lost its group order");
         439  +
        order.remove(*supplier, group);
         440  +
        if order.len() == 0 {
         441  +
            self.suppliers_by_group.remove(&eligibility_group);
         442  +
        }
         443  +
    }
         444  +
         445  +
    fn first_peer_supplier(
         446  +
        &self,
         447  +
        eligibility_group: &EligibilityGroup,
         448  +
        requester: PartitionId,
         449  +
    ) -> Option<PartitionId> {
         450  +
        let head = self.suppliers_by_group.get(eligibility_group)?.head()?;
         451  +
        if head != requester {
         452  +
            return Some(head);
         453  +
        }
         454  +
        self.records
         455  +
            .get(&head)
         456  +
            .expect("H2 supplier order head disappeared")
         457  +
            .group_index
         458  +
            .links()
         459  +
            .and_then(|links| links.next)
         460  +
    }
         461  +
         462  +
    #[cfg(any(debug_assertions, test))]
         463  +
    fn assert_consistent(&self, demand: &DemandSchedule, excluded_reclaim: Option<PartitionId>) {
         464  +
        for (supplier, record) in &self.records {
         465  +
            assert_eq!(
         466  +
                record.generation().is_some(),
         467  +
                record.group_index.links().is_some(),
         468  +
                "H2 group index did not match routable supply"
         469  +
            );
         470  +
            if record.group_index.links().is_some() {
         471  +
                assert!(
         472  +
                    self.suppliers_by_group
         473  +
                        .contains_key(&record.eligibility_group),
         474  +
                    "H2 supplier lost its eligibility-group order"
         475  +
                );
         476  +
            }
         477  +
            assert_eq!(
         478  +
                record.idle() && excluded_reclaim != Some(*supplier),
         479  +
                record.reclaim_links.is_some(),
         480  +
                "H2 reclaim index did not match idle supply"
         481  +
            );
         482  +
            debug_assert_ne!(
         483  +
                record.reclaim_links.as_ref().and_then(|links| links.next),
         484  +
                Some(*supplier)
         485  +
            );
         486  +
            debug_assert_ne!(
         487  +
                record.group_index.links().and_then(|links| links.next),
         488  +
                Some(*supplier)
         489  +
            );
         490  +
        }
         491  +
        for (group, order) in &self.suppliers_by_group {
         492  +
            let expected = self
         493  +
                .records
         494  +
                .values()
         495  +
                .filter(|record| {
         496  +
                    record.eligibility_group == *group && record.group_index.links().is_some()
         497  +
                })
         498  +
                .count();
         499  +
            order.assert_consistent(
         500  +
                expected,
         501  +
                self.records.len(),
         502  +
                "HTTP/2 supplier order",
         503  +
                |supplier| {
         504  +
                    *self
         505  +
                        .records
         506  +
                        .get(&supplier)
         507  +
                        .expect("ordered H2 supplier disappeared")
         508  +
                        .group_index
         509  +
                        .links()
         510  +
                        .expect("ordered H2 supplier lost its links")
         511  +
                },
         512  +
            );
         513  +
        }
         514  +
        let expected_reclaim = self
         515  +
            .records
         516  +
            .values()
         517  +
            .filter(|record| record.reclaim_links.is_some())
         518  +
            .count();
         519  +
        self.reclaim_order.assert_consistent(
         520  +
            expected_reclaim,
         521  +
            self.records.len(),
         522  +
            "HTTP/2 reclaim order",
         523  +
            |supplier| {
         524  +
                self.records
         525  +
                    .get(&supplier)
         526  +
                    .expect("ordered reclaimable H2 supplier disappeared")
         527  +
                    .reclaim_links
         528  +
                    .expect("ordered reclaimable H2 supplier lost its links")
         529  +
            },
         530  +
        );
         531  +
        let expected_ready = self
         532  +
            .suppliers_by_group
         533  +
            .keys()
         534  +
            .filter(|group| {
         535  +
                demand.queued_group_head(group).is_some_and(|queued| {
         536  +
                    queued.requirement.accepts_h2()
         537  +
                        && self.first_peer_supplier(group, queued.requester).is_some()
         538  +
                })
         539  +
            })
         540  +
            .cloned()
         541  +
            .collect::<BTreeSet<_>>();
         542  +
        assert_eq!(
         543  +
            expected_ready, self.route_ready_groups,
         544  +
            "HTTP/2 route-ready groups did not match demand and supply"
         545  +
        );
         546  +
    }
         547  +
}
         548  +
         549  +
impl H2Supply {
         550  +
    pub(super) fn has_route_ready_group(&self) -> bool {
         551  +
        self.index.has_route_ready_group()
         552  +
    }
         553  +
         554  +
    fn apply_revision(
         555  +
        &mut self,
         556  +
        supplier: PartitionId,
         557  +
        eligibility_group: EligibilityGroup,
         558  +
        revision: SupplyRevision<H2SupplyStatus>,
         559  +
        demand: &DemandSchedule,
         560  +
    ) {
         561  +
        let excluded_reclaim = self.reclaiming.as_ref().map(|reclaim| reclaim.supplier);
         562  +
        self.index.apply_revision(
         563  +
            supplier,
         564  +
            eligibility_group,
         565  +
            revision,
         566  +
            excluded_reclaim,
         567  +
            demand,
         568  +
        );
         569  +
        self.assert_consistent(demand);
         570  +
    }
         571  +
         572  +
    fn remove_exact_generation(
         573  +
        &mut self,
         574  +
        supplier: &PartitionId,
         575  +
        generation: H2GenerationId,
         576  +
        demand: &DemandSchedule,
         577  +
    ) {
         578  +
        self.index
         579  +
            .remove_exact_generation(supplier, generation, demand);
         580  +
        self.assert_consistent(demand);
         581  +
    }
         582  +
         583  +
    pub(super) fn reconcile_group(&mut self, group: &EligibilityGroup, demand: &DemandSchedule) {
         584  +
        self.index.reconcile_group(group, demand);
         585  +
    }
         586  +
         587  +
    pub(super) fn prepare_route(
         588  +
        &mut self,
         589  +
        demand: &mut DemandSchedule,
         590  +
        assignment_id: DemandAssignmentId,
         591  +
    ) -> Option<PreparedH2Route> {
         592  +
        let route_match = self.index.select_route_match(demand)?;
         593  +
        let assignment = demand.prepare_group_assignment(
         594  +
            &route_match.eligibility_group,
         595  +
            &route_match.requester,
         596  +
            route_match.demand,
         597  +
            assignment_id,
         598  +
        )?;
         599  +
        self.index.last_route_group = Some(route_match.eligibility_group.clone());
         600  +
        self.index
         601  +
            .reconcile_group(&route_match.eligibility_group, demand);
         602  +
        self.assert_consistent(demand);
         603  +
        Some(PreparedH2Route {
         604  +
            assignment,
         605  +
            supplier: route_match.supplier,
         606  +
            generation: route_match.generation,
         607  +
            eligibility_group: route_match.eligibility_group,
         608  +
        })
         609  +
    }
         610  +
         611  +
    pub(super) fn prepare_reclaim(&mut self, demand: &DemandSchedule) -> Option<PreparedH2Reclaim> {
         612  +
        if self.reclaiming.is_some() {
         613  +
            return None;
         614  +
        }
         615  +
        let queued = demand.queued_head()?;
         616  +
        if queued.requirement.accepts_h2() {
         617  +
            return None;
         618  +
        }
         619  +
        let (supplier, generation, eligibility_group) = self.index.select_reclaim_generation()?;
         620  +
        let prepared = PreparedH2Reclaim {
         621  +
            requester: queued.requester,
         622  +
            demand: queued.demand,
         623  +
            supplier,
         624  +
            generation,
         625  +
            eligibility_group,
         626  +
        };
         627  +
        self.reclaiming = Some(prepared.clone());
         628  +
        self.assert_consistent(demand);
         629  +
        Some(prepared)
         630  +
    }
         631  +
         632  +
    fn settle_reclaim(
         633  +
        &mut self,
         634  +
        prepared: &PreparedH2Reclaim,
         635  +
        revision: Option<SupplyRevision<H2SupplyStatus>>,
         636  +
        demand: &DemandSchedule,
         637  +
    ) {
         638  +
        assert_eq!(self.reclaiming.as_ref(), Some(prepared));
         639  +
        self.reclaiming = None;
         640  +
        match revision {
         641  +
            Some(revision) => self.index.apply_revision(
         642  +
                prepared.supplier,
         643  +
                prepared.eligibility_group.clone(),
         644  +
                revision,
         645  +
                None,
         646  +
                demand,
         647  +
            ),
         648  +
            None => {
         649  +
                self.index.unlink_supplier(&prepared.supplier);
         650  +
                self.index.records.remove(&prepared.supplier);
         651  +
                self.index
         652  +
                    .reconcile_group(&prepared.eligibility_group, demand);
         653  +
            }
         654  +
        }
         655  +
        self.index
         656  +
            .link_reclaim_if_selectable(prepared.supplier, None);
         657  +
        self.assert_consistent(demand);
         658  +
    }
         659  +
         660  +
    fn assert_consistent(&self, demand: &DemandSchedule) {
         661  +
        #[cfg(not(any(debug_assertions, test)))]
         662  +
        let _ = demand;
         663  +
        #[cfg(any(debug_assertions, test))]
         664  +
        {
         665  +
            if std::thread::panicking() {
         666  +
                return;
         667  +
            }
         668  +
            self.index.assert_consistent(
         669  +
                demand,
         670  +
                self.reclaiming.as_ref().map(|reclaim| reclaim.supplier),
         671  +
            );
         672  +
        }
         673  +
    }
         674  +
}
         675  +
         676  +
/// Identity-only route installation with a terminal admission fallback.
         677  +
pub(in crate::client::pool) struct H2RouteGuard {
         678  +
    /// Admission owner of the assigned demand and indexed supply.
         679  +
    admission: Arc<OriginAdmission>,
         680  +
    /// Identities revalidated across the unlocked cell transitions.
         681  +
    prepared: PreparedH2Route,
         682  +
    /// Terminal acknowledgement submitted if the crossing unwinds.
         683  +
    on_drop: Option<DemandAssignmentOutcome>,
         684  +
}
         685  +
         686  +
impl H2RouteGuard {
         687  +
    pub(super) fn new(admission: Arc<OriginAdmission>, prepared: PreparedH2Route) -> Self {
         688  +
        Self {
         689  +
            admission,
         690  +
            prepared,
         691  +
            on_drop: Some(DemandAssignmentOutcome::RetrySamePosition),
         692  +
        }
         693  +
    }
         694  +
         695  +
    /// Attaches the selected peer route, then settles the demand assignment.
         696  +
    ///
         697  +
    /// A missing connection cell or stale generation withdraws that exact
         698  +
    /// supply revision and retries the same demand residence. A missing
         699  +
    /// requesting cell retires the demand. Requesting-cell rejection retries
         700  +
    /// only when the original demand and route installation are still useful.
         701  +
    pub(super) fn attach_route(self) -> Option<AdmissionAction> {
         702  +
        let generation = self.prepared.generation;
         703  +
        let Some(connection_cell) = self.admission.cell(&self.prepared.supplier) else {
         704  +
            return self.settle(DemandAssignmentOutcome::RetrySamePosition, Some(generation));
         705  +
        };
         706  +
        if !OriginCell::h2_generation_is_accepting(&connection_cell, self.prepared.generation) {
         707  +
            return self.settle(DemandAssignmentOutcome::RetrySamePosition, Some(generation));
         708  +
        }
         709  +
         710  +
        let Some(requesting_cell) = self.admission.cell(&self.prepared.assignment.requester) else {
         711  +
            return self.settle(DemandAssignmentOutcome::Refused { successor: None }, None);
         712  +
        };
         713  +
        let route = H2Route::new(&connection_cell, self.prepared.generation);
         714  +
        if !OriginCell::attach_h2_route(
         715  +
            &requesting_cell,
         716  +
            route,
         717  +
            &self.prepared.eligibility_group,
         718  +
            self.prepared.assignment.demand,
         719  +
        ) {
         720  +
            return self.settle(DemandAssignmentOutcome::RetrySamePosition, None);
         721  +
        }
         722  +
         723  +
        let next = self.settle(DemandAssignmentOutcome::Accepted { successor: None }, None);
         724  +
        OriginCell::offer_local_h2(&requesting_cell);
         725  +
        OriginCell::offer_peer_h2(&requesting_cell);
         726  +
        next
         727  +
    }
         728  +
         729  +
    fn settle(
         730  +
        mut self,
         731  +
        outcome: DemandAssignmentOutcome,
         732  +
        stale_generation: Option<H2GenerationId>,
         733  +
    ) -> Option<AdmissionAction> {
         734  +
        let trace_outcome = match &outcome {
         735  +
            DemandAssignmentOutcome::Accepted { .. } => "accepted",
         736  +
            DemandAssignmentOutcome::RetrySamePosition => "retry",
         737  +
            DemandAssignmentOutcome::Refused { .. } => "refused",
         738  +
        };
         739  +
        // The unlocked cell crossing is complete. Admission owns the assignment
         740  +
        // again, so unwinding must not replay a failed acknowledgement.
         741  +
        self.on_drop = None;
         742  +
        let next = OriginAdmission::settle_h2_route(
         743  +
            &self.admission,
         744  +
            &self.prepared,
         745  +
            stale_generation,
         746  +
            outcome,
         747  +
        );
         748  +
        self.trace(trace_outcome);
         749  +
        next
         750  +
    }
         751  +
         752  +
    fn trace(&self, outcome: &str) {
         753  +
        tracing::trace!(
         754  +
            request_partition = ?self.prepared.assignment.requester,
         755  +
            connection_partition = ?self.prepared.supplier,
         756  +
            origin_scheme = %self.admission.origin().scheme(),
         757  +
            origin_host = self.admission.origin().host(),
         758  +
            origin_port = ?self.admission.origin().port(),
         759  +
            h2_generation = ?self.prepared.generation,
         760  +
            demand = ?self.prepared.assignment.demand,
         761  +
            outcome,
         762  +
            "HTTP/2 peer route completed"
         763  +
        );
         764  +
    }
         765  +
}
         766  +
         767  +
impl Drop for H2RouteGuard {
         768  +
    fn drop(&mut self) {
         769  +
        let Some(result) = self.on_drop.take() else {
         770  +
            return;
         771  +
        };
         772  +
        let next = OriginAdmission::settle_h2_route(&self.admission, &self.prepared, None, result);
         773  +
        self.trace("guard_drop");
         774  +
        OriginAdmission::run_action_chain(next);
         775  +
    }
         776  +
}
         777  +
         778  +
/// Exact idle-generation crossing with an admission repair fallback.
         779  +
pub(in crate::client::pool) struct H2CapacityReclaim {
         780  +
    /// Admission authority that owns the reclaim reservation.
         781  +
    admission: Arc<OriginAdmission>,
         782  +
    /// Prepared reclaim still owned by this fallback.
         783  +
    prepared: Option<PreparedH2Reclaim>,
         784  +
}
         785  +
         786  +
impl H2CapacityReclaim {
         787  +
    /// Creates one unlocked exact-generation reclaim crossing.
         788  +
    pub(super) fn new(admission: Arc<OriginAdmission>, prepared: PreparedH2Reclaim) -> Self {
         789  +
        Self {
         790  +
            admission,
         791  +
            prepared: Some(prepared),
         792  +
        }
         793  +
    }
         794  +
         795  +
    /// Attempts idle reclaim and returns the next bounded-origin action.
         796  +
    pub(super) fn reclaim_capacity(mut self) -> Option<AdmissionAction> {
         797  +
        let prepared = self
         798  +
            .prepared
         799  +
            .as_ref()
         800  +
            .expect("HTTP/2 reclaim action consumed more than once")
         801  +
            .clone();
         802  +
        let (revision, reclaimed) = match self.admission.cell(&prepared.supplier) {
         803  +
            Some(cell) => {
         804  +
                let (revision, connection_id) =
         805  +
                    OriginCell::reclaim_idle_h2(&cell, prepared.generation);
         806  +
                (Some(revision), connection_id)
         807  +
            }
         808  +
            None => (None, None),
         809  +
        };
         810  +
        self.prepared = None;
         811  +
        let next = OriginAdmission::settle_h2_reclaim(&self.admission, &prepared, revision);
         812  +
        tracing::trace!(
         813  +
            connection_id = ?reclaimed,
         814  +
            request_partition = ?prepared.requester,
         815  +
            connection_partition = ?prepared.supplier,
         816  +
            origin_scheme = %self.admission.origin().scheme(),
         817  +
            origin_host = self.admission.origin().host(),
         818  +
            origin_port = ?self.admission.origin().port(),
         819  +
            h2_generation = ?prepared.generation,
         820  +
            demand = ?prepared.demand,
         821  +
            outcome = if reclaimed.is_some() { "reclaimed" } else { "rejected" },
         822  +
            "HTTP/2 idle reclaim completed"
         823  +
        );
         824  +
        next
         825  +
    }
         826  +
}
         827  +
         828  +
impl Drop for H2CapacityReclaim {
         829  +
    fn drop(&mut self) {
         830  +
        let Some(prepared) = self.prepared.take() else {
         831  +
            return;
         832  +
        };
         833  +
        let revision = self
         834  +
            .admission
         835  +
            .cell(&prepared.supplier)
         836  +
            .map(|cell| cell.current_h2_supply_revision());
         837  +
        let next = OriginAdmission::settle_h2_reclaim(&self.admission, &prepared, revision);
         838  +
        tracing::trace!(
         839  +
            request_partition = ?prepared.requester,
         840  +
            connection_partition = ?prepared.supplier,
         841  +
            origin_scheme = %self.admission.origin().scheme(),
         842  +
            origin_host = self.admission.origin().host(),
         843  +
            origin_port = ?self.admission.origin().port(),
         844  +
            h2_generation = ?prepared.generation,
         845  +
            demand = ?prepared.demand,
         846  +
            outcome = "guard_drop",
         847  +
            "HTTP/2 idle reclaim completed"
         848  +
        );
         849  +
        OriginAdmission::run_action_chain(next);
         850  +
    }
         851  +
}
         852  +
         853  +
pub(super) fn apply_supply_revision(
         854  +
    admission: &Arc<OriginAdmission>,
         855  +
    supplier: PartitionId,
         856  +
    eligibility_group: EligibilityGroup,
         857  +
    revision: SupplyRevision<H2SupplyStatus>,
         858  +
) {
         859  +
    let action = {
         860  +
        let mut state = admission.state.lock();
         861  +
        let super::AdmissionState {
         862  +
            h2_supply, demand, ..
         863  +
        } = &mut *state;
         864  +
        h2_supply.apply_revision(supplier, eligibility_group, revision, demand);
         865  +
        OriginAdmission::prepare_action(admission, &mut state)
         866  +
    };
         867  +
    OriginAdmission::run_action_chain(action);
         868  +
}
         869  +
         870  +
pub(super) fn settle_route(
         871  +
    admission: &Arc<OriginAdmission>,
         872  +
    prepared: &PreparedH2Route,
         873  +
    stale_generation: Option<H2GenerationId>,
         874  +
    outcome: DemandAssignmentOutcome,
         875  +
) -> Option<AdmissionAction> {
         876  +
    let mut state = admission.state.lock();
         877  +
    if let Some(generation) = stale_generation {
         878  +
        let super::AdmissionState {
         879  +
            h2_supply, demand, ..
         880  +
        } = &mut *state;
         881  +
        h2_supply.remove_exact_generation(&prepared.supplier, generation, demand);
         882  +
    }
         883  +
    state.settle_assignment(&prepared.assignment, outcome);
         884  +
    OriginAdmission::prepare_action(admission, &mut state)
         885  +
}
         886  +
         887  +
pub(super) fn settle_reclaim(
         888  +
    admission: &Arc<OriginAdmission>,
         889  +
    prepared: &PreparedH2Reclaim,
         890  +
    revision: Option<SupplyRevision<H2SupplyStatus>>,
         891  +
) -> Option<AdmissionAction> {
         892  +
    let mut state = admission.state.lock();
         893  +
    let super::AdmissionState {
         894  +
        h2_supply, demand, ..
         895  +
    } = &mut *state;
         896  +
    h2_supply.settle_reclaim(prepared, revision, demand);
         897  +
    OriginAdmission::prepare_action(admission, &mut state)
         898  +
}
         899  +
         900  +
#[cfg(all(test, not(smithy_http_client_loom)))]
         901  +
mod tests {
         902  +
    use super::*;
         903  +
    use crate::client::pool::admission::{DemandSnapshot, ProtocolRequirement, SnapshotVersion};
         904  +
         905  +
    fn partition(index: usize) -> PartitionId {
         906  +
        PartitionId::from_index(index)
         907  +
    }
         908  +
         909  +
    fn generation(value: u64) -> H2GenerationId {
         910  +
        H2GenerationId::for_test(value)
         911  +
    }
         912  +
         913  +
    fn demand(id: u64, group: EligibilityGroup) -> DemandSnapshot {
         914  +
        DemandSnapshot::active(
         915  +
            DemandId::from_u64(id),
         916  +
            SnapshotVersion::INITIAL,
         917  +
            ProtocolRequirement::H2Required,
         918  +
            group,
         919  +
        )
         920  +
    }
         921  +
         922  +
    fn h1_demand(id: u64, group: EligibilityGroup) -> DemandSnapshot {
         923  +
        DemandSnapshot::active(
         924  +
            DemandId::from_u64(id),
         925  +
            SnapshotVersion::INITIAL,
         926  +
            ProtocolRequirement::H1Required,
         927  +
            group,
         928  +
        )
         929  +
    }
         930  +
         931  +
    fn accepting(revision: u64, generation: H2GenerationId) -> SupplyRevision<H2SupplyStatus> {
         932  +
        SupplyRevision::new(
         933  +
            revision,
         934  +
            H2SupplyStatus::Accepting {
         935  +
                generation,
         936  +
                idle: false,
         937  +
            },
         938  +
        )
         939  +
    }
         940  +
         941  +
    fn idle(revision: u64, generation: H2GenerationId) -> SupplyRevision<H2SupplyStatus> {
         942  +
        SupplyRevision::new(
         943  +
            revision,
         944  +
            H2SupplyStatus::Accepting {
         945  +
                generation,
         946  +
                idle: true,
         947  +
            },
         948  +
        )
         949  +
    }
         950  +
         951  +
    #[test]
         952  +
    fn exact_generation_replacement_rejects_stale_removal() {
         953  +
        let group = EligibilityGroup::Pool;
         954  +
        let connection_partition = partition(1);
         955  +
        let requesting_partition = partition(2);
         956  +
        let first = generation(10);
         957  +
        let second = generation(11);
         958  +
        let mut schedule = DemandSchedule::default();
         959  +
        schedule.apply_snapshot(requesting_partition, demand(1, group.clone()));
         960  +
        let mut supply = H2Supply::default();
         961  +
        supply.apply_revision(
         962  +
            connection_partition,
         963  +
            group.clone(),
         964  +
            accepting(1, first),
         965  +
            &schedule,
         966  +
        );
         967  +
        supply.apply_revision(
         968  +
            connection_partition,
         969  +
            group.clone(),
         970  +
            accepting(2, second),
         971  +
            &schedule,
         972  +
        );
         973  +
         974  +
        supply.remove_exact_generation(&connection_partition, first, &schedule);
         975  +
        let prepared = supply
         976  +
            .prepare_route(&mut schedule, DemandAssignmentId(1))
         977  +
            .expect("newer supply revision should remain selectable");
         978  +
        assert_eq!(second, prepared.generation);
         979  +
        assert_eq!(connection_partition, prepared.supplier);
         980  +
        assert_eq!(requesting_partition, prepared.assignment.requester);
         981  +
    }
         982  +
         983  +
    #[test]
         984  +
    fn current_generation_removal_repairs_the_connection_index() {
         985  +
        let group = EligibilityGroup::Pool;
         986  +
        let connection_partition = partition(1);
         987  +
        let requesting_partition = partition(2);
         988  +
        let current = generation(10);
         989  +
        let mut schedule = DemandSchedule::default();
         990  +
        schedule.apply_snapshot(requesting_partition, demand(1, group.clone()));
         991  +
        let mut supply = H2Supply::default();
         992  +
        supply.apply_revision(
         993  +
            connection_partition,
         994  +
            group,
         995  +
            accepting(1, current),
         996  +
            &schedule,
         997  +
        );
         998  +
         999  +
        supply.remove_exact_generation(&connection_partition, current, &schedule);
        1000  +
        1001  +
        assert!(!supply.has_route_ready_group());
        1002  +
        assert!(supply.index.suppliers_by_group.is_empty());
        1003  +
        assert!(matches!(
        1004  +
            supply.index.records[&connection_partition].group_index,
        1005  +
            H2GroupIndexState::Unavailable
        1006  +
        ));
        1007  +
    }
        1008  +
        1009  +
    #[test]
        1010  +
    fn route_selection_skips_the_requesting_cell_and_selects_a_peer_in_its_group() {
        1011  +
        let pool = EligibilityGroup::Pool;
        1012  +
        let isolated = EligibilityGroup::Partition(partition(3));
        1013  +
        let requesting_partition = partition(1);
        1014  +
        let peer = partition(2);
        1015  +
        let other_group = partition(3);
        1016  +
        let mut schedule = DemandSchedule::default();
        1017  +
        schedule.apply_snapshot(requesting_partition, demand(1, pool.clone()));
        1018  +
        schedule.apply_snapshot(other_group, demand(2, isolated.clone()));
        1019  +
        let mut supply = H2Supply::default();
        1020  +
        supply.apply_revision(
        1021  +
            requesting_partition,
        1022  +
            pool.clone(),
        1023  +
            accepting(1, generation(1)),
        1024  +
            &schedule,
        1025  +
        );
        1026  +
        supply.apply_revision(peer, pool, accepting(1, generation(2)), &schedule);
        1027  +
        supply.apply_revision(
        1028  +
            other_group,
        1029  +
            isolated,
        1030  +
            accepting(1, generation(3)),
        1031  +
            &schedule,
        1032  +
        );
        1033  +
        1034  +
        let prepared = supply
        1035  +
            .prepare_route(&mut schedule, DemandAssignmentId(1))
        1036  +
            .expect("pool demand should find its peer connection");
        1037  +
        assert_eq!(requesting_partition, prepared.assignment.requester);
        1038  +
        assert_eq!(peer, prepared.supplier);
        1039  +
        assert_eq!(generation(2), prepared.generation);
        1040  +
    }
        1041  +
        1042  +
    #[test]
        1043  +
    fn older_supply_revision_cannot_replace_the_current_generation() {
        1044  +
        let group = EligibilityGroup::Pool;
        1045  +
        let connection_partition = partition(1);
        1046  +
        let requesting_partition = partition(2);
        1047  +
        let current = generation(2);
        1048  +
        let mut schedule = DemandSchedule::default();
        1049  +
        schedule.apply_snapshot(requesting_partition, demand(1, group.clone()));
        1050  +
        let mut supply = H2Supply::default();
        1051  +
        supply.apply_revision(
        1052  +
            connection_partition,
        1053  +
            group.clone(),
        1054  +
            accepting(2, current),
        1055  +
            &schedule,
        1056  +
        );
        1057  +
        supply.apply_revision(
        1058  +
            connection_partition,
        1059  +
            group,
        1060  +
            accepting(1, generation(1)),
        1061  +
            &schedule,
        1062  +
        );
        1063  +
        1064  +
        let prepared = supply
        1065  +
            .prepare_route(&mut schedule, DemandAssignmentId(1))
        1066  +
            .expect("current supply revision should remain selectable");
        1067  +
        assert_eq!(current, prepared.generation);
        1068  +
    }
        1069  +
        1070  +
    #[test]
        1071  +
    fn h1_required_demand_is_not_route_ready() {
        1072  +
        let group = EligibilityGroup::Pool;
        1073  +
        let requesting_partition = partition(1);
        1074  +
        let connection_partition = partition(2);
        1075  +
        let mut schedule = DemandSchedule::default();
        1076  +
        schedule.apply_snapshot(
        1077  +
            requesting_partition,
        1078  +
            DemandSnapshot::active(
        1079  +
                DemandId::from_u64(1),
        1080  +
                SnapshotVersion::INITIAL,
        1081  +
                ProtocolRequirement::H1Required,
        1082  +
                group.clone(),
        1083  +
            ),
        1084  +
        );
        1085  +
        let mut supply = H2Supply::default();
        1086  +
        supply.apply_revision(
        1087  +
            connection_partition,
        1088  +
            group,
        1089  +
            accepting(1, generation(1)),
        1090  +
            &schedule,
        1091  +
        );
        1092  +
        1093  +
        assert!(!supply.has_route_ready_group());
        1094  +
        assert!(supply
        1095  +
            .prepare_route(&mut schedule, DemandAssignmentId(1))
        1096  +
            .is_none());
        1097  +
    }
        1098  +
        1099  +
    #[test]
        1100  +
    fn route_turns_rotate_across_ready_groups() {
        1101  +
        let first_group = EligibilityGroup::Partition(partition(10));
        1102  +
        let second_group = EligibilityGroup::Partition(partition(20));
        1103  +
        let first_request = partition(1);
        1104  +
        let first_connection = partition(2);
        1105  +
        let second_request = partition(3);
        1106  +
        let second_connection = partition(4);
        1107  +
        let mut schedule = DemandSchedule::default();
        1108  +
        schedule.apply_snapshot(first_request, demand(1, first_group.clone()));
        1109  +
        schedule.apply_snapshot(second_request, demand(2, second_group.clone()));
        1110  +
        let mut supply = H2Supply::default();
        1111  +
        supply.apply_revision(
        1112  +
            first_connection,
        1113  +
            first_group.clone(),
        1114  +
            accepting(1, generation(1)),
        1115  +
            &schedule,
        1116  +
        );
        1117  +
        supply.apply_revision(
        1118  +
            second_connection,
        1119  +
            second_group.clone(),
        1120  +
            accepting(1, generation(2)),
        1121  +
            &schedule,
        1122  +
        );
        1123  +
        1124  +
        let first = supply
        1125  +
            .prepare_route(&mut schedule, DemandAssignmentId(1))
        1126  +
            .expect("first ready group was not selected");
        1127  +
        schedule.settle_assignment(
        1128  +
            &first.assignment,
        1129  +
            DemandAssignmentOutcome::RetrySamePosition,
        1130  +
        );
        1131  +
        supply.reconcile_group(&first.eligibility_group, &schedule);
        1132  +
        1133  +
        let second = supply
        1134  +
            .prepare_route(&mut schedule, DemandAssignmentId(2))
        1135  +
            .expect("second ready group was not selected");
        1136  +
        assert_ne!(first.eligibility_group, second.eligibility_group);
        1137  +
    }
        1138  +
        1139  +
    #[test]
        1140  +
    #[should_panic(expected = "HTTP/2 route-ready groups did not match demand and supply")]
        1141  +
    fn consistency_check_rejects_a_missing_ready_group() {
        1142  +
        let group = EligibilityGroup::Pool;
        1143  +
        let connection_partition = partition(1);
        1144  +
        let requesting_partition = partition(2);
        1145  +
        let mut schedule = DemandSchedule::default();
        1146  +
        schedule.apply_snapshot(requesting_partition, demand(1, group.clone()));
        1147  +
        let mut supply = H2Supply::default();
        1148  +
        supply.apply_revision(
        1149  +
            connection_partition,
        1150  +
            group,
        1151  +
            accepting(1, generation(1)),
        1152  +
            &schedule,
        1153  +
        );
        1154  +
        1155  +
        supply.index.route_ready_groups.clear();
        1156  +
        1157  +
        supply.assert_consistent(&schedule);
        1158  +
    }
        1159  +
        1160  +
    #[test]
        1161  +
    fn only_idle_h2_is_selected_for_h1_required_reclaim() {
        1162  +
        let group = EligibilityGroup::Pool;
        1163  +
        let requesting_partition = partition(1);
        1164  +
        let connection_partition = partition(2);
        1165  +
        let current = generation(10);
        1166  +
        let mut schedule = DemandSchedule::default();
        1167  +
        schedule.apply_snapshot(requesting_partition, h1_demand(1, group.clone()));
        1168  +
        let mut supply = H2Supply::default();
        1169  +
        supply.apply_revision(
        1170  +
            connection_partition,
        1171  +
            group.clone(),
        1172  +
            accepting(1, current),
        1173  +
            &schedule,
        1174  +
        );
        1175  +
        1176  +
        assert!(
        1177  +
            supply.prepare_reclaim(&schedule).is_none(),
        1178  +
            "busy HTTP/2 generation entered reclaim order"
        1179  +
        );
        1180  +
        1181  +
        supply.apply_revision(connection_partition, group, idle(2, current), &schedule);
        1182  +
        let prepared = supply
        1183  +
            .prepare_reclaim(&schedule)
        1184  +
            .expect("idle HTTP/2 generation was not selected");
        1185  +
        assert_eq!(requesting_partition, prepared.requester);
        1186  +
        assert_eq!(connection_partition, prepared.supplier);
        1187  +
        assert_eq!(current, prepared.generation);
        1188  +
        assert!(supply.index.reclaim_order.head().is_none());
        1189  +
        1190  +
        supply.settle_reclaim(&prepared, Some(idle(2, current)), &schedule);
        1191  +
        assert_eq!(
        1192  +
            Some(connection_partition),
        1193  +
            supply.index.reclaim_order.head()
        1194  +
        );
        1195  +
    }
        1196  +
        1197  +
    #[test]
        1198  +
    fn reclaim_completion_cannot_restore_a_replaced_generation() {
        1199  +
        let group = EligibilityGroup::Pool;
        1200  +
        let requesting_partition = partition(1);
        1201  +
        let connection_partition = partition(2);
        1202  +
        let old = generation(10);
        1203  +
        let replacement = generation(11);
        1204  +
        let mut schedule = DemandSchedule::default();
        1205  +
        schedule.apply_snapshot(requesting_partition, h1_demand(1, group.clone()));
        1206  +
        let mut supply = H2Supply::default();
        1207  +
        supply.apply_revision(connection_partition, group.clone(), idle(1, old), &schedule);
        1208  +
        let prepared = supply
        1209  +
            .prepare_reclaim(&schedule)
        1210  +
            .expect("idle HTTP/2 generation was not selected");
        1211  +
        1212  +
        supply.apply_revision(connection_partition, group, idle(2, replacement), &schedule);
        1213  +
        supply.settle_reclaim(&prepared, Some(idle(1, old)), &schedule);
        1214  +
        1215  +
        let next = supply
        1216  +
            .prepare_reclaim(&schedule)
        1217  +
            .expect("replacement generation was not restored to reclaim order");
        1218  +
        assert_eq!(replacement, next.generation);
        1219  +
    }
        1220  +
}