AWS SDK

AWS SDK

rev. 214b1cfffb5029555908aba78574468110a9ad23

Files changed:

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/docs/design/connection-pool.md

@@ -477,477 +542,544 @@
  497    497   
The facade clones the complete non-exhaustive `HttpConnectorSettings` and extracts the operation components
  498    498   
its request policy uses. Its `call` moves a `Client` clone and that policy into the request future, which is the
  499    499   
strong pool reference already shown above. A facade may be cached as an implementation optimization, but its
  500    500   
identity is never a partition, origin, pool, or admission key. Differing connect or read timeouts therefore do
  501    501   
not multiply `max_connections_per_host`: every facade for one `Client` reaches the same origin-wide admission
  502    502   
authority.
  503    503   
  504    504   
Connect and read timeouts remain operation policy rather than connection identity. The facade uses the
  505    505   
operation's `AsyncSleep` from `RuntimeComponents` when present, then the same default-sleep fallback as the
  506    506   
existing client. A configured timeout requires the resulting sleep implementation; its absence never disables
  507         -
the timeout. The read timeout wraps `PoolConnector::call` through response headers. A connect timeout wraps
  508         -
only the transport-connector operation through TLS and ALPN for an establishment attempt owned by that
  509         -
request; the Hyper protocol handshake remains under the read timeout. A request joining an existing HTTP/2
  510         -
flight owns no connector operation, so its connect timeout does not apply to that shared flight; its read
  511         -
timeout or caller cancellation can remove that participant. If it later becomes an establishment driver, its
  512         -
own connect timeout applies to that attempt.
         507  +
the timeout. The read timeout bounds `PoolConnector::call` through response headers, including time spent
         508  +
waiting for acquisition or a shared HTTP/2 flight. A connect timeout wraps only the transport connector through
         509  +
TLS and ALPN. Once the connected transport moves to the connection-partition owner task, neither request
         510  +
timeout supervises or cancels that task's Hyper handshake; the pool has no separate handshake timeout.
         511  +
A request joining an existing HTTP/2 flight owns no connector operation, so its connect timeout does not apply
         512  +
to that flight. Its read timeout or caller cancellation removes that participant while the owner task
         513  +
continues. A fully cancelled successful flight may therefore install an idle generation that remains until
         514  +
ordinary close or idle expiration.
  513    515   
  514    516   
Idle maintenance is pool policy, not operation policy. Its `TimeSource` and `AsyncSleep` are fixed by the pool
  515    517   
builder and shared by the partition maintenance tasks; they do not depend on which operation first asks for a
  516    518   
facade. Operation `RuntimeComponents` may vary between smithy clients sharing a pool without changing idle
  517    519   
age or creating another pool.
  518    520   
  519    521   
`Client::validate_base_client_config` runs an idempotent transport preflight so native trust roots, when
  520    522   
applicable, load when this HTTP client is selected rather than on its first request. The preflight opens no
  521    523   
socket and creates no settings-keyed pool or connector cache. `validate_final_config` remains a no-op.
  522    524   
`connector_metadata` reports `hyper/1.x`, preserving the HTTP client identifier used in user-agent metadata.
@@ -583,585 +642,657 @@
  603    605   
#### Establishment
  604    606   
  605    607   
A connection's socket, driver, and reactor must all live on one runtime, because Tokio registers a socket with
  606    608   
the I/O reactor of whatever runtime creates it — it captures the current runtime's handle at socket creation
  607    609   
([`poll_evented.rs:111`](https://github.com/tokio-rs/tokio/blob/tokio-1.53.1/tokio/src/io/poll_evented.rs#L111))
  608    610   
— and the connector is what creates the socket. If the driver ran on one runtime while the socket was created
  609    611   
on another, every readiness event would cross runtimes, and the socket's reactor would outlive or predecease
  610    612   
the driver that holds it. So establishment — connector, transport, TLS, ALPN, handshake — and the driver it
  611    613   
produces run on the same runtime.
  612    614   
         615  +
Connection establishment and installed connection lifetime are separate
         616  +
ownership phases. The establishment authority exists before DNS and owns the
         617  +
connector future and any bounded-origin permit. After DNS, socket connection,
         618  +
proxy negotiation, TLS, and ALPN produce connected I/O with a selected HTTP
         619  +
protocol, the pool creates `ConnectionState` in `PendingOpen`. At that point
         620  +
transport establishment and protocol selection are complete, but Hyper has not
         621  +
produced the request handle required for dispatch. Successful Hyper protocol
         622  +
setup moves the state to `Open` and attaches the bounded-origin permit before
         623  +
cell installation makes the connection discoverable. Lifecycle observation
         624  +
treats failed DNS, transport, and TLS attempts independently from
         625  +
installed-connection events; failed attempts have no synthetic
         626  +
`ConnectionState`.
         627  +
  613    628   
An **explicit partition** names its owner runtime through `DriverSpawner`. The
  614    629   
**anonymous partition** captures the current Tokio runtime on first use. A
  615    630   
request may be polled on another runtime, so every new connection submits the
  616    631   
still-unpolled connector, transport, TLS/ALPN, and Hyper handshake future to
  617    632   
the partition owner. Completion updates the cell and wakes the requesting task;
  618    633   
the resulting driver and pending return work use the same spawner. This policy
  619    634   
costs one task submission and wake per new connection. Local reuse and dispatch
  620    635   
on an established connection do not pay that handoff.
  621    636   
  622    637   
The submitted establishment future carries its own completion guard. If a spawner discards the future before
@@ -653,668 +865,904 @@
  673    688   
before calling it, and issues one call per readiness. The dispatch handle Hyper returns is a different thing.
  674    689   
Its readiness — whether the connection can accept another request — is an inherent
  675    690   
[`SendRequest::poll_ready`](https://github.com/hyperium/hyper/blob/v1.11.0/src/client/conn/http1.rs#L156),
  676    691   
not a Tower `Service`: Hyper has no Tower dependency, and the HTTP/2 form ignores its `Context` entirely
  677    692   
([`http2.rs:97`](https://github.com/hyperium/hyper/blob/v1.11.0/src/client/conn/http2.rs#L97)). Connection
  678    693   
readiness is thus a per-protocol dispatch question the pool answers against the connection's state, and does
  679    694   
not share machinery with the connector's Tower readiness. The two are related only by name.
  680    695   
  681    696   
#### HTTP/1 attempts and HTTP/2 flights
  682    697   
  683         -
The pool supports HTTP/1.1 and HTTP/2 and lets connector ALPN select the protocol. Request version controls
  684         -
dispatch compatibility, not connector negotiation: an HTTP/2-marked request cannot dispatch on H1, an
  685         -
HTTP/1.1-marked request may dispatch on H2, and request version is not part of the origin key.
         698  +
One HTTP/2 connection carries many concurrent request streams. The pool calls
         699  +
one installed incarnation of that connection a *generation*. Replacements
         700  +
receive new generation identities so delayed routes, GOAWAY handling, close
         701  +
work, and request completion cannot affect a newer connection. An
         702  +
`H2Activation` is a reservation for a prospective stream on one exact
         703  +
generation; it becomes an accepted request lease only after Hyper accepts the
         704  +
request.
         705  +
         706  +
The pool supports HTTP/1.1 and HTTP/2 and lets connector ALPN select the protocol. Each request has one of
         707  +
three requirements. `H1Required` covers accepted request forms that require HTTP/1 wire semantics, including
         708  +
HTTP/1.0, ordinary `CONNECT`, and Upgrade. `H1Compatible` may use either protocol. `H2Required` cannot dispatch
         709  +
on HTTP/1. Request protocol is not part of the origin key.
         710  +
         711  +
The default rustls connector offers only `http/1.1` for `H1Required` and offers `h2, http/1.1` otherwise.
         712  +
The s2n connector has a fixed HTTP ALPN offer. Connectors injected through the unstable test utility own
         713  +
their protocol configuration and do not receive the pool's offer. Those paths may therefore negotiate H2
         714  +
for `H1Required`; the pool rejects that result as a protocol mismatch before Hyper establishment.
         715  +
`H2Required` does not make an H2-only offer. If the server
         716  +
selects HTTP/1, the pool retains the useful H1 connection for compatible demand and returns an
         717  +
unsupported-version error with that connection's metadata to the launching request.
  686    718   
  687    719   
The two protocols establish differently because they reuse differently. An HTTP/1 connection carries one
  688    720   
request at a time, so a cell that needs more concurrency needs more connections: each establishment is an
  689    721   
independent *attempt*, and several may run at once, each producing a connection for the request that launched
  690    722   
it. An HTTP/2 connection multiplexes, so one connection serves the whole cell; a second connection to an
  691    723   
origin the cell already serves is nearly pure waste. HTTP/2 establishment is therefore a *flight*,
  692    724   
coordinated so the cell keeps at most one accepting HTTP/2 generation — a request arriving during a flight
  693    725   
waits for it rather than starting its own, and the resulting generation is published to compatible waiters.
  694    726   
  695    727   
##### Post-ALPN convergence
  696    728   
  697    729   
The initial API imposes no pool-wide HTTP/1-only or HTTP/2-only policy. Connector ALPN resolves a transport's
  698         -
protocol only after connect, so concurrent misses may each own a transport until then. Attempts remain
  699         -
independent when they negotiate H1 and converge on the cell's one flight only when they negotiate H2.
         730  +
protocol only after connect, so concurrent misses may each own a transport until then. Compatible attempts
         731  +
remain independent when they negotiate H1 and converge on the cell's one flight only when they negotiate H2.
         732  +
An H1-required attempt that nevertheless negotiates H2 terminates with a protocol-mismatch error.
  700    733   
  701    734   
The logical owner carried through this decision is an *establishment authority*: the transport and, on a
  702    735   
bounded origin, its capacity lease. The authority has exactly one owner even though the request waiting for
  703    736   
its result does not own it; cancellation of that request cannot silently drop the transport or permit. After
  704    737   
ALPN, the owner performs one cell-local select-or-join transition before starting the Hyper protocol handshake:
  705    738   
  706    739   
```text
  707    740   
automatic attempt owns transport + optional capacity lease + launching waiter
  708    741   
  |
  709    742   
  +-- ALPN = H1
  710    743   
  |     `-- run H1 handshake
  711    744   
  |           +-- success -> install H1 record; record takes capacity lease
  712    745   
  |           |     +-- launching waiter accepts H1 -> dispatch
  713    746   
  |           |     `-- waiter requires H2 -> retain H1 for compatible demand;
  714    747   
  |           |                              return the existing unsupported-version error
  715    748   
  |           `-- error -> close transport; return lease; fail launching waiter
  716    749   
  |
  717    750   
  `-- ALPN = H2
  718         -
        `-- atomically inspect this cell's accepting generation and H2 flight
  719         -
              +-- accepting generation -> register waiter against generation
  720         -
              |                           close losing transport; return its lease
  721         -
              +-- flight exists --------> register waiter as follower
  722         -
              |                           close losing transport; return its lease
  723         -
              `-- neither exists -------> install flight as driver
  724         -
                                          flight takes transport + lease
  725         -
                                            |
  726         -
                                            +-- H2 handshake succeeds
  727         -
                                            |     -> install record; record takes lease
  728         -
                                            |     -> publish generation; serve participants
  729         -
                                            `-- error
  730         -
                                                  -> close transport; return lease
  731         -
                                                  -> fail participants
         751  +
        +-- H1 required -> close transport; return lease; fail launching waiter
         752  +
        |
         753  +
        `-- H2 compatible
         754  +
              `-- atomically inspect this cell's accepting generation and H2 flight
         755  +
                    +-- accepting generation -> register waiter against generation
         756  +
                    |                           close losing transport; return its lease
         757  +
                    +-- flight exists --------> register waiter as participant
         758  +
                    |                           close losing transport; return its lease
         759  +
                    `-- neither exists -------> register flight; owner task drives it
         760  +
                                                owner task retains transport + lease
         761  +
                                                  |
         762  +
                                                  +-- H2 handshake succeeds
         763  +
                                                  |     -> install record; record takes lease
         764  +
                                                  |     -> publish generation; serve participants
         765  +
                                                  `-- error
         766  +
                                                        -> close transport; return lease
         767  +
                                                        -> fail participants
  732    768   
```
  733    769   
  734    770   
The inspection and either registration or flight installation are one transition under the cell's
  735    771   
coordination. This is the linearization point: many automatic attempts may reach H2 ALPN, but at most one
  736    772   
becomes the flight. A losing authority has not started a Hyper driver, so its guard closes the negotiated
  737    773   
transport and returns its capacity lease; its launching waiter remains represented exactly once, as a
  738         -
generation user or flight follower. The winning flight installs the connection record before publishing the
  739         -
generation, so no waiter can observe a generation without a driver and capacity owner behind it. Activation
         774  +
generation user or flight participant. The owner task completes the handshake, transfers capacity to the
         775  +
connection record, and installs the generation before submitting the driver. A request activated during that
         776  +
submission interval is retained by Hyper's dispatch channel until the driver is polled. Activation
  740    777   
revalidates the generation identity and accepting state; if either changed after registration, the waiter
  741    778   
returns to acquisition rather than dispatching through stale state.
  742    779   
  743    780   
A waiter that joins a flight or generation retains its original waiter sequence. On publication, the
  744    781   
[generation gate](#http2-publication) orders flight participants and already committed compatible local
  745    782   
waiters together; joining an accepting generation follows that same activation path. Post-ALPN convergence
  746    783   
therefore cannot let a later attempt barge ahead of an older compatible waiter.
  747    784   
  748    785   
Request version changes only compatibility. When an automatic attempt launched by an HTTP/2-marked request
  749    786   
negotiates HTTP/1, the H1 connection remains useful and is handed to compatible local demand or the idle set.
  750    787   
The launching request receives the same unsupported-version classification and connection metadata as the
  751    788   
existing client; it neither dispatches on H1 nor loops establishing until ALPN happens to choose H2.
  752    789   
  753         -
A flight owns a generation identity, a set of participant waiter identities, and optionally a pool-completion
  754         -
interest transferred from a started attempt whose launching waiter was served or cancelled. Cancelling one
  755         -
participant removes only that participant. The flight continues while another participant, compatible demand,
  756         -
or pool-completion interest remains. The completion interest ends when the attempt installs, converges with
  757         -
existing state, or fails; it prevents the no-participant rule from cancelling a started acquisition loser. If
  758         -
none of those interests remains, dropping the establishment authority closes the transport and returns the
  759         -
lease. A dropped flight task performs the same guarded cleanup and reconciles every still-live participant back
  760         -
to acquisition, so no waiter remains attached to a missing flight. Completion from an old generation identity
  761         -
is stale and cannot clear or publish over a successor.
         790  +
A flight record owns its identity and participant waiter identities. The
         791  +
connection-partition owner task separately owns the transport and optional
         792  +
capacity lease while it drives the handshake. Cancelling a participant removes
         793  +
only that waiter; it does not cancel the owner task. If every participant
         794  +
cancels, the task may still install an idle generation, which retains bounded
         795  +
capacity until ordinary close or idle expiration. Task drop closes the
         796  +
transport, returns capacity, and fails every participant still retained by the
         797  +
exact flight. Completion from an old flight identity is stale and cannot clear
         798  +
or publish over a successor.
  762    799   
  763    800   
The ownership transfer is therefore fixed at each boundary:
  764    801   
  765    802   
| State               | Transport owner           | Capacity owner                              | Waiting-request owner              |
  766    803   
| ------------------- | ------------------------- | ------------------------------------------- | ---------------------------------- |
  767    804   
| independent attempt | establishment authority   | its optional capacity lease                 | cell waiter entry                  |
  768    805   
| H1 installed        | H1 connection record      | record's capacity lease                     | checked-out H1 guard or cell queue |
  769         -
| H2 flight driver    | flight authority          | flight's capacity lease                     | flight participant entry           |
         806  +
| H2 flight driver    | connection owner task     | owner task until record installation        | flight participant entry           |
  770    807   
| H2 joiner           | existing flight or record | existing owner; own capacity lease returned | participant or activation          |
  771    808   
| H2 published        | H2 connection record      | record's capacity lease                     | H2 request lease after activation  |
  772    809   
| failure or drop     | cleanup guard             | guard until admission return                | terminal result or re-acquisition  |
  773    810   
  774    811   
This is the mechanism behind the miss policy from
  775    812   
[Local connection selection](#local-connection-selection): a partition that misses locally establishes its
  776    813   
own connection. It reaches for a peer's connection only when it cannot establish, which the next section
  777    814   
covers.
  778    815   
  779    816   
#### Obligations
  780    817   
  781    818   
* **Establishment placement** [safety] — the connector, transport setup, TLS, ALPN, and handshake for a
  782    819   
  connection are polled only on its placement runtime: the runtime bound by the anonymous partition's first
  783    820   
  establishment or the runtime named by an explicit partition.
  784    821   
* **Hyper task placement** [safety] — tasks Hyper spawns for a connection run on that connection's partition
  785    822   
  runtime.
  786    823   
* **Single permit owner** [safety] — at every point in establishment a bounded connection's permit has
  787    824   
  exactly one owner, and a failed or dropped establishment returns its permit to admission.
  788    825   
* **Driver termination closes the record** [safety] — normal driver completion and cancellation of the
  789    826   
  guarded driver task both request the record's exactly-once logical close without owning or retaining its
  790    827   
  capacity lease.
  791    828   
* **Connector readiness** [safety] — the connector is driven to ready before each call, and each readiness
  792    829   
  admits one call.
  793    830   
* **Single generation** [safety] — a cell keeps at most one accepting HTTP/2 generation; establishments that
  794    831   
  resolve HTTP/2 collapse to one, and the losing transports close and return their permits.
  795    832   
* **Atomic H2 convergence** [safety] — after automatic ALPN selects H2, one cell-local transition either
  796    833   
  activates an accepting generation, joins the one current flight, or installs the caller as that flight's
  797    834   
  driver before any H2 handshake or publication begins.
  798    835   
* **Losing-attempt cleanup** [safety] — an automatic H2 attempt that joins existing state closes its
  799    836   
  unhandshaken transport and returns its capacity lease while retaining its launching waiter exactly once.
  800    837   
* **Compatibility-preserving H1 result** [safety] — an H1 result is retained for compatible demand; an
  801         -
  H2-required launching waiter receives the existing unsupported-version result and never dispatches on H1 or
  802         -
  loops establishment for a different ALPN outcome.
         838  +
  H2-required launching waiter receives the existing unsupported-version result with connection metadata and
         839  +
  never dispatches on H1 or loops establishment for a different ALPN outcome.
         840  +
* **Required-H1 negotiation** [safety] — an H1-required request never dispatches on H2. A connector that
         841  +
  accepts per-attempt ALPN offers receives an HTTP/1-only offer for that request.
  803    842   
* **Flight cancellation** [safety] — participant cancellation removes only that participant; terminal flight
  804    843   
  drop closes its transport, returns its lease, and leaves no live waiter attached to the retired flight
  805    844   
  identity.
  806    845   
  807    846   
### Bounded-capacity coordination
  808    847   
  809    848   
A bounded origin at its limit cannot establish: admission has no permit to issue. The permit it needs may
  810    849   
still exist — held by a connection this waiter cannot use, or held in another partition's cell — but a
  811    850   
present permit is not a usable one. So two things have to happen. A waiter parks until a permit it can use
  812    851   
becomes available, and capacity that exists elsewhere moves to where the demand is. This section covers how a
  813    852   
cell signals demand, how the pool decides which capacity a waiter may take, the two operations that move
  814    853   
capacity, and how a freed resource is delivered to a parked waiter. All of it is bounded-mode only; an
  815    854   
unbounded origin never reaches here.
  816    855   
  817    856   
#### Demand
  818    857   
  819    858   
A cell's demand is one aggregate state, not a per-request count: a cell either wants another connection or it
  820    859   
does not. The state is active while any request waits and inactive when none do, so it is a single standing
  821    860   
ticket per cell rather than one ticket per waiting request.
  822    861   
  823    862   
The waiting requests queue behind the ticket in arrival order. An arriving permit goes to the request at the
  824    863   
head, which stops waiting; the next becomes the head. The ticket is how returning connections and other
  825    864   
cells find a cell with unmet demand; the queue is how that cell chooses whom to serve first.
  826    865   
  827         -
The ticket carries one thing beyond its presence: whether the head request can use an HTTP/1 connection. A
  828         -
peer with a reusable HTTP/1 connection to lend needs to know its offer will be taken before it acts, and a
  829         -
head that requires HTTP/2 cannot use an HTTP/1 handle. This is the minimum the signal must distinguish;
  830         -
finer matching is a dispatch-time question, not a demand-signal one.
         866  +
The ticket carries one thing beyond its presence: the head request's protocol requirement. Admission must know
         867  +
whether an HTTP/1 return, an HTTP/2 publication, or either can satisfy the head before it reserves a resource.
         868  +
This is the minimum the signal must distinguish; finer matching remains a dispatch-time question.
  831    869   
  832    870   
The logical snapshot published to admission is:
  833    871   
  834    872   
```rust
  835    873   
enum ProtocolRequirement {
         874  +
    H1Required,
  836    875   
    H1Compatible,
  837    876   
    H2Required,
  838    877   
}
  839    878   
  840    879   
enum DemandState {
  841    880   
    Active {
  842    881   
        head: ProtocolRequirement,
  843    882   
        eligibility_group: EligibilityGroup,
  844    883   
    },
  845    884   
    Inactive,
@@ -910,949 +969,1012 @@
  930    969   
    NetworkInterface,
  931    970   
    Pool,
  932    971   
}
  933    972   
```
  934    973   
  935    974   
`Partition` permits no cross-cell dispatch. `NetworkInterface` groups partitions by the exact configured
  936    975   
interface, with all unbound partitions in one group. `Pool` permits reuse across every partition. Scope
  937    976   
controls only dispatch eligibility: it does not constrain reclaim, which closes a connection and transfers
  938    977   
capacity rather than moving I/O authority.
  939    978   
         979  +
An **eligibility group** is the exact set of partitions whose reuse policies
         980  +
allow them to share a connection. Each partition belongs to one such group for
         981  +
an origin, derived from the configured scope above.
         982  +
  940    983   
Dispatching through a connection that already exists consumes no capacity; that connection already holds one
  941    984   
of the admitted permits. So an eligible reusable connection serves the waiter whether or not a permit is
  942    985   
free — a pool at its limit still dispatches freely through all `N` of its connections. Capacity is consulted
  943    986   
only when no eligible connection can serve the waiter and the cell must open one. The two questions do not
  944    987   
read each other's state, and only that second case reaches admission: eligibility decides reuse, capacity
  945    988   
decides establishment. Keeping them separate is what lets the two operations below relieve a shortage of one
  946    989   
without disturbing the other.
  947    990   
  948    991   
#### Borrow and reclaim
  949    992   
@@ -1290,1333 +1404,1492 @@
 1310   1353   
H2 publication carries no `AcquisitionPayload`. The connection-owning record continues to own the capacity
 1311   1354   
lease, while a `(connection-owning cell, generation identity)` notice says that compatible requests may take
 1312   1355   
H2 request leases. Publishing a new local generation first installs the record and accepting generation under
 1313   1356   
the connection-owning cell lock, then makes that identity visible to compatible local waiters. They are woken
 1314   1357   
and admitted in bounded local turns; publication does not scan or synchronously wake an unbounded queue.
 1315   1358   
 1316   1359   
Publication also installs a local fairness gate:
 1317   1360   
 1318   1361   
```rust
 1319   1362   
enum GenerationGate {
 1320         -
    Prioritizing { cutoff: WaiterSequence },
 1321         -
    Open,
        1363  +
    Closed,
        1364  +
    Prioritizing {
        1365  +
        generation: H2GenerationId,
        1366  +
        cutoff: WaiterId,
        1367  +
        activating: Option<WaiterId>,
        1368  +
    },
        1369  +
    Open { generation: H2GenerationId },
 1322   1370   
}
 1323   1371   
```
 1324   1372   
 1325         -
The cutoff is the newest compatible waiter already committed when publication becomes visible. While the gate
 1326         -
is `Prioritizing`, those waiters are offered generation activation oldest first in bounded turns, and a newer
 1327         -
arrival cannot activate through the generation ahead of them. Cancellation removes its waiter from the owed
 1328         -
set. When no waiter at or before the cutoff remains, the gate opens and later arrivals use the ordinary local
 1329         -
path. Generation invalidation removes the gate and sends any unserved committed waiter back through
 1330         -
acquisition. The gate orders activation opportunities; each successful activation still creates its own H2
 1331         -
request lease.
        1373  +
`Closed` names no usable generation. The cutoff is the newest compatible
        1374  +
waiter committed when a generation becomes visible. While the gate is
        1375  +
`Prioritizing`, those waiters receive activation oldest first, and
        1376  +
`activating` retains the one priority turn crossing to Hyper acceptance or
        1377  +
cancellation. When no waiter at or before the cutoff remains, the gate becomes
        1378  +
`Open`. Open generations issue concurrent prospective activations; Hyper owns
        1379  +
stream credit and flow control. Generation invalidation closes the gate and
        1380  +
returns unserved transferred waiters to acquisition. Each successful
        1381  +
activation creates its own H2 request lease.
        1382  +
        1383  +
A requesting cell retains peer-route crossing state separately:
        1384  +
        1385  +
```rust
        1386  +
struct PeerH2Route {
        1387  +
    route: H2Route,
        1388  +
    gate: GenerationGate,
        1389  +
    crossing: Option<WaiterId>,
        1390  +
}
        1391  +
```
        1392  +
        1393  +
Queued peer-route activations cross the connection-cell lock one at a time so a failed exact-generation
        1394  +
reservation can restore the requesting cell's oldest turn. Once the route gate is open, direct arrivals
        1395  +
reserve independent prospective leases concurrently; `crossing` does not serialize them.
 1332   1396   
 1333   1397   
The peer-cell index holds one group-scoped advertisement for each accepting H2 generation. Record and
 1334   1398   
generation installation precede advertisement; transition out of accepting removes it. Either a new
 1335   1399   
advertisement or new group demand schedules a bounded publication turn from stored connection and demand heads.
 1336   1400   
The advertisement carries only connection-owning cell and generation identity, not a dispatch handle or
 1337   1401   
capacity lease. The publication guard revalidates it at the connection-owning and requesting cells. A stale
 1338   1402   
advertisement is removed or updated before the next turn, so peer H2 discovery does not scan cells.
 1339   1403   
 1340   1404   
Under bounded pressure, an accepting generation may also be announced to the
 1341   1405   
head of its eligibility-group all-protocol view. Admission fences that demand
 1342   1406   
generation in `Delivering` with a `DeliveryId`; the publication action carries
 1343   1407   
only connection-owning cell and generation identities plus an acknowledgement
 1344   1408   
fallback, never the record's capacity lease.
 1345   1409   
The requesting cell revalidates the generation identity, accepting state, demand
 1346   1410   
generation, and reuse scope.
 1347   1411   
Acceptance makes the generation visible to compatible waiters there; rejection discards the stale notice
 1348   1412   
while the generation and
 1349   1413   
permit remain at the connection-owning cell. Acceptance acknowledges after requesting-cell visibility and the
 1350   1414   
named head's activation opportunity are committed, not after every local waiter has activated. Remaining local
 1351   1415   
waiters proceed through the generation gate in bounded turns and no longer advertise a connection need while
 1352   1416   
that generation remains usable. Later group tickets are handled by subsequent bounded publication turns, so
 1353   1417   
one-to-many visibility does not turn one host action into work proportional to partition count.
        1418  +
        1419  +
An H1-required head cannot use an advertised HTTP/2 generation. If bounded origin capacity is exhausted,
        1420  +
admission may instead reserve the oldest advertised idle generation for reclaim. The connection cell
        1421  +
revalidates the exact generation, accepting residence, and zero prospective or accepted request counts before
        1422  +
closing it. Logical close returns its capacity to origin admission, which can then grant establishment to the
        1423  +
H1-required head. A generation with request work is not reclaimed. Its transition to idle publishes a new
        1424  +
advertisement and makes reclaim eligible.
        1425  +
        1426  +
Admission attempts this reclaim only when the transport factory guarantees that
        1427  +
an H1-required establishment uses HTTP/1. Without that guarantee, admission
        1428  +
could close a healthy H2 connection, negotiate H2 again, reject the result, and
        1429  +
repeat without satisfying the waiting request. The default rustls and cleartext
        1430  +
connectors provide the guarantee. The fixed-ALPN s2n connector and connectors
        1431  +
injected through the unstable test utility do not, so H1-required demand
        1432  +
remains queued behind their healthy H2 capacity until an ordinary close
        1433  +
releases a permit.
        1434  +
        1435  +
For bounded origins, a generation publishes only its zero-to-one and one-to-zero request-work transitions.
        1436  +
Intermediate multiplexed request counts remain cell-local. Unbounded origins have no admission publication.
        1437  +
 1354   1438   
Dropping a pending publication guard submits its `on_drop` acknowledgement so the fence retries or closes;
 1355   1439   
there is no single-owner payload to refunnel. Committing publication stores the requesting-cell
 1356   1440   
acknowledgement, which is submitted before the guard disarms.
 1357   1441   
 1358   1442   
This separates publication from single delivery: a permit or H1 has one owner and one requesting cell, while
 1359   1443   
an H2 generation remains owned by its connection cell and may be announced repeatedly. The transitions must
 1360   1444   
be model checked; their invariants are stated in [Correctness invariants](#correctness-invariants), with the
 1361   1445   
checks specified in [Appendix B](#appendix-b-validation).
 1362   1446   
 1363   1447   
#### Obligations
 1364   1448   
 1365   1449   
* **Bounded demand** [safety] — a cell carries at most one active demand generation regardless of how many
 1366   1450   
  requests wait, so demand accumulates no deficit and one residence receives at most one terminal outcome.
 1367   1451   
* **Snapshot ordering** [safety] — admission retains the newest complete demand version and rejects an action
 1368   1452   
  for a retired generation, so out-of-order publication cannot resurrect cancelled or satisfied demand.
 1369   1453   
* **Eligibility and capacity independence** [safety] — the capacity decision and the eligibility decision do
 1370   1454   
  not read each other's state.
 1371   1455   
* **Placement under transfer** [safety] — neither borrow nor reclaim moves a connection's driver or I/O off
 1372   1456   
  its owning partition.
 1373   1457   
* **Reclaim scope independence** [safety] — reclaim moves a permit without dispatching across a partition
 1374   1458   
  boundary, so it is not constrained by the reuse scope.
        1459  +
* **Protocol-compatible reclaim** [safety] — admission reclaims idle H2 capacity for H1-required demand only
        1460  +
  when the transport factory guarantees an HTTP/1 establishment result.
        1461  +
* **H2 work publication** [safety, optimization] — bounded cells publish when a generation crosses between
        1462  +
  zero and nonzero request work; intermediate stream counts require no admission update.
 1375   1463   
* **Single delivery** [safety] — one delivery identity owns at most one permit or provisional H1, commits it
 1376   1464   
  to at most one requesting waiter, and retains its scheduling fence until requesting-cell acknowledgement.
 1377   1465   
* **Refunnelling** [safety] — rejection, supersession, cancellation, task drop, or panic returns every
 1378   1466   
  undelivered permit to admission and every undelivered H1 to its connection-owning cell exactly once.
 1379   1467   
* **Publication ownership** [safety] — H2 publication carries generation identity, never the connection's
 1380   1468   
  capacity lease; request activation takes a request lease while the connection record remains the capacity owner.
 1381   1469   
* **Publication priority** [liveness] — publication closes the generation gate before visibility and offers
 1382   1470   
  activation to waiters committed at publication before newer arrivals, in bounded oldest-first turns.
 1383   1471   
* **Cross-cell order** [safety] — H1 borrow and reclaim both serve the current origin head, preferring an
 1384   1472   
  eligible peer connection for borrow and otherwise reclaiming an origin peer. Peer H2 publication uses its
@@ -1501,1589 +1561,1660 @@
 1521   1609   
endpoint and can rearm the same adapter for a later selection without retaining the rejected generation.
 1522   1610   
 1523   1611   
#### Retry, timeout, and errors
 1524   1612   
 1525   1613   
There is one authority for transparent dispatch retry: Hyper must return the original, unsent request from
 1526   1614   
`try_send_request`, and the selected connection must have been reused. The pool restores the absolute URI,
 1527   1615   
disarms any unaccepted H2 body endpoint, retires or invalidates the stale selected connection, and sends that
 1528   1616   
same request through acquisition again. An unsent failure from a fresh connection is terminal and still
 1529   1617   
retires a sender Hyper reported unable to accept the request; its inert body endpoint and selected guard cannot
 1530   1618   
survive the error. The pool does not clone a request or replay one Hyper accepted.
 1531         -
The existing caller timeout and cancellation remain able to terminate repeated stale selections.
        1619  +
        1620  +
`AcquisitionResult::Reacquire` is the internal transition for a waiter whose selected generation became stale
        1621  +
before Hyper observed the request. It carries no request copy and is not an SDK retry attempt: the original
        1622  +
request remains owned by the pool future and re-enters acquisition. A request returned unsent by Hyper uses the
        1623  +
same loop only under the reused-connection rule above. Negotiated-protocol mismatch and a fresh Hyper error
        1624  +
without the original envelope are terminal.
        1625  +
        1626  +
After dispatch receives an H2 activation, the request may select at most two replacement H2 generations.
        1627  +
A third pre-acceptance rejection returns its retained connection error. This bound is independent of caller
        1628  +
timeout and cancellation and prevents repeated GOAWAY or close races from creating an unbounded establishment
        1629  +
loop.
        1630  +
 1532   1631   
With Hyper 1.11, the returned-request path after `try_send_request` is reachable for H1; H2 dispatch errors
 1533   1632   
after call do not carry the original request and are terminal. An H2 readiness failure observed before call is
 1534   1633   
different: the request is still locally owned and may return to acquisition without being replayed.
 1535   1634   
 1536   1635   
Every error after Hyper accepts the envelope is terminal for this dispatch. H1 conservatively retires because
 1537   1636   
the request may have reached the wire. H2 releases or resets the affected stream and drives its send and
 1538   1637   
receive endpoints to terminal state; a stream-local reset does not by itself invalidate the accepting
 1539   1638   
generation. GOAWAY, a closed dispatcher, a connection error, or other connection-fatal evidence does
 1540   1639   
invalidate that generation. The resulting `ConnectorError` preserves the current source chain and
 1541   1640   
classification, including timeout, user, I/O, incomplete-message transient,
@@ -1995,2094 +2055,2154 @@
 2015   2114   
`establishing` starts when an attempt or flight is admitted and ends when it fails or installs a record.
 2016   2115   
`h1_idle` and `h1_active` partition logically open H1 records; checked-out and `Reserved` H1 records are
 2017   2116   
active, while only dispatch-eligible records in the idle set are idle.
 2018   2117   
`h2_accepting` counts generations that may issue request leases, while `h2_active_streams` counts accepted
 2019   2118   
request leases across accepting and draining generations. Logical close moves a connection out of those
 2020   2119   
admitted gauges and, while its transport remains, into `h1_draining` or `h2_draining`. An upgraded H1 remains
 2021   2120   
H1-draining until its wrapped root I/O drops. `physically_live` starts when the connector's returned transport
 2022   2121   
is wrapped for
 2023   2122   
lifecycle tracking and ends only at root-I/O drop, so it includes handshaking, admitted, draining, and
 2024   2123   
upgraded transports and may exceed the configured limit. `waiting_requests` counts requests registered in
 2025         -
acquisition that do not yet own a dispatch authority, including flight followers.
        2124  +
acquisition that do not yet own a dispatch authority, including flight participants.
 2026   2125   
 2027   2126   
`OriginStats::admitted` derives the saturating sum of `establishing`, `h1_idle`, `h1_active`, and
 2028   2127   
`h2_accepting` from the captured partition rows; it is not another shared counter or admission authority.
 2029   2128   
Draining connections are excluded because they hold no permits. `limit` is the configured per-origin bound,
 2030   2129   
or `None` for an unbounded origin. A connection and its streams are attributed to the owner partition even
 2031   2130   
when another partition dispatches through them; a waiting request is attributed to its requesting partition.
 2032   2131   
 2033   2132   
Each gauge changes at its named transition before any corresponding callback. The counters use relaxed or
 2034   2133   
otherwise non-coordinating reads so observation adds no origin-wide write to a local reuse hit. Consequently,
 2035   2134   
an `OriginStats` value is not an atomic multi-counter snapshot: a transition can appear in one loaded field
@@ -2084,2183 +2162,2274 @@
 2104   2203   
**Reclaim** — closing a connection so its permit can move to another cell. Transfers capacity, not I/O.
 2105   2204   
 2106   2205   
**Capacity lease**, **request lease**, and **handle** — a capacity lease is the exclusive hold on one permit
 2107   2206   
and moves from establishment to the connection record. An H2 request lease owns one prospective or accepted
 2108   2207   
stream's two-ended lifecycle but no permit. A dispatch handle can address a connection and owns neither kind of
 2109   2208   
capacity.
 2110   2209   
 2111   2210   
**Retry authority** — proof that the same request may be dispatched again. Only Hyper returning the original
 2112   2211   
request unsent from a reused connection creates this authority; request clonability does not.
 2113   2212   
        2213  +
**Reacquire** — returning a still pool-owned request to acquisition after its selected protocol state becomes
        2214  +
stale before dispatch. It is an internal state transition, not an SDK retry or authority to clone an accepted
        2215  +
request.
        2216  +
 2114   2217   
**Publish** and **deliver** — publication makes state visible to many unnamed readers; delivery hands one
 2115   2218   
value to one waiting party. `DemandResidence` tracks a ticket's residence and acknowledgement fence;
 2116   2219   
`DeliveryGuardState` owns a one-to-one payload while it crosses locks.
 2117   2220   
 2118   2221   
**Attempt** and **flight** — an HTTP/1 establishment is an attempt, independent of other attempts; an
 2119   2222   
HTTP/2 establishment is a flight. Automatic attempts remain independent before ALPN, then atomically join or
 2120         -
install at most one post-ALPN H2 flight per cell.
        2223  +
install at most one post-ALPN H2 flight per cell. The flight record owns participant identities; the
        2224  +
connection-partition owner task owns handshake completion.
        2225  +
        2226  +
**Advertisement** and **route** — an advertisement is admission's report that one connection cell has an
        2227  +
accepting H2 generation. Publication installs a route in a requesting cell. The route names that exact
        2228  +
connection cell and generation but owns no sender, socket, driver, or capacity.
 2121   2229   
 2122   2230   
**Logical close** — a connection stops accepting new work and releases its permit. **Physical close** —
 2123   2231   
the socket is gone. Physical close follows logical close by an unbounded interval, so live sockets can
 2124   2232   
outnumber admitted connections; `max_connections_per_host` bounds admitted connections, not file descriptors,
 2125   2233   
and no finite bound on live sockets follows from it.
 2126   2234   
 2127   2235   
**Draining generation** and **draining connection** — a draining H2 generation accepts no new request leases
 2128   2236   
while already accepted streams finish. A draining connection has logically closed and released its permit but
 2129   2237   
still has physically live root I/O, and is counted by `h1_draining` or `h2_draining`. An H2 record may be both
 2130   2238   
while its accepted streams and transport finish.
 2131   2239   
 2132         -
**Generation** — an HTTP/2 connection's dispatch epoch, a first-class object with a lifecycle.
        2240  +
**Generation** — one uniquely identified incarnation of an installed HTTP/2
        2241  +
connection. It owns the authoritative request handle, accepting or draining
        2242  +
state, request counts, and connection capacity. A replacement connection has a
        2243  +
new generation identity so delayed work for the previous incarnation is
        2244  +
rejected.
 2133   2245   
 2134   2246   
**Demand generation** — one cell queue head's `DemandId`. A **snapshot version** orders complete publications
 2135   2247   
for that generation. Readers retain the newest publication and reject work for a retired generation.
 2136   2248   
 2137   2249   
**Obligation** — a duty a component owes, stated as one sentence an implementation either satisfies or does
 2138   2250   
not. `[safety]` obligations forbid a state; `[liveness]` obligations require an outcome; `[optimization]`
 2139   2251   
obligations bound a cost, and violating one is a regression rather than a defect.
 2140   2252   
 2141   2253   
## Correctness invariants
 2142   2254   
@@ -2358,2470 +2506,2639 @@
 2378   2490   
an explicit partition using the reserved anonymous identity. `BuildError` reports the setting and value that
 2379   2491   
failed and implements `Error`; callers are not expected to branch on an exhaustive variant set.
 2380   2492   
 2381   2493   
When `partitions` is never set, construction creates the one anonymous, unbound partition. Once set, the
 2382   2494   
supplied nonempty set is the complete explicit topology and no anonymous partition is added.
 2383   2495   
`max_connections_per_host` is unset by default, and an unset bound constructs no admission machinery. When
 2384   2496   
set, it bounds one scheme-host-port origin across all partitions and interface groups, not per partition. HTTP
 2385   2497   
and HTTPS and distinct non-default ports are bounded separately. The limit counts every establishing, idle, and
 2386   2498   
active connection rather than only idle connections.
 2387   2499   
 2388         -
The initial builder exposes no pool-wide HTTP/1-only or HTTP/2-only policy. The connector determines the
 2389         -
negotiated protocol, while request version controls dispatch compatibility after negotiation and does not
 2390         -
configure connector ALPN. A future protocol-policy API requires a concrete Smithy caller and a definition of
 2391         -
how that policy composes when multiple clients share one pool.
        2500  +
The initial builder exposes no pool-wide HTTP/1-only or HTTP/2-only policy. Accepted request forms that require
        2501  +
HTTP/1 wire semantics narrow the per-attempt offer to `http/1.1` when the connector supports it; other requests
        2502  +
use the default `h2, http/1.1` offer. An HTTP/2-marked request does not force an H2-only offer. A future
        2503  +
protocol-policy API requires a concrete Smithy caller and a definition of how that policy composes when
        2504  +
multiple clients share one pool.
        2505  +
        2506  +
The default rustls connector applies the per-attempt offer. The s2n connector's
        2507  +
fixed HTTP ALPN list cannot be narrowed by the pool. A connector injected
        2508  +
through the unstable test utility also owns its own protocol configuration and
        2509  +
does not receive the pool's offer. The pool validates the negotiated result but
        2510  +
does not replace either connector's TLS implementation. Making the s2n offer
        2511  +
configurable is an upstream connector follow-up.
        2512  +
        2513  +
Because those paths cannot force HTTP/1, bounded admission does not reclaim a
        2514  +
healthy idle H2 generation solely to serve H1-required demand. Closing that
        2515  +
generation could negotiate H2 again and discard useful capacity without
        2516  +
satisfying the request. The request remains queued until ordinary close
        2517  +
releases capacity.
 2392   2518   
 2393   2519   
An unset `idle_timeout` uses a 90-second default. Passing `None` to the fluent setter disables idle
 2394   2520   
timeout. The mutable setter preserves all three configuration states: outer `None` restores the default,
 2395   2521   
`Some(None)` disables the timeout, and `Some(Some(duration))` selects a duration. The builder's time source and
 2396   2522   
sleep implementation drive pool maintenance; their defaults are the production system clock and Tokio sleep.
 2397   2523   
They are pool inputs and are not replaced by per-operation `RuntimeComponents`.
 2398   2524   
 2399   2525   
Building retains configuration and assembles the reusable transport factory but opens no socket. Native trust
 2400   2526   
loading is deferred to the idempotent connector preflight invoked when the `Client` is selected by smithy, or
 2401   2527   
to first establishment when used without smithy validation. Whether a configured interface exists and can be
 2402   2528   
used is therefore reported later as a connector error on establishment, not `BuildError`. Likewise,
 2403   2529   
`TokioDriverSpawner::current` retains its own documented panic outside a Tokio runtime because that constructor
 2404   2530   
is invoked before the spawner is passed to the pool.
 2405   2531   
 2406   2532   
### Module structure
 2407   2533   
 2408   2534   
```text
 2409   2535   
aws-smithy-http-client/src/client/
 2410   2536   
  pool.rs              — ConnectionPool ownership and public re-exports
 2411   2537   
  pool/
 2412   2538   
    builder.rs         — Builder typestate, validation, connector assembly
 2413   2539   
    client.rs          — Client, PoolConnector, and ClientBuildError
 2414   2540   
    partition.rs       — partition declarations and runtime/interface placement
 2415   2541   
    origin.rs          — owned OriginKey, borrowed lookup, and canonicalization
 2416   2542   
    registry.rs        — PartitionRegistry, PartitionState, and stable cell publication
 2417   2543   
    cell.rs            — OriginCell and cell-level acquisition coordination
 2418   2544   
    cell/
 2419   2545   
      h1.rs            — HTTP/1 records, sender ownership, and reuse reservation
        2546  +
      h2.rs            — HTTP/2 flights, generations, routes, gates, and request leases
 2420   2547   
      waiters.rs       — local acquisition queue and delivery reservation
 2421   2548   
    admission.rs       — bounded-origin capacity and unlocked action driving
 2422   2549   
    admission/
 2423   2550   
      demand.rs        — versioned demand order and delivery fences
        2551  +
      order.rs         — checked intrusive order shared by admission indexes
 2424   2552   
      reuse.rs         — H1 availability order and cross-cell reuse operations
 2425   2553   
      delivery.rs      — capacity/H1 crossing guards and acknowledgements
        2554  +
      publication.rs   — HTTP/2 advertisements, peer publication, and fences
 2426   2555   
    establish.rs       — transport construction below protocol establishment
 2427   2556   
    establish/
 2428         -
      h1.rs            — HTTP/1 connect, handshake, installation, and driver
        2557  +
      h1.rs            — HTTP/1 handshake, installation, and driver
        2558  +
      h2.rs            — post-ALPN flight convergence, handshake, and driver
 2429   2559   
    dispatch.rs        — protocol-neutral request routing
 2430   2560   
    dispatch/
 2431         -
      h1.rs            — HTTP/1 acquisition, dispatch, retry, and response ownership
        2561  +
      h1.rs            — HTTP/1 dispatch, reacquisition, and response ownership
        2562  +
      h2.rs            — HTTP/2 dispatch and two-ended request completion
 2432   2563   
    maintenance.rs     — idle-deadline scheduling and partition task lifetime
 2433   2564   
    connection.rs      — records, leases, logical close, physical completion
 2434   2565   
    events.rs          — listener and lifecycle event types
 2435   2566   
    stats.rs           — origin/partition snapshots and lifecycle gauges
 2436   2567   
aws-smithy-http-client/src/
 2437   2568   
  sync/
 2438   2569   
    mod.rs              — standard-library and Loom backend selection
 2439   2570   
    std.rs              — production synchronization facade
 2440   2571   
    loom.rs             — modeled synchronization facade
 2441   2572   
```
 2442   2573   
 2443   2574   
The inventory describes ownership boundaries, not implementation order. The `pool` module re-exports every
 2444   2575   
public type above; private modules may be split or combined without changing the architecture so long as the
 2445   2576   
lock, ownership, and hot-path boundaries remain intact.
 2446   2577   
 2447   2578   
The transport-connector contract below the pool is unchanged: it is a `Service<Uri>` yielding
 2448   2579   
`(IO, Connected)`. The pool composes transport connectors and does not replace that contract; the smithy
 2449   2580   
`HttpConnector` above the pool is the request-policy facade described earlier.
 2450   2581   
 2451   2582   
---
 2452   2583   
 2453   2584   
## Appendix B: Validation
 2454   2585   
 2455   2586   
Validation supplies implementation evidence for the contracts above; it does not redefine them. The
 2456   2587   
implementation-neutral connection harness and existing-client behavior suites in smithy-rs
 2457   2588   
[PR #4767](https://github.com/smithy-lang/smithy-rs/pull/4767) form the compatibility baseline. Pool-specific
 2458   2589   
tests preserved on `archive/conn-pool-4708` are an additional inventory, not an acceptance target: applicable
 2459   2590   
contracts may be retained or rewritten, implementation-specific assumptions may be obsolete, and the owned
 2460   2591   
state machine requires coverage that prototype tests did not contain.
 2461   2592   
 2462   2593   
The evidence levels have distinct jobs:
 2463   2594   
 2464   2595   
* **Unit, property, and bounded transition tests** cover construction, canonicalization, indexing,
 2465   2596   
  accounting, and explicit state-machine transitions. Where focused state-space enumeration is used, the test
 2466   2597   
  identifies its operation alphabet and bound; ordinary transition tests are not described as exhaustive.
 2467   2598   
* Focused **Loom kernels** compile the production synchronization-bearing code against Loom and exercise
 2468   2599   
  concurrent cell publication, permit and H1 delivery, H1 selection and return, borrowed-H1 materialization,
 2469         -
  reuse cancellation, logical close, and maintenance publication or shutdown. They model these
 2470         -
  ownership boundaries rather than sockets or the complete network client. HTTP/2 generation publication and
 2471         -
  request-lease kernels are added with those mechanisms.
        2600  +
  reuse cancellation, logical close, maintenance publication or shutdown, HTTP/2 activation against close,
        2601  +
  two-ended request completion against generation close, route installation against requesting-cell
        2602  +
  cancellation, and peer publication against connection-cell close. They model these ownership boundaries
        2603  +
  rather than sockets or the complete network client.
 2472   2604   
* **Controlled-runtime tests** use injected time, sleep, connectors, and executors to force cancellation at
 2473   2605   
  ownership-distinct cancellation boundaries, submitted-future drop, idle deadlines, independent-runtime
 2474   2606   
  request movement, explicit placement checks, and connector or handshake failure.
 2475   2607   
* The **wire harness** verifies HTTP/1.1 and HTTP/2 behavior against scripted peers, including reuse,
 2476         -
  multiplexing, ALPN, GOAWAY, stream reset, incomplete bodies, upgrades, poisoning, and transport close.
        2608  +
  multiplexing, request-specific ALPN, GOAWAY before and after first-stream acceptance, stream reset,
        2609  +
  incomplete bodies, upgrades, poisoning, and transport close.
 2477   2610   
* **Differential tests** run the same implementation-neutral behavior contracts against the current
 2478   2611   
  Hyper-util-backed client and this pool. Any difference in request behavior, metadata, timeout scope, or error
 2479   2612   
  classification requires an explicit design decision rather than a rewritten oracle.
 2480   2613   
* **Benchmarks and stress tests** establish that the optimization and liveness contracts remain true at
 2481   2614   
  production concurrency and topology.
 2482   2615   
 2483   2616   
The required evidence maps to the architecture as follows:
 2484   2617   
 2485   2618   
| Mechanism                                                             | Primary evidence                                                                  | What it must establish                                                                                                                                                                                                                                         |
 2486   2619   
| --------------------------------------------------------------------- | --------------------------------------------------------------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- |

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

@@ -1,1 +104,176 @@
    7      7   
//!
    8      8   
//! A [`ConnectionPool`] owns connection policy and a fixed set of partitions.
    9      9   
//! A [`Client`] binds Smithy HTTP operations to one partition. Pools without
   10     10   
//! explicit [`Partition`] values contain one anonymous partition; otherwise a
   11     11   
//! client selects a declared partition by [`PartitionId`].
   12     12   
//!
   13     13   
//! A partition owns connection establishment, protocol drivers, and idle
   14     14   
//! maintenance. [`ConnectionReuseScope`] controls whether another partition may
   15     15   
//! dispatch through those connections. Reuse transfers protocol dispatch
   16     16   
//! authority; it never moves the socket, driver, or capacity accounting.
          17  +
//! Partitions in the same eligibility group are exactly those whose configured
          18  +
//! reuse scope permits them to share a connection.
          19  +
//!
          20  +
//! Connection establishment and installed connection lifetime are separate
          21  +
//! ownership phases:
          22  +
//!
          23  +
//! ```text
          24  +
//! establishment task
          25  +
//!     |-- DNS, socket, proxy, TLS, and ALPN
          26  +
//!     `-- negotiated transport
          27  +
//!             `-- PendingOpen
          28  +
//!                     |-- Hyper protocol setup
          29  +
//!                     `-- Open
          30  +
//!                             `-- pool installation
          31  +
//!                                     `-- discoverable -> draining -> closed
          32  +
//! ```
          33  +
//!
          34  +
//! The establishment task and its permit represent work that may fail before a
          35  +
//! physical connection exists. `ConnectionState` begins only after the
          36  +
//! connector returns connected I/O and selects HTTP/1 or HTTP/2. For TLS, that
          37  +
//! boundary follows TLS and ALPN. Hyper protocol setup moves the state from
          38  +
//! `PendingOpen` to `Open`; cell installation then makes the connection
          39  +
//! discoverable. Establishment and connection events can therefore be observed
          40  +
//! independently without representing failed attempts as installed
          41  +
//! connections.
   17     42   
//!
   18     43   
//! # State ownership
   19     44   
//!
   20     45   
//! ```text
   21     46   
//! ConnectionPool
   22     47   
//! `-- PoolInner
   23     48   
//!     |-- immutable policy and transport factory
   24     49   
//!     `-- PartitionRegistry
   25     50   
//!         |-- PartitionState per partition
   26     51   
//!         |   |-- runtime placement and idle maintenance
   27     52   
//!         |   `-- OriginCell per canonical origin
   28     53   
//!         |       |-- acquisition queue
   29     54   
//!         |       `-- protocol connection records
   30     55   
//!         `-- OriginAdmission per bounded origin
   31     56   
//!             |-- connection permits
   32     57   
//!             |-- cross-cell demand order
   33     58   
//!             `-- peer reuse indexes
   34     59   
//! ```
   35     60   
//!
   36     61   
//! One `OriginCell` lock owns local acquisition order and protocol residence.
   37     62   
//! For a bounded origin, `OriginAdmission` separately owns the origin-wide
   38         -
//! connection limit and cross-cell scheduling. Delivery values carry their own
   39         -
//! payload and cancellation fallback between those lock domains; cell and
   40         -
//! admission locks are never held together.
          63  +
//! connection limit and cross-cell scheduling. An H2 request-lease lock owns
          64  +
//! its two endpoint bits. `ConnectionState` owns logical connection lifetime,
          65  +
//! and partition maintenance owns its scheduler state.
          66  +
//!
          67  +
//! No two pool locks are held together. Delivery and publication guards carry
          68  +
//! payload or identity between cell and admission scopes. H2 lease completion
          69  +
//! detaches its dispatch guard before entering connection or cell state.
          70  +
//! Maintenance detaches cells and wakers before expiration or wake callbacks.
   41     71   
//!
   42     72   
//! # HTTP/1 request lifecycle
   43     73   
//!
   44     74   
//! Hyper represents an HTTP/1 connection with one exclusive
   45     75   
//! `SendRequest<SdkBody>` handle. This module calls that handle the sender. The
   46     76   
//! sender authorizes request dispatch but does not own the socket or protocol
   47     77   
//! driver.
   48     78   
//!
   49     79   
//! ```text
   50     80   
//! Client(partition, request)
   51     81   
//! `-- OriginCell(partition, origin)
   52     82   
//!     |-- local idle sender --------------------------> H1Selection
   53     83   
//!     `-- acquisition queue
   54     84   
//!         |-- returned local or eligible peer sender -> H1Selection
   55     85   
//!         |-- capacity permit -> establish HTTP/1 ---> H1Selection
   56     86   
//!         `-- reclaimed peer capacity -> establish ---> H1Selection
   57     87   
//!
   58     88   
//! H1Selection -- Hyper accepts request --> H1Exchange
   59     89   
//! H1Exchange
   60     90   
//!     |-- complete response + ready --> offer to owning OriginCell
   61     91   
//!     `-- failure, cancellation, or upgrade ----------> retire pool record
   62     92   
//! ```
   63     93   
//!
   64     94   
//! A local hit touches only the cell lock. On a miss, one queued acquisition
   65     95   
//! remains authoritative while a returned sender and establishment race to
   66     96   
//! satisfy it. Bounded origins may borrow an eligible peer sender or reclaim a
   67     97   
//! peer connection and transfer its permit; active connections are not
   68     98   
//! reclaimed.
   69     99   
//!
   70    100   
//! Dispatch commits against logical close before Hyper receives the request.
   71    101   
//! Once accepted, the response lifecycle retains the sender until Hyper proves
   72    102   
//! a complete reusable message boundary. Failure retires the connection, and
   73    103   
//! an upgrade closes the pool record before exposing upgraded root I/O.
   74    104   
//!
         105  +
//! # HTTP/2 request lifecycle
         106  +
//!
         107  +
//! One HTTP/2 connection carries many concurrent request streams. The pool
         108  +
//! calls one installed incarnation of that connection a generation. A
         109  +
//! replacement connection receives a new generation identity so delayed
         110  +
//! close, route, and completion work cannot affect it.
         111  +
//!
         112  +
//! The connection-owning cell retains the generation's authoritative Hyper
         113  +
//! request handle and capacity. A requesting cell may retain only a route that
         114  +
//! names the owning cell and one exact accepting generation. Each use
         115  +
//! revalidates that route before cloning a transient request handle.
         116  +
//!
         117  +
//! ```text
         118  +
//! Client(partition, request)
         119  +
//! `-- OriginCell(partition, origin)
         120  +
//!     |-- local accepting generation ----------------> H2Activation
         121  +
//!     `-- acquisition queue
         122  +
//!         |-- local flight result --------------------> H2Activation
         123  +
//!         |-- eligible peer generation route --------> H2Activation
         124  +
//!         `-- capacity permit -> connect + ALPN
         125  +
//!             |-- HTTP/2 -> join or drive one flight -> H2Activation
         126  +
//!             `-- HTTP/1 -> H1Selection or incompatible-version error
         127  +
//!
         128  +
//! H2Activation -- Hyper accepts request --> accepted request lease
         129  +
//! accepted request lease
         130  +
//!     |-- request body ends or drops -----> send endpoint complete
         131  +
//!     `-- response body ends or drops ----> receive endpoint complete
         132  +
//! both endpoints complete ----------------> release generation request count
         133  +
//! ```
         134  +
//!
         135  +
//! `H2Activation` reserves pool accounting for a prospective stream on one
         136  +
//! exact generation. It is not yet an HTTP/2 stream. Dropping it before Hyper
         137  +
//! accepts the request returns its generation-gate turn and request count.
         138  +
//! Acceptance creates two independent endpoints because an upload and response
         139  +
//! can finish in either order. Logical close stops new activations and releases
         140  +
//! bounded capacity; accepted streams retain the draining generation until
         141  +
//! both endpoints end. Hyper remains responsible for stream identifiers,
         142  +
//! stream credit, and flow control.
         143  +
//!
         144  +
//! Peer publication moves only route identity. The socket, protocol driver,
         145  +
//! request handle, and capacity remain with the connection-owning partition.
         146  +
//!
   75    147   
//! `ConnectionState` separates logical close, accepted-request accounting, and
   76    148   
//! root-I/O ownership. Logical close rejects new dispatch and releases bounded
   77    149   
//! capacity. `DispatchGuard` follows an accepted request, while
   78    150   
//! `PhysicalConnectionGuard` follows root I/O until the pool no longer owns
   79    151   
//! that transport; neither describes the operating system TCP state. All
   80    152   
//! connection-owned work runs through the partition [`DriverSpawner`].
   81    153   
   82    154   
#![cfg_attr(
   83    155   
    smithy_http_client_loom,
   84    156   
    allow(

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

@@ -1,1 +88,110 @@
    3      3   
 * SPDX-License-Identifier: Apache-2.0
    4      4   
 */
    5      5   
    6      6   
//! Capacity and demand shared by the cells of one bounded origin.
    7      7   
//!
    8      8   
//! [`OriginAdmission`] stores the origin once and keys its partition cells,
    9      9   
//! demand records, and availability indexes by [`PartitionId`]. Its lock is the
   10     10   
//! sole authority for available capacity, origin-wide demand order, and reuse
   11     11   
//! operations. Cell locks are never held with the admission lock.
   12     12   
//!
   13         -
//! A bounded acquisition crosses the layer in this order:
          13  +
//! A bounded acquisition moves through the layer in this order:
   14     14   
//!
   15     15   
//! 1. A cell publishes a complete [`DemandSnapshot`] after releasing its lock.
   16     16   
//! 2. [`DemandSchedule`] replaces that partition snapshot and links active
   17         -
//!    demand in origin-wide FIFO order.
   18         -
//! 3. Admission takes one unit of capacity or reserves a peer HTTP/1 sender for
   19         -
//!    the oldest demand that can make progress.
   20         -
//! 4. [`DeliveryGuard`] carries the demand fence and payload out of admission.
   21         -
//!    It completes fallible connection-cell work before locking the requesting
   22         -
//!    cell. If peer-sender revalidation fails, no requesting waiter has been
   23         -
//!    reserved.
   24         -
//! 5. The requesting cell reserves its waiter, then `commit` transfers the
   25         -
//!    materialized payload into the waiter result.
   26         -
//! 6. `DeliveryAck` closes the fence. Rejection and drop return owned values
   27         -
//!    before admission makes the demand schedulable again.
          17  +
//!    demand in the origin order and its eligibility-group order.
          18  +
//! 3. Capacity delivery and HTTP/1 reuse select from the origin head. HTTP/2
          19  +
//!    publication pairs a group head with an advertised peer generation.
          20  +
//! 4. A [`DeliveryGuard`] materializes its one capacity or HTTP/1 payload before
          21  +
//!    reserving the requesting waiter. Payload materialization failure
          22  +
//!    therefore leaves the waiter in its existing residence.
          23  +
//! 5. An [`H2PublicationGuard`] carries identities through the same unlocked
          24  +
//!    handoff. It revalidates the connection-owning generation, then installs
          25  +
//!    a route and activation opportunity in the requesting cell.
          26  +
//! 6. The requesting cell becomes authoritative before either guard submits its
          27  +
//!    admission acknowledgement. Rejection and drop execute the same terminal
          28  +
//!    acknowledgement paths.
   28     29   
//!
   29         -
//! Values crossing either lock boundary own their fallback, so cancellation
   30         -
//! cannot lose capacity or an HTTP/1 sender.
          30  +
//! Capacity and HTTP/1 handoffs own the payload they must return on failure.
          31  +
//! HTTP/2 publication moves no payload: the connection-owning cell retains its
          32  +
//! request handle, driver, socket, and capacity. No cell lock is held with
          33  +
//! the admission lock, and connection-owning and requesting cell locks are
          34  +
//! never held together.
   31     35   
   32     36   
use super::cell::OriginCell;
   33     37   
use super::origin::OriginKey;
   34     38   
use super::partition::{EligibilityGroup, PartitionId};
   35     39   
use crate::sync::{Arc, Mutex, Weak};
   36     40   
use std::collections::HashMap;
   37     41   
use std::fmt;
   38     42   
use std::num::NonZeroUsize;
   39     43   
   40     44   
mod delivery;
   41     45   
mod demand;
   42     46   
mod order;
          47  +
mod publication;
   43     48   
pub(in crate::client::pool) mod reuse;
   44     49   
   45     50   
use self::demand::{DemandSchedule, PreparedCapacityDelivery};
   46     51   
use self::order::{IntrusiveLinks, IntrusiveOrder};
          52  +
use self::publication::{
          53  +
    H2Publication, H2PublicationGuard, H2ReclaimAction, PreparedH2Publication, PreparedH2Reclaim,
          54  +
};
   47     55   
pub(in crate::client::pool) use delivery::DeliveryGuard;
          56  +
pub(in crate::client::pool) use publication::H2AdvertisementSnapshot;
   48     57   
   49     58   
/// Protocol capability required by the head waiter in a cell.
   50     59   
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
   51     60   
pub(crate) enum ProtocolRequirement {
          61  +
    /// The waiter requires HTTP/1 wire semantics.
          62  +
    H1Required,
   52     63   
    /// The waiter may dispatch over HTTP/1 or HTTP/2.
   53     64   
    H1Compatible,
   54     65   
    /// The waiter requires HTTP/2.
   55         -
    #[allow(dead_code, reason = "reserved for HTTP/2-only acquisition")]
   56     66   
    H2Required,
   57     67   
}
   58     68   
          69  +
impl ProtocolRequirement {
          70  +
    /// Returns whether an HTTP/1 sender can satisfy this requirement.
          71  +
    pub(crate) fn accepts_h1(self) -> bool {
          72  +
        self != Self::H2Required
          73  +
    }
          74  +
          75  +
    /// Returns whether an HTTP/2 activation can satisfy this requirement.
          76  +
    pub(crate) fn accepts_h2(self) -> bool {
          77  +
        self != Self::H1Required
          78  +
    }
          79  +
}
          80  +
   59     81   
/// Identity of one cell-local demand generation.
   60     82   
///
   61     83   
/// A generation begins when a waiter becomes the cell's bounded-demand head.
   62     84   
/// The cell identity and this value together identify its publications.
   63     85   
#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
   64     86   
pub(crate) struct DemandId(u64);
   65     87   
   66     88   
impl DemandId {
   67     89   
    /// Constructs an identity allocated by one cell's monotonic demand counter.
   68     90   
    pub(crate) const fn from_u64(value: u64) -> Self {
@@ -117,139 +302,377 @@
  137    159   
        Self {
  138    160   
            id,
  139    161   
            version,
  140    162   
            state: DemandState::Active {
  141    163   
                head,
  142    164   
                eligibility_group,
  143    165   
            },
  144    166   
        }
  145    167   
    }
  146    168   
         169  +
    /// Returns whether this active demand can use an HTTP/2 generation.
         170  +
    pub(crate) fn accepts_h2(&self) -> bool {
         171  +
        matches!(
         172  +
            self.state,
         173  +
            DemandState::Active { head, .. } if head.accepts_h2()
         174  +
        )
         175  +
    }
         176  +
  147    177   
    /// Retires a demand generation at the supplied publication version.
  148    178   
    pub(crate) fn inactive(id: DemandId, version: SnapshotVersion) -> Self {
  149    179   
        Self {
  150    180   
            id,
  151    181   
            version,
  152    182   
            state: DemandState::Inactive,
  153    183   
        }
  154    184   
    }
  155    185   
         186  +
    /// Returns the demand identity for cross-module transition tests.
         187  +
    #[cfg(test)]
         188  +
    pub(in crate::client::pool) fn id_for_test(&self) -> DemandId {
         189  +
        self.id
         190  +
    }
         191  +
  156    192   
    /// Returns whether this snapshot still requests capacity.
  157         -
    fn is_active(&self) -> bool {
         193  +
    pub(in crate::client::pool) fn is_active(&self) -> bool {
  158    194   
        matches!(self.state, DemandState::Active { .. })
  159    195   
    }
  160    196   
  161    197   
    /// Returns whether this snapshot may replace `current`.
  162    198   
    ///
  163    199   
    /// A newer generation supersedes every snapshot of an older generation;
  164    200   
    /// publications within one generation are ordered by snapshot version.
  165    201   
    fn is_newer_than(&self, current: &Self) -> bool {
  166    202   
        self.id > current.id || (self.id == current.id && self.version > current.version)
  167    203   
    }
  168    204   
}
  169    205   
  170    206   
/// One conserved unit of bounded-origin connection capacity.
  171    207   
///
  172    208   
/// This non-`Copy` value proves that admission removed one unit from its
  173    209   
/// available count. Identities are never reused so diagnostics remain
  174    210   
/// unambiguous.
  175    211   
pub(super) struct Permit(u64);
  176    212   
  177    213   
impl fmt::Debug for Permit {
  178    214   
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
  179    215   
        f.debug_tuple("Permit").field(&self.0).finish()
  180    216   
    }
  181    217   
}
  182    218   
  183    219   
/// Stable identity of one admission-to-cell delivery.
  184    220   
#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
  185    221   
pub(super) struct DeliveryId(u64);
  186    222   
  187    223   
/// Shared admission authority for one bounded origin.
  188    224   
///
  189    225   
/// The state lock is the only authority for available capacity while it
  190    226   
/// resides in admission, and for demand order while no delivery is crossing
  191    227   
/// to a cell. Delivery work is detached before this lock is released.
  192    228   
pub(crate) struct OriginAdmission {
  193    229   
    /// Canonical origin shared by every partition represented in `state`.
  194    230   
    origin: OriginKey,
         231  +
    /// Whether admission may close idle H2 capacity for H1-required demand.
         232  +
    allow_h2_reclaim_for_h1: bool,
  195    233   
    /// Capacity, demand, and delivery-fence state for this origin.
  196    234   
    state: Mutex<AdmissionState>,
  197    235   
}
  198    236   
  199    237   
impl OriginAdmission {
  200    238   
    /// Creates admission for at most `limit` logically open connections.
  201         -
    pub(crate) fn new(origin: OriginKey, limit: NonZeroUsize) -> Arc<Self> {
         239  +
    pub(crate) fn new(
         240  +
        origin: OriginKey,
         241  +
        limit: NonZeroUsize,
         242  +
        allow_h2_reclaim_for_h1: bool,
         243  +
    ) -> Arc<Self> {
  202    244   
        Arc::new(Self {
  203    245   
            origin,
         246  +
            allow_h2_reclaim_for_h1,
  204    247   
            state: Mutex::new(AdmissionState::new(limit)),
  205    248   
        })
  206    249   
    }
  207    250   
  208    251   
    #[cfg(test)]
  209    252   
    pub(super) fn for_test(limit: NonZeroUsize) -> Arc<Self> {
         253  +
        Self::for_test_with_h2_reclaim(limit, true)
         254  +
    }
         255  +
         256  +
    #[cfg(test)]
         257  +
    pub(super) fn for_test_with_h2_reclaim(
         258  +
        limit: NonZeroUsize,
         259  +
        allow_h2_reclaim_for_h1: bool,
         260  +
    ) -> Arc<Self> {
  210    261   
        Self::new(
  211    262   
            OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None)
  212    263   
                .expect("test origin is valid"),
  213    264   
            limit,
         265  +
            allow_h2_reclaim_for_h1,
  214    266   
        )
  215    267   
    }
  216    268   
  217    269   
    /// Returns the canonical origin coordinated by this admission authority.
  218    270   
    fn origin(&self) -> &OriginKey {
  219    271   
        &self.origin
  220    272   
    }
  221    273   
  222    274   
    /// Registers or returns the unique retained cell for an identity.
  223    275   
    pub(crate) fn register_cell(origin: &Arc<Self>, candidate: Arc<OriginCell>) -> Arc<OriginCell> {
  224    276   
        let partition = candidate.id().partition();
  225    277   
        assert_eq!(
  226    278   
            origin.origin(),
  227    279   
            candidate.id().origin(),
  228    280   
            "cell origin did not match its admission authority"
  229    281   
        );
  230    282   
        let mut state = origin.state.lock();
  231    283   
        if let Some(existing) = state.cells.get(&partition).and_then(Weak::upgrade) {
  232    284   
            return existing;
  233    285   
        }
  234    286   
        state.cells.insert(partition, Weak::from_arc(&candidate));
  235    287   
        candidate
  236    288   
    }
  237    289   
  238    290   
    /// Publishes a complete demand snapshot and drives any resulting delivery.
  239    291   
    pub(crate) fn publish_demand(
  240    292   
        origin: &Arc<Self>,
  241    293   
        requesting_partition: PartitionId,
  242    294   
        snapshot: DemandSnapshot,
  243    295   
    ) {
  244    296   
        let action = {
  245    297   
            let mut state = origin.state.lock();
  246    298   
            state.publish_demand(requesting_partition, snapshot);
  247    299   
            Self::prepare_action(origin, &mut state)
  248    300   
        };
  249    301   
        Self::drive(action);
  250    302   
    }
  251    303   
  252         -
    /// Extracts the next bounded-origin action while admission is locked.
         304  +
    /// Selects at most one action while admission is locked.
         305  +
    ///
         306  +
    /// Cancellation cleanup runs first, followed by available capacity,
         307  +
    /// compatible HTTP/2 publication, HTTP/1 reuse, and idle-H2 reclaim. The
         308  +
    /// returned action owns everything needed to run after releasing the
         309  +
    /// admission lock. Its completion prepares the next action, forming an
         310  +
    /// iterative pump without nesting admission and cell locks.
  253    311   
    fn prepare_action(origin: &Arc<Self>, state: &mut AdmissionState) -> Option<AdmissionAction> {
  254    312   
        if let Some(cancellation) = state.h1.prepare_cancellation() {
  255    313   
            return Some(AdmissionAction::H1(reuse::H1ReuseAction::cancel(
  256    314   
                origin.clone(),
  257    315   
                cancellation,
  258    316   
            )));
  259    317   
        }
  260    318   
        if let Some(pending) = state.schedule_one() {
  261    319   
            return Some(AdmissionAction::Delivery(DeliveryGuard::capacity(
  262    320   
                origin.clone(),
  263    321   
                pending.delivery,
  264    322   
                pending.requesting_partition,
  265    323   
                pending.demand,
  266    324   
                pending.permit,
  267    325   
            )));
  268    326   
        }
         327  +
        if let Some(publication) = state.prepare_h2_publication() {
         328  +
            return Some(AdmissionAction::H2(H2PublicationGuard::new(
         329  +
                origin.clone(),
         330  +
                publication,
         331  +
            )));
         332  +
        }
         333  +
        if let Some(reuse) = state.h1.prepare_reuse(&state.demand_schedule) {
         334  +
            return Some(AdmissionAction::H1(reuse::H1ReuseAction::install(
         335  +
                origin.clone(),
         336  +
                reuse,
         337  +
            )));
         338  +
        }
         339  +
        if !origin.allow_h2_reclaim_for_h1 {
         340  +
            return None;
         341  +
        }
  269    342   
        state
  270         -
            .h1
  271         -
            .prepare_reuse(&state.demand_schedule)
  272         -
            .map(|reuse| AdmissionAction::H1(reuse::H1ReuseAction::install(origin.clone(), reuse)))
         343  +
            .h2
         344  +
            .prepare_reclaim(&state.demand_schedule)
         345  +
            .map(|reclaim| {
         346  +
                AdmissionAction::H2Reclaim(H2ReclaimAction::new(origin.clone(), reclaim))
         347  +
            })
  273    348   
    }
  274    349   
  275    350   
    /// Upgrades a registered requesting cell without holding the admission lock.
  276    351   
    fn cell(&self, id: &PartitionId) -> Option<Arc<OriginCell>> {
  277    352   
        let cell = {
  278    353   
            let state = self.state.lock();
  279    354   
            state.cells.get(id).cloned()
  280    355   
        };
  281    356   
        cell.and_then(|cell| cell.upgrade())
  282    357   
    }
@@ -304,379 +444,586 @@
  324    399   
        result: DeliveryAckResult,
  325    400   
    ) -> Option<AdmissionAction> {
  326    401   
        let mut state = origin.state.lock();
  327    402   
        if let Some(permit) = permit {
  328    403   
            state.return_permit(permit);
  329    404   
        }
  330    405   
        state.finish_delivery(delivery, requesting_partition, result);
  331    406   
        Self::prepare_action(origin, &mut state)
  332    407   
    }
  333    408   
         409  +
    /// Replaces one connection cell's complete HTTP/2 advertisement.
         410  +
    pub(in crate::client::pool) fn update_h2_advertisement(
         411  +
        origin: &Arc<Self>,
         412  +
        connection_partition: PartitionId,
         413  +
        group: EligibilityGroup,
         414  +
        snapshot: H2AdvertisementSnapshot,
         415  +
    ) {
         416  +
        let action = {
         417  +
            let mut state = origin.state.lock();
         418  +
            let AdmissionState {
         419  +
                h2,
         420  +
                demand_schedule,
         421  +
                ..
         422  +
            } = &mut *state;
         423  +
            h2.update(connection_partition, group, snapshot, demand_schedule);
         424  +
            Self::prepare_action(origin, &mut state)
         425  +
        };
         426  +
        Self::drive(action);
         427  +
    }
         428  +
         429  +
    /// Applies one H2 publication acknowledgement and prepares its successor.
         430  +
    fn finish_h2_publication(
         431  +
        origin: &Arc<Self>,
         432  +
        prepared: &PreparedH2Publication,
         433  +
        stale_generation: Option<super::cell::h2::H2GenerationId>,
         434  +
        result: DeliveryAckResult,
         435  +
    ) -> Option<AdmissionAction> {
         436  +
        let mut state = origin.state.lock();
         437  +
        if let Some(generation) = stale_generation {
         438  +
            let AdmissionState {
         439  +
                h2,
         440  +
                demand_schedule,
         441  +
                ..
         442  +
            } = &mut *state;
         443  +
            h2.remove_if_exact(&prepared.connection_partition, generation, demand_schedule);
         444  +
        }
         445  +
        state.finish_delivery(prepared.delivery, &prepared.requesting_partition, result);
         446  +
        Self::prepare_action(origin, &mut state)
         447  +
    }
         448  +
         449  +
    /// Completes one exact idle-H2 reclaim crossing.
         450  +
    fn finish_h2_reclaim(
         451  +
        origin: &Arc<Self>,
         452  +
        prepared: &PreparedH2Reclaim,
         453  +
        snapshot: Option<H2AdvertisementSnapshot>,
         454  +
    ) -> Option<AdmissionAction> {
         455  +
        let mut state = origin.state.lock();
         456  +
        let AdmissionState {
         457  +
            h2,
         458  +
            demand_schedule,
         459  +
            ..
         460  +
        } = &mut *state;
         461  +
        h2.finish_reclaim(prepared, snapshot, demand_schedule);
         462  +
        Self::prepare_action(origin, &mut state)
         463  +
    }
  334    464   
    #[cfg(test)]
  335    465   
    pub(super) fn publish_action_without_driving(
  336    466   
        origin: &Arc<Self>,
  337    467   
        requesting_partition: PartitionId,
  338    468   
        snapshot: DemandSnapshot,
  339    469   
    ) -> Option<AdmissionAction> {
  340    470   
        let mut state = origin.state.lock();
  341    471   
        state.publish_demand(requesting_partition, snapshot);
  342    472   
        Self::prepare_action(origin, &mut state)
  343    473   
    }
  344    474   
  345    475   
    #[cfg(test)]
  346    476   
    pub(super) fn publish_without_driving(
  347    477   
        origin: &Arc<Self>,
  348    478   
        requesting_partition: PartitionId,
  349    479   
        snapshot: DemandSnapshot,
  350    480   
    ) -> Option<DeliveryGuard> {
  351    481   
        match Self::publish_action_without_driving(origin, requesting_partition, snapshot) {
  352    482   
            Some(AdmissionAction::Delivery(delivery)) => Some(delivery),
  353    483   
            Some(AdmissionAction::H1(_)) => {
  354    484   
                panic!("capacity-only test unexpectedly prepared an HTTP/1 action")
  355    485   
            }
         486  +
            Some(AdmissionAction::H2(_)) => {
         487  +
                panic!("capacity-only test unexpectedly prepared an HTTP/2 action")
         488  +
            }
         489  +
            Some(AdmissionAction::H2Reclaim(_)) => {
         490  +
                panic!("capacity-only test unexpectedly prepared an HTTP/2 reclaim action")
         491  +
            }
  356    492   
            None => None,
  357    493   
        }
  358    494   
    }
  359    495   
  360    496   
    #[cfg(test)]
  361    497   
    fn counts(&self) -> AdmissionCounts {
  362    498   
        let state = self.state.lock();
  363    499   
        AdmissionCounts {
  364    500   
            limit: state.limit,
  365    501   
            available: state.available_capacity(),
  366    502   
            ordered: state.demand_schedule.len(),
  367    503   
            queued: state.demand_schedule.queued_len(),
  368    504   
            delivering: state.demand_schedule.delivering_len(),
  369    505   
        }
  370    506   
    }
  371    507   
  372    508   
    #[cfg(test)]
  373    509   
    pub(super) fn lease_for_test(origin: &Arc<Self>) -> CapacityLease {
  374    510   
        let permit = origin
  375    511   
            .state
  376    512   
            .lock()
  377    513   
            .take_permit()
  378    514   
            .expect("test origin had no available capacity");
  379    515   
        CapacityLease::new(origin.clone(), permit)
  380    516   
    }
  381    517   
  382    518   
    #[cfg(test)]
  383    519   
    pub(super) fn available_capacity_for_test(&self) -> usize {
  384    520   
        self.state.lock().available_capacity()
  385    521   
    }
  386    522   
  387    523   
    #[cfg(test)]
  388    524   
    pub(super) fn ordered_demand_count_for_test(&self) -> usize {
  389    525   
        self.state.lock().demand_schedule.len()
  390    526   
    }
  391    527   
  392    528   
    #[cfg(all(test, smithy_http_client_loom))]
  393    529   
    pub(super) fn clear_modeled_cells_for_test(&self) {
  394    530   
        // Loom has no modeled Weak, so its synchronization facade retains
  395    531   
        // cells strongly. Explicit teardown prevents that model-only
  396    532   
        // substitution from appearing as an Arc leak.
  397    533   
        self.state.lock().cells.clear();
  398    534   
    }
  399    535   
}
  400    536   
  401         -
/// One unlocked step prepared while holding the bounded-origin lock.
         537  +
/// One detached step prepared while holding the bounded-origin lock.
  402    538   
pub(super) enum AdmissionAction {
  403         -
    /// One capacity or borrowed-H1 payload crossing to a requesting cell.
         539  +
    /// One capacity or borrowed-H1 payload handed to a requesting cell.
  404    540   
    Delivery(DeliveryGuard),
  405    541   
    /// HTTP/1 availability, reservation, or borrowed-sender work.
  406    542   
    H1(reuse::H1ReuseAction),
         543  +
    /// HTTP/2 generation visibility handed to one requesting cell.
         544  +
    H2(H2PublicationGuard),
         545  +
    /// Exact idle HTTP/2 generation reserved at its connection cell.
         546  +
    H2Reclaim(H2ReclaimAction),
  407    547   
}
  408    548   
  409    549   
impl AdmissionAction {
  410         -
    /// Executes one lock-domain crossing and returns the next prepared step.
         550  +
    /// Executes one detached action and returns the next prepared step.
  411    551   
    fn drive_once(self) -> Option<Self> {
  412    552   
        match self {
  413    553   
            Self::Delivery(delivery) => delivery.deliver_once(),
  414    554   
            Self::H1(action) => action.drive_once(),
         555  +
            Self::H2(publication) => publication.publish_once(),
         556  +
            Self::H2Reclaim(reclaim) => reclaim.drive_once(),
  415    557   
        }
  416    558   
    }
  417    559   
  418    560   
    /// Advances one crossing without recursively driving its successor.
  419    561   
    #[cfg(all(test, smithy_http_client_loom))]
  420    562   
    pub(super) fn drive_once_for_test(self) -> Option<Self> {
  421    563   
        self.drive_once()
  422    564   
    }
  423    565   
}
  424    566   
@@ -470,612 +630,810 @@
  490    632   
    /// Requesting cells, held weakly to avoid an ownership cycle.
  491    633   
    cells: HashMap<PartitionId, Weak<OriginCell>>,
  492    634   
    /// Permits not owned by a delivery, attempt, or connection.
  493    635   
    available: usize,
  494    636   
    /// Next never-reused permit identity.
  495    637   
    next_permit_id: u64,
  496    638   
    /// Cross-cell demand records and their origin-wide scheduling order.
  497    639   
    demand_schedule: DemandSchedule,
  498    640   
    /// HTTP/1 availability reports and cross-cell reuse operations.
  499    641   
    h1: reuse::H1Reuse,
         642  +
    /// HTTP/2 advertisements and eligibility-group publication turns.
         643  +
    h2: H2Publication,
  500    644   
    /// Next never-reused delivery identity.
  501    645   
    next_delivery: u64,
  502    646   
    /// Configured connection bound used to check capacity conservation.
  503    647   
    limit: usize,
  504    648   
}
  505    649   
  506    650   
impl AdmissionState {
  507    651   
    /// Creates a ledger with all configured capacity available.
  508    652   
    fn new(limit: NonZeroUsize) -> Self {
  509    653   
        let limit = limit.get();
  510    654   
        Self {
  511    655   
            cells: HashMap::new(),
  512    656   
            available: limit,
  513    657   
            next_permit_id: 0,
  514    658   
            demand_schedule: DemandSchedule::default(),
  515    659   
            h1: reuse::H1Reuse::default(),
         660  +
            h2: H2Publication::default(),
  516    661   
            next_delivery: 0,
  517    662   
            limit,
  518    663   
        }
  519    664   
    }
  520    665   
  521    666   
    /// Removes one available permit from admission.
  522    667   
    fn take_permit(&mut self) -> Option<Permit> {
  523    668   
        if self.available == 0 {
  524    669   
            return None;
  525    670   
        }
  526    671   
        let id = self.next_permit_id;
  527    672   
        let next_permit_id = id.checked_add(1).expect("permit identity exhausted");
  528    673   
        self.available -= 1;
  529    674   
        self.next_permit_id = next_permit_id;
  530    675   
        Some(Permit(id))
  531    676   
    }
  532    677   
  533    678   
    /// Consumes a returned permit and restores one available slot.
  534    679   
    fn return_permit(&mut self, _permit: Permit) {
  535    680   
        let available = self
  536    681   
            .available
  537    682   
            .checked_add(1)
  538    683   
            .expect("available capacity count overflowed");
  539    684   
        assert!(
  540    685   
            available <= self.limit,
  541    686   
            "available capacity exceeded the configured limit"
  542    687   
        );
  543    688   
        self.available = available;
  544    689   
    }
  545    690   
  546    691   
    #[cfg(test)]
  547    692   
    fn available_capacity(&self) -> usize {
  548    693   
        self.available
  549    694   
    }
  550    695   
  551    696   
    /// Applies one complete cell publication to cross-cell scheduling.
  552    697   
    fn publish_demand(&mut self, requesting_partition: PartitionId, snapshot: DemandSnapshot) {
         698  +
        let old_group = self.demand_schedule.group_for(&requesting_partition);
  553    699   
        self.demand_schedule.publish(requesting_partition, snapshot);
         700  +
        self.reconcile_demand_indexes(&requesting_partition, old_group);
         701  +
    }
         702  +
         703  +
    /// Refreshes protocol indexes derived from the canonical demand schedule.
         704  +
    ///
         705  +
    /// HTTP/1 indexes depend on the requesting partition's origin position.
         706  +
    /// HTTP/2 publication depends on both the previous and current eligibility
         707  +
    /// groups, so a group change must repair each view. The derived indexes do
         708  +
    /// not own demand or ordering.
         709  +
    fn reconcile_demand_indexes(
         710  +
        &mut self,
         711  +
        requesting_partition: &PartitionId,
         712  +
        old_group: Option<EligibilityGroup>,
         713  +
    ) {
  554    714   
        self.h1
  555         -
            .reconcile_requesting_cell(&requesting_partition, &self.demand_schedule);
         715  +
            .reconcile_requesting_cell(requesting_partition, &self.demand_schedule);
         716  +
        if let Some(old_group) = old_group {
         717  +
            self.h2.reconcile_group(&old_group, &self.demand_schedule);
         718  +
        }
         719  +
        if let Some(group) = self.demand_schedule.group_for(requesting_partition) {
         720  +
            self.h2.reconcile_group(&group, &self.demand_schedule);
         721  +
        }
  556    722   
    }
  557    723   
  558    724   
    /// Pairs the oldest deliverable demand with one available permit.
  559    725   
    fn schedule_one(&mut self) -> Option<PreparedCapacityDelivery> {
  560    726   
        if !self.demand_schedule.head_is_queued() {
  561    727   
            return None;
  562    728   
        }
  563    729   
  564    730   
        let permit = self.take_permit()?;
  565    731   
        let delivery = self.take_delivery_id();
         732  +
        let old_group = self
         733  +
            .demand_schedule
         734  +
            .queued_head()
         735  +
            .and_then(|head| self.demand_schedule.group_for(&head.requesting_partition));
  566    736   
        let scheduled = self
  567    737   
            .demand_schedule
  568         -
            .reserve_head(delivery)
         738  +
            .reserve_origin_head(delivery)
  569    739   
            .expect("queued demand head disappeared");
  570         -
        self.h1
  571         -
            .reconcile_requesting_cell(&scheduled.requesting_partition, &self.demand_schedule);
         740  +
        self.reconcile_demand_indexes(&scheduled.requesting_partition, old_group);
  572    741   
        Some(PreparedCapacityDelivery {
  573    742   
            permit,
  574    743   
            delivery,
  575    744   
            requesting_partition: scheduled.requesting_partition,
  576    745   
            demand: scheduled.demand,
  577    746   
        })
  578    747   
    }
  579    748   
  580    749   
    /// Delegates delivery-fence revalidation to the demand schedule.
  581    750   
    #[cfg(test)]
  582    751   
    fn delivery_is_current(
  583    752   
        &self,
  584    753   
        delivery: DeliveryId,
  585    754   
        requesting_partition: &PartitionId,
  586    755   
        demand: DemandId,
  587    756   
    ) -> bool {
  588    757   
        self.demand_schedule
  589    758   
            .delivery_is_current(delivery, requesting_partition, demand)
  590    759   
    }
  591    760   
  592    761   
    /// Applies a requesting cell acknowledgement to the demand schedule.
  593    762   
    fn finish_delivery(
  594    763   
        &mut self,
  595    764   
        delivery: DeliveryId,
  596    765   
        requesting_partition: &PartitionId,
  597    766   
        result: DeliveryAckResult,
  598    767   
    ) {
         768  +
        let old_group = self.demand_schedule.group_for(requesting_partition);
  599    769   
        self.demand_schedule
  600    770   
            .finish_delivery(delivery, requesting_partition, result);
         771  +
        self.reconcile_demand_indexes(requesting_partition, old_group);
         772  +
    }
         773  +
         774  +
    /// Reserves one eligibility-group head for identity-only H2 publication.
         775  +
    fn prepare_h2_publication(&mut self) -> Option<PreparedH2Publication> {
         776  +
        if !self.h2.has_ready_group() {
         777  +
            return None;
         778  +
        }
         779  +
        let delivery = self.take_delivery_id();
         780  +
        self.h2.prepare(&mut self.demand_schedule, delivery)
  601    781   
    }
  602    782   
  603    783   
    /// Allocates a delivery-fence identity that is never reused by this origin.
  604    784   
    fn take_delivery_id(&mut self) -> DeliveryId {
  605    785   
        let value = self.next_delivery;
  606    786   
        self.next_delivery = value.checked_add(1).expect("delivery identity exhausted");
  607    787   
        DeliveryId(value)
  608    788   
    }
  609    789   
}
  610    790   

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

@@ -383,383 +442,458 @@
  403    403   
        /// Connection whose sender crossed to the requesting cell.
  404    404   
        connection_id: ConnectionId,
  405    405   
        /// Reuse operation completed by this acknowledgement.
  406    406   
        reuse_id: ReuseId,
  407    407   
        /// Cell whose local reuse reservation must complete.
  408    408   
        connection_partition: PartitionId,
  409    409   
    },
  410    410   
}
  411    411   
  412    412   
impl DeliveryAck {
         413  +
    /// Avoids republishing a successor already covered by visible HTTP/2 state.
         414  +
    ///
         415  +
    /// Capacity or HTTP/1 delivery may reveal another waiter after the
         416  +
    /// requesting cell already gained a local generation or peer route. That
         417  +
    /// visible HTTP/2 state can serve an H2-compatible successor directly. If
         418  +
    /// it later closes, its close path publishes the cell's current demand.
         419  +
    pub(in crate::client::pool) fn suppress_h2_successor(&mut self) {
         420  +
        if self
         421  +
            .successor
         422  +
            .as_ref()
         423  +
            .is_some_and(DemandSnapshot::accepts_h2)
         424  +
        {
         425  +
            self.successor = None;
         426  +
        }
         427  +
    }
         428  +
  413    429   
    /// Acknowledges that requesting cell state accepted the acquisition event.
  414    430   
    pub(in crate::client::pool) fn accept(mut self) -> Option<AdmissionAction> {
  415    431   
        let kind = self
  416    432   
            .kind
  417    433   
            .take()
  418    434   
            .expect("delivery acknowledgement completed more than once");
  419    435   
        let successor = self.successor.take();
  420    436   
        self.finish(kind, DeliveryAckResult::Accepted { successor }, None)
  421    437   
    }
  422    438   

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

@@ -1,1 +364,548 @@
    1      1   
/*
    2      2   
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
    3      3   
 * SPDX-License-Identifier: Apache-2.0
    4      4   
 */
    5      5   
    6      6   
//! Versioned cross-cell demand scheduling for one bounded origin.
    7      7   
//!
    8         -
//! Each retained cell publishes a complete demand snapshot. Active snapshots
    9         -
//! occupy one origin-wide FIFO, and a delivery remains at the head as a fence
   10         -
//! until its requesting cell acknowledges ownership or rejection.
           8  +
//! Each cell publishes at most one active demand: its acquisition queue's
           9  +
//! current head. That demand may be satisfied either by origin-wide connection
          10  +
//! capacity or by an existing HTTP/2 connection in its eligibility group.
          11  +
//! These resources require different orderings.
          12  +
//!
          13  +
//! For example, the origin order may contain `A(group X), B(group Y)` while
          14  +
//! only group Y has a reusable HTTP/2 connection. A remains first for the next
          15  +
//! available connection permit, while the group-Y connection can serve B
          16  +
//! without consuming that permit or delaying A. The group order finds B
          17  +
//! directly instead of scanning every partition.
          18  +
//!
          19  +
//! The same demand occupies both orders because capacity and HTTP/2 publication
          20  +
//! can race to satisfy it. Reserving either position retains a fence in both
          21  +
//! orders until the requesting cell accepts or rejects the handoff. The other
          22  +
//! resource therefore cannot serve the same demand while the first handoff is
          23  +
//! running outside the admission lock.
   11     24   
   12     25   
use super::{
   13     26   
    DeliveryAckResult, DeliveryId, DemandId, DemandSnapshot, DemandState, IntrusiveLinks,
   14     27   
    IntrusiveOrder, Permit, ProtocolRequirement,
   15     28   
};
   16     29   
use crate::client::pool::partition::{EligibilityGroup, PartitionId};
   17     30   
use std::collections::HashMap;
   18     31   
   19         -
/// Cross-cell demand records and their origin-wide scheduling order.
          32  +
/// Cross-cell demand records and their scheduling orders.
   20     33   
///
   21     34   
/// At every completed transition:
   22     35   
///
   23     36   
/// - `records` owns the newest snapshot for every retained cell.
   24         -
/// - `IntrusiveOrder::Active` contains exactly the `Queued` and `Delivering`
   25         -
///   records.
   26         -
/// - Queue links live inside those ordered residence variants.
   27         -
/// - A `Delivering` record remains at the head as a scheduling fence.
          37  +
/// - the origin order and the record's eligibility-group order contain exactly
          38  +
///   the `Queued` and `Delivering` records;
          39  +
/// - links for both views live inside those ordered residence variants;
          40  +
/// - an origin delivery fences the origin head; and
          41  +
/// - an HTTP/2 publication fences its eligibility-group head.
   28     42   
///
   29     43   
/// Admission coordinates capacity extraction with this schedule while holding
   30     44   
/// the same origin lock.
   31     45   
#[derive(Debug, Default)]
   32     46   
pub(super) struct DemandSchedule {
   33     47   
    /// Latest demand and scheduling residence for each retained cell.
   34     48   
    records: HashMap<PartitionId, DemandRecord>,
   35         -
    /// Origin-wide order, including an outstanding delivery fence.
   36         -
    order: IntrusiveOrder<PartitionId>,
          49  +
    /// Origin-wide order used by one-to-one capacity and HTTP/1 reuse.
          50  +
    origin_order: IntrusiveOrder<PartitionId>,
          51  +
    /// Reusable-HTTP/2 demand order for each connection-reuse group.
          52  +
    group_orders: HashMap<EligibilityGroup, IntrusiveOrder<PartitionId>>,
   37     53   
}
   38     54   
   39     55   
/// Complete origin-order head used to choose one HTTP/1 reuse action.
   40     56   
#[derive(Clone, Debug)]
   41     57   
pub(super) struct QueuedDemand {
   42     58   
    /// Cell whose oldest waiter owns this demand generation.
   43     59   
    pub(super) requesting_partition: PartitionId,
   44     60   
    /// Demand identity revalidated at each crossing.
   45     61   
    pub(super) demand: DemandId,
   46     62   
    /// Protocol capability required by the requesting cell waiter.
   47     63   
    pub(super) requirement: ProtocolRequirement,
   48     64   
    /// Connection-owning cells whose H1 senders may satisfy this demand.
   49     65   
    pub(super) eligibility_group: EligibilityGroup,
   50     66   
}
   51     67   
   52     68   
/// Latest snapshot and scheduling residence for one stable cell.
   53     69   
#[derive(Debug)]
   54     70   
struct DemandRecord {
   55     71   
    /// Newest complete publication observed for the cell.
   56     72   
    pub(super) latest: DemandSnapshot,
          73  +
    /// Stable group retained while an inactive replacement crosses a fence.
          74  +
    group: Option<EligibilityGroup>,
   57     75   
    /// Scheduling residence, including links while ordered.
   58     76   
    residence: DemandResidence,
   59     77   
}
   60     78   
          79  +
/// Links retained by one demand in both scheduling views.
          80  +
#[derive(Clone, Debug)]
          81  +
struct DemandLinks {
          82  +
    /// Position in origin-wide capacity and HTTP/1 order.
          83  +
    origin: IntrusiveLinks<PartitionId>,
          84  +
    /// Position in the demand's all-protocol eligibility-group order.
          85  +
    group: IntrusiveLinks<PartitionId>,
          86  +
}
          87  +
          88  +
/// Order whose head is fenced by one delivery crossing.
          89  +
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
          90  +
enum DeliveryView {
          91  +
    /// Capacity or HTTP/1 delivery selected from origin order.
          92  +
    Origin,
          93  +
    /// HTTP/2 publication selected from eligibility-group order.
          94  +
    Group,
          95  +
}
          96  +
          97  +
/// Selects one of a demand residence's intrusive link sets.
          98  +
#[derive(Clone, Copy)]
          99  +
enum DemandOrderView {
         100  +
    /// Origin-wide capacity and HTTP/1 order.
         101  +
    Origin,
         102  +
    /// Eligibility-group all-protocol order.
         103  +
    Group,
         104  +
}
         105  +
   61    106   
/// Admission residence for one partition's latest demand.
   62    107   
///
   63    108   
/// ```text
   64    109   
/// Idle -- active publication ------------------------------> Queued
   65    110   
/// Queued -- reserve head ----------------------------------> Delivering
   66    111   
/// Queued -- inactive or replacement publication ----------> Idle
   67    112   
/// Delivering -- retry unchanged demand --------------------> Queued
   68    113   
/// Delivering -- accept, reject, or replacement retry ------> Idle
   69    114   
/// Idle -- active successor remains ------------------------> Queued
   70    115   
/// ```
   71    116   
#[derive(Clone, Debug)]
   72    117   
enum DemandResidence {
   73    118   
    /// The cell has no demand represented in scheduling.
   74    119   
    Idle,
   75    120   
    /// The demand is waiting in origin order.
   76    121   
    Queued {
   77    122   
        /// Demand generation represented by this residence.
   78    123   
        demand: DemandId,
   79         -
        /// Origin-wide scheduling links and arrival sequence.
   80         -
        links: IntrusiveLinks<PartitionId>,
         124  +
        /// Origin-wide and eligibility-group scheduling links.
         125  +
        links: DemandLinks,
   81    126   
    },
   82         -
    /// One delivery guard owns capacity for this demand.
         127  +
    /// One delivery or publication guard fences this demand.
   83    128   
    Delivering {
   84         -
        /// Demand generation fenced at the order head.
         129  +
        /// Demand generation fenced at one scheduling head.
   85    130   
        demand: DemandId,
         131  +
        /// Snapshot version current when admission created the fence.
         132  +
        version: super::SnapshotVersion,
   86    133   
        /// Delivery allowed to complete this fence.
   87    134   
        delivery: DeliveryId,
   88         -
        /// Origin-wide scheduling links retained until acknowledgement.
   89         -
        links: IntrusiveLinks<PartitionId>,
         135  +
        /// Scheduling view whose head owns this crossing.
         136  +
        #[cfg_attr(
         137  +
            not(debug_assertions),
         138  +
            allow(
         139  +
                dead_code,
         140  +
                reason = "the scheduling view is retained for debug invariant checks"
         141  +
            )
         142  +
        )]
         143  +
        view: DeliveryView,
         144  +
        /// Both scheduling positions retained until acknowledgement.
         145  +
        links: DemandLinks,
   90    146   
    },
   91    147   
}
   92    148   
   93    149   
impl DemandResidence {
   94         -
    #[cfg(debug_assertions)]
   95         -
    /// Borrows order links while this record is queued or delivering.
   96         -
    fn links(&self) -> Option<&IntrusiveLinks<PartitionId>> {
   97         -
        match self {
   98         -
            Self::Idle => None,
   99         -
            Self::Queued { links, .. } | Self::Delivering { links, .. } => Some(links),
         150  +
    #[cfg(any(debug_assertions, test))]
         151  +
    /// Borrows one view's links while this record is ordered.
         152  +
    fn links(&self, view: DemandOrderView) -> Option<&IntrusiveLinks<PartitionId>> {
         153  +
        let links = match self {
         154  +
            Self::Idle => return None,
         155  +
            Self::Queued { links, .. } | Self::Delivering { links, .. } => links,
         156  +
        };
         157  +
        match view {
         158  +
            DemandOrderView::Origin => Some(&links.origin),
         159  +
            DemandOrderView::Group => Some(&links.group),
  100    160   
        }
  101    161   
    }
  102    162   
  103         -
    /// Mutably borrows links from a record known to be ordered.
         163  +
    /// Mutably borrows one view's links from an ordered record.
  104    164   
    ///
  105    165   
    /// # Panics
  106    166   
    ///
  107    167   
    /// Panics when called for an idle record.
  108         -
    fn links_mut(&mut self) -> &mut IntrusiveLinks<PartitionId> {
  109         -
        match self {
  110         -
            Self::Idle => panic!("idle demand has no order links"),
         168  +
    fn links_mut(&mut self, view: DemandOrderView) -> &mut IntrusiveLinks<PartitionId> {
         169  +
        let links = match self {
         170  +
            Self::Idle => panic!("idle demand has no scheduling links"),
  111    171   
            Self::Queued { links, .. } | Self::Delivering { links, .. } => links,
         172  +
        };
         173  +
        match view {
         174  +
            DemandOrderView::Origin => &mut links.origin,
         175  +
            DemandOrderView::Group => &mut links.group,
  112    176   
        }
  113    177   
    }
  114    178   
  115         -
    /// Detaches links while moving a record out of origin-wide order.
  116         -
    fn into_links(self) -> Option<IntrusiveLinks<PartitionId>> {
         179  +
    /// Detaches both link sets while moving a record out of scheduling.
         180  +
    fn into_links(self) -> Option<DemandLinks> {
  117    181   
        match self {
  118    182   
            Self::Idle => None,
  119    183   
            Self::Queued { links, .. } | Self::Delivering { links, .. } => Some(links),
  120    184   
        }
  121    185   
    }
  122    186   
}
  123    187   
  124    188   
/// Demand reserved at the order head for one capacity delivery.
  125    189   
pub(super) struct ScheduledDemand {
  126    190   
    /// Selected destination cell.
  127    191   
    pub(super) requesting_partition: PartitionId,
  128    192   
    /// Demand identity current when the crossing was prepared.
  129    193   
    pub(super) demand: DemandId,
  130    194   
}
  131    195   
  132    196   
/// Connection capacity extracted under admission lock before crossing to a cell.
  133    197   
#[derive(Debug)]
  134    198   
pub(super) struct PreparedCapacityDelivery {
  135    199   
    /// Permit transferred into the delivery guard.
  136    200   
    pub(super) permit: Permit,
  137    201   
    /// Fence identity allocated for this crossing.
  138    202   
    pub(super) delivery: DeliveryId,
  139    203   
    /// Selected destination cell.
  140    204   
    pub(super) requesting_partition: PartitionId,
  141    205   
    /// Demand identity current when the crossing was prepared.
  142    206   
    pub(super) demand: DemandId,
  143    207   
}
  144    208   
  145    209   
impl DemandSchedule {
  146    210   
    /// Applies a complete snapshot and updates its cell's scheduling residence.
  147    211   
    pub(super) fn publish(&mut self, requesting_partition: PartitionId, snapshot: DemandSnapshot) {
  148    212   
        if let Some(current) = self.records.get(&requesting_partition) {
  149    213   
            if !snapshot.is_newer_than(&current.latest) {
  150    214   
                return;
  151    215   
            }
  152    216   
        } else {
  153    217   
            self.records.insert(
  154    218   
                requesting_partition,
  155    219   
                DemandRecord {
  156    220   
                    latest: snapshot.clone(),
         221  +
                    group: None,
  157    222   
                    residence: DemandResidence::Idle,
  158    223   
                },
  159    224   
            );
  160    225   
        }
  161    226   
  162    227   
        let should_remove = matches!(
  163    228   
            &self
  164    229   
                .records
  165    230   
                .get(&requesting_partition)
  166    231   
                .expect("published demand record disappeared")
  167    232   
                .residence,
  168    233   
            DemandResidence::Queued { demand, .. }
  169    234   
                if *demand != snapshot.id || !snapshot.is_active()
  170    235   
        );
  171    236   
        if should_remove {
  172    237   
            self.remove_from_order(&requesting_partition);
  173    238   
        }
  174    239   
         240  +
        if let DemandState::Active {
         241  +
            eligibility_group, ..
         242  +
        } = &snapshot.state
         243  +
        {
         244  +
            self.records
         245  +
                .get_mut(&requesting_partition)
         246  +
                .expect("published demand record disappeared")
         247  +
                .group = Some(eligibility_group.clone());
         248  +
        }
  175    249   
        self.records
  176    250   
            .get_mut(&requesting_partition)
  177    251   
            .expect("published demand record disappeared")
  178    252   
            .latest = snapshot;
  179    253   
  180    254   
        let record = self
  181    255   
            .records
  182    256   
            .get(&requesting_partition)
  183    257   
            .expect("published demand record disappeared");
  184    258   
        if record.latest.is_active() && matches!(&record.residence, DemandResidence::Idle) {
  185    259   
            self.enqueue(requesting_partition);
  186    260   
        }
  187    261   
        self.assert_consistent();
  188    262   
    }
  189    263   
  190         -
    /// Appends an idle active demand to the origin-wide order.
         264  +
    /// Appends an idle active demand to both scheduling orders.
  191    265   
    fn enqueue(&mut self, requesting_partition: PartitionId) {
  192         -
        let links = self.order.push_back(requesting_partition);
  193         -
        if let Some(previous) = links.previous {
         266  +
        let group = self
         267  +
            .group_for(&requesting_partition)
         268  +
            .expect("enqueued demand was inactive");
         269  +
        let origin = self.origin_order.push_back(requesting_partition);
         270  +
        let group_links = self
         271  +
            .group_orders
         272  +
            .entry(group)
         273  +
            .or_default()
         274  +
            .push_back(requesting_partition);
         275  +
        if let Some(previous) = origin.previous {
  194    276   
            self.records
  195    277   
                .get_mut(&previous)
  196         -
                .expect("order tail disappeared")
         278  +
                .expect("origin demand tail disappeared")
  197    279   
                .residence
  198         -
                .links_mut()
         280  +
                .links_mut(DemandOrderView::Origin)
         281  +
                .next = Some(requesting_partition);
         282  +
        }
         283  +
        if let Some(previous) = group_links.previous {
         284  +
            self.records
         285  +
                .get_mut(&previous)
         286  +
                .expect("group demand tail disappeared")
         287  +
                .residence
         288  +
                .links_mut(DemandOrderView::Group)
  199    289   
                .next = Some(requesting_partition);
  200    290   
        }
  201    291   
  202    292   
        let record = self
  203    293   
            .records
  204    294   
            .get_mut(&requesting_partition)
  205    295   
            .expect("queued demand record disappeared");
  206    296   
        debug_assert!(matches!(record.residence, DemandResidence::Idle));
  207    297   
        let demand = record.latest.id;
  208         -
        record.residence = DemandResidence::Queued { demand, links };
         298  +
        record.residence = DemandResidence::Queued {
         299  +
            demand,
         300  +
            links: DemandLinks {
         301  +
                origin,
         302  +
                group: group_links,
         303  +
            },
         304  +
        };
  209    305   
    }
  210    306   
  211         -
    /// Removes an ordered demand and leaves its retained record idle.
         307  +
    /// Removes an ordered demand from both views and leaves its record idle.
  212    308   
    fn remove_from_order(&mut self, requesting_partition: &PartitionId) {
         309  +
        let group = self
         310  +
            .group_for(requesting_partition)
         311  +
            .expect("removed demand had no eligibility group");
  213    312   
        let residence = {
  214    313   
            let record = self
  215    314   
                .records
  216    315   
                .get_mut(requesting_partition)
  217    316   
                .expect("removed demand record disappeared");
  218    317   
            std::mem::replace(&mut record.residence, DemandResidence::Idle)
  219    318   
        };
  220    319   
        let links = residence
  221    320   
            .into_links()
  222         -
            .expect("removed demand had no order links");
         321  +
            .expect("removed demand had no scheduling links");
  223    322   
  224         -
        if let Some(previous) = links.previous {
         323  +
        if let Some(previous) = links.origin.previous {
  225    324   
            self.records
  226    325   
                .get_mut(&previous)
  227         -
                .expect("previous demand disappeared")
         326  +
                .expect("previous origin demand disappeared")
  228    327   
                .residence
  229         -
                .links_mut()
  230         -
                .next = links.next;
         328  +
                .links_mut(DemandOrderView::Origin)
         329  +
                .next = links.origin.next;
  231    330   
        }
  232         -
        if let Some(next) = links.next {
         331  +
        if let Some(next) = links.origin.next {
  233    332   
            self.records
  234    333   
                .get_mut(&next)
  235         -
                .expect("next demand disappeared")
         334  +
                .expect("next origin demand disappeared")
  236    335   
                .residence
  237         -
                .links_mut()
  238         -
                .previous = links.previous;
         336  +
                .links_mut(DemandOrderView::Origin)
         337  +
                .previous = links.origin.previous;
  239    338   
        }
         339  +
        self.origin_order
         340  +
            .remove(*requesting_partition, links.origin);
  240    341   
  241         -
        self.order.remove(*requesting_partition, links);
         342  +
        if let Some(previous) = links.group.previous {
         343  +
            self.records
         344  +
                .get_mut(&previous)
         345  +
                .expect("previous group demand disappeared")
         346  +
                .residence
         347  +
                .links_mut(DemandOrderView::Group)
         348  +
                .next = links.group.next;
         349  +
        }
         350  +
        if let Some(next) = links.group.next {
         351  +
            self.records
         352  +
                .get_mut(&next)
         353  +
                .expect("next group demand disappeared")
         354  +
                .residence
         355  +
                .links_mut(DemandOrderView::Group)
         356  +
                .previous = links.group.previous;
         357  +
        }
         358  +
        let order = self
         359  +
            .group_orders
         360  +
            .get_mut(&group)
         361  +
            .expect("ordered demand lost its eligibility-group order");
         362  +
        order.remove(*requesting_partition, links.group);
         363  +
        if order.len() == 0 {
         364  +
            self.group_orders.remove(&group);
         365  +
        }
  242    366   
    }
  243    367   
  244    368   
    /// Returns whether the head can begin a new one-to-one delivery.
  245    369   
    pub(super) fn head_is_queued(&self) -> bool {
  246         -
        let Some(head) = self.order.head() else {
         370  +
        let Some(head) = self.origin_order.head() else {
  247    371   
            return false;
  248    372   
        };
  249    373   
        matches!(
  250    374   
            &self
  251    375   
                .records
  252    376   
                .get(&head)
  253    377   
                .expect("order head disappeared")
  254    378   
                .residence,
  255    379   
            DemandResidence::Queued { .. }
  256    380   
        )
  257    381   
    }
  258    382   
  259    383   
    /// Returns the complete origin-order head when it may begin reuse.
  260    384   
    pub(super) fn queued_head(&self) -> Option<QueuedDemand> {
  261         -
        let head = self.order.head()?;
         385  +
        let head = self.origin_order.head()?;
  262    386   
        let record = self.records.get(&head).expect("order head disappeared");
  263    387   
        let DemandResidence::Queued { demand, .. } = &record.residence else {
  264    388   
            return None;
  265    389   
        };
  266    390   
        let DemandState::Active {
  267    391   
            head: requirement,
  268    392   
            eligibility_group,
  269    393   
        } = &record.latest.state
  270    394   
        else {
  271    395   
            unreachable!("queued demand became inactive");
  272    396   
        };
  273    397   
        Some(QueuedDemand {
  274    398   
            requesting_partition: head,
  275    399   
            demand: *demand,
  276    400   
            requirement: *requirement,
  277    401   
            eligibility_group: eligibility_group.clone(),
  278    402   
        })
  279    403   
    }
  280    404   
         405  +
    /// Returns one eligibility-group head when it may receive H2 visibility.
         406  +
    pub(super) fn queued_group_head(&self, group: &EligibilityGroup) -> Option<QueuedDemand> {
         407  +
        let head = self.group_orders.get(group)?.head()?;
         408  +
        let record = self
         409  +
            .records
         410  +
            .get(&head)
         411  +
            .expect("group demand head disappeared");
         412  +
        let DemandResidence::Queued { demand, .. } = &record.residence else {
         413  +
            return None;
         414  +
        };
         415  +
        let DemandState::Active {
         416  +
            head: requirement,
         417  +
            eligibility_group,
         418  +
        } = &record.latest.state
         419  +
        else {
         420  +
            unreachable!("queued group demand became inactive");
         421  +
        };
         422  +
        debug_assert_eq!(eligibility_group, group);
         423  +
        Some(QueuedDemand {
         424  +
            requesting_partition: head,
         425  +
            demand: *demand,
         426  +
            requirement: *requirement,
         427  +
            eligibility_group: eligibility_group.clone(),
         428  +
        })
         429  +
    }
         430  +
         431  +
    /// Returns the latest eligibility group retained for one cell.
         432  +
    pub(super) fn group_for(&self, requesting_partition: &PartitionId) -> Option<EligibilityGroup> {
         433  +
        self.records
         434  +
            .get(requesting_partition)
         435  +
            .and_then(|record| record.group.clone())
         436  +
    }
         437  +
  281    438   
    /// Returns whether `requesting_partition` still has this demand queued for a new action.
  282    439   
    pub(super) fn is_current_queued(
  283    440   
        &self,
  284    441   
        requesting_partition: &PartitionId,
  285    442   
        demand: DemandId,
  286    443   
    ) -> bool {
  287    444   
        self.records
  288    445   
            .get(requesting_partition)
  289    446   
            .is_some_and(|record| {
  290    447   
                record.latest.id == demand
  291    448   
                    && record.latest.is_active()
  292    449   
                    && matches!(
  293    450   
                        record.residence,
  294    451   
                        DemandResidence::Queued {
  295    452   
                            demand: current,
  296    453   
                            ..
  297    454   
                        } if current == demand
  298    455   
                    )
  299    456   
            })
  300    457   
    }
  301    458   
  302    459   
    /// Fences one reuse operation's demand while it remains the origin-order head.
  303    460   
    pub(super) fn reserve_reuse_demand(
  304    461   
        &mut self,
  305    462   
        requesting_partition: &PartitionId,
  306    463   
        demand: DemandId,
  307    464   
        delivery: DeliveryId,
  308    465   
    ) -> Option<ScheduledDemand> {
  309    466   
        if !self.is_current_queued(requesting_partition, demand) {
  310    467   
            return None;
  311    468   
        }
  312         -
        let head = self.order.head()?;
         469  +
        let head = self.origin_order.head()?;
  313    470   
        if head != *requesting_partition {
  314    471   
            return None;
  315    472   
        }
  316         -
        self.reserve_head(delivery)
         473  +
        self.reserve_origin_head(delivery)
  317    474   
    }
  318    475   
  319         -
    /// Changes the queued head into a delivery fence at the same order position.
  320         -
    pub(super) fn reserve_head(&mut self, delivery: DeliveryId) -> Option<ScheduledDemand> {
  321         -
        let head = self.order.head()?;
  322         -
        let requesting_partition = head;
         476  +
    /// Fences one eligibility-group head for an H2 publication.
         477  +
    pub(super) fn reserve_group_head(
         478  +
        &mut self,
         479  +
        group: &EligibilityGroup,
         480  +
        requesting_partition: &PartitionId,
         481  +
        demand: DemandId,
         482  +
        delivery: DeliveryId,
         483  +
    ) -> Option<ScheduledDemand> {
         484  +
        if !self.is_current_queued(requesting_partition, demand)
         485  +
            || self.group_orders.get(group)?.head() != Some(*requesting_partition)
         486  +
        {
         487  +
            return None;
         488  +
        }
         489  +
        self.reserve(*requesting_partition, delivery, DeliveryView::Group)
         490  +
    }
         491  +
         492  +
    /// Changes the origin head into a delivery fence at the same positions.
         493  +
    pub(super) fn reserve_origin_head(&mut self, delivery: DeliveryId) -> Option<ScheduledDemand> {
         494  +
        let head = self.origin_order.head()?;
         495  +
        self.reserve(head, delivery, DeliveryView::Origin)
         496  +
    }
         497  +
         498  +
    /// Changes one queued demand into a delivery fence.
         499  +
    fn reserve(
         500  +
        &mut self,
         501  +
        requesting_partition: PartitionId,
         502  +
        delivery: DeliveryId,
         503  +
        view: DeliveryView,
         504  +
    ) -> Option<ScheduledDemand> {
  323    505   
        let record = self
  324    506   
            .records
  325    507   
            .get_mut(&requesting_partition)
  326    508   
            .expect("order head disappeared");
  327    509   
        let residence = std::mem::replace(&mut record.residence, DemandResidence::Idle);
  328    510   
        match residence {
  329    511   
            DemandResidence::Queued { demand, links } => {
  330    512   
                debug_assert_eq!(record.latest.id, demand);
  331    513   
                debug_assert!(record.latest.is_active());
  332    514   
                record.residence = DemandResidence::Delivering {
  333    515   
                    demand,
         516  +
                    version: record.latest.version,
  334    517   
                    delivery,
         518  +
                    view,
  335    519   
                    links,
  336    520   
                };
  337    521   
                self.assert_consistent();
  338    522   
                Some(ScheduledDemand {
  339    523   
                    requesting_partition,
  340    524   
                    demand,
  341    525   
                })
  342    526   
            }
  343    527   
            residence => {
  344    528   
                record.residence = residence;
@@ -369,553 +577,1104 @@
  389    573   
        delivery: DeliveryId,
  390    574   
        requesting_partition: &PartitionId,
  391    575   
        result: DeliveryAckResult,
  392    576   
    ) {
  393    577   
        let Some(record) = self.records.get(requesting_partition) else {
  394    578   
            return;
  395    579   
        };
  396    580   
        let delivered_demand = match &record.residence {
  397    581   
            DemandResidence::Delivering {
  398    582   
                demand,
         583  +
                version,
  399    584   
                delivery: current,
  400    585   
                ..
  401         -
            } if *current == delivery => *demand,
         586  +
            } if *current == delivery => (*demand, *version),
  402    587   
            _ => return,
  403    588   
        };
         589  +
        let (delivered_demand, delivered_version) = delivered_demand;
  404    590   
  405    591   
        match result {
  406    592   
            DeliveryAckResult::RetrySameResidence => {
  407    593   
                let record = self
  408    594   
                    .records
  409    595   
                    .get(requesting_partition)
  410    596   
                    .expect("delivery demand record disappeared");
  411    597   
                if record.latest.id == delivered_demand && record.latest.is_active() {
  412    598   
                    let record = self
  413    599   
                        .records
  414    600   
                        .get_mut(requesting_partition)
  415    601   
                        .expect("delivery demand record disappeared");
  416    602   
                    let residence = std::mem::replace(&mut record.residence, DemandResidence::Idle);
  417    603   
                    let DemandResidence::Delivering {
  418    604   
                        demand,
         605  +
                        version: _,
  419    606   
                        delivery: current,
         607  +
                        view: _,
  420    608   
                        links,
  421    609   
                    } = residence
  422    610   
                    else {
  423    611   
                        unreachable!("delivery fence disappeared");
  424    612   
                    };
  425    613   
                    debug_assert_eq!(current, delivery);
  426    614   
                    record.residence = DemandResidence::Queued { demand, links };
  427    615   
                    self.assert_consistent();
  428    616   
                    return;
  429    617   
                }
  430    618   
  431    619   
                self.remove_from_order(requesting_partition);
  432    620   
                if self
  433    621   
                    .records
  434    622   
                    .get(requesting_partition)
  435    623   
                    .expect("delivery demand record disappeared")
  436    624   
                    .latest
  437    625   
                    .is_active()
  438    626   
                {
  439    627   
                    self.enqueue(*requesting_partition);
  440    628   
                }
  441    629   
            }
  442    630   
            DeliveryAckResult::Accepted { successor }
  443    631   
            | DeliveryAckResult::Rejected { successor } => {
  444    632   
                self.remove_from_order(requesting_partition);
  445    633   
  446    634   
                let install_successor = successor.as_ref().is_some_and(|successor| {
  447    635   
                    successor.id > delivered_demand
  448    636   
                        && successor.is_newer_than(
  449    637   
                            &self
  450    638   
                                .records
  451    639   
                                .get(requesting_partition)
  452    640   
                                .expect("delivery demand record disappeared")
  453    641   
                                .latest,
  454    642   
                        )
  455    643   
                });
  456    644   
                if install_successor {
  457    645   
                    self.records
  458    646   
                        .get_mut(requesting_partition)
  459    647   
                        .expect("delivery demand record disappeared")
  460    648   
                        .latest = successor.expect("validated successor disappeared");
  461         -
                } else if self
  462         -
                    .records
  463         -
                    .get(requesting_partition)
  464         -
                    .expect("delivery demand record disappeared")
  465         -
                    .latest
  466         -
                    .id
  467         -
                    == delivered_demand
  468         -
                {
         649  +
                } else {
         650  +
                    let retirement =
         651  +
                        DemandSnapshot::inactive(delivered_demand, delivered_version.next());
  469    652   
                    let record = self
  470    653   
                        .records
  471    654   
                        .get_mut(requesting_partition)
  472    655   
                        .expect("delivery demand record disappeared");
  473         -
                    record.latest =
  474         -
                        DemandSnapshot::inactive(delivered_demand, record.latest.version.next());
         656  +
                    if retirement.is_newer_than(&record.latest) {
         657  +
                        record.latest = retirement;
         658  +
                    }
  475    659   
                }
  476    660   
  477    661   
                if self
  478    662   
                    .records
  479    663   
                    .get(requesting_partition)
  480    664   
                    .expect("delivery demand record disappeared")
  481    665   
                    .latest
  482    666   
                    .is_active()
  483    667   
                {
  484    668   
                    self.enqueue(*requesting_partition);
  485    669   
                }
  486    670   
            }
  487    671   
        }
  488    672   
        self.assert_consistent();
  489    673   
    }
  490    674   
  491    675   
    /// Returns the latest complete snapshot retained for `requesting_partition`.
  492    676   
    #[cfg(test)]
  493    677   
    pub(super) fn latest_for_test(
  494    678   
        &self,
  495    679   
        requesting_partition: &PartitionId,
  496    680   
    ) -> Option<&DemandSnapshot> {
  497    681   
        self.records
  498    682   
            .get(requesting_partition)
  499    683   
            .map(|record| &record.latest)
  500    684   
    }
  501    685   
  502    686   
    #[cfg(test)]
  503    687   
    pub(super) fn len(&self) -> usize {
  504         -
        self.order.len()
         688  +
        self.origin_order.len()
  505    689   
    }
  506    690   
  507    691   
    #[cfg(test)]
  508    692   
    pub(super) fn queued_len(&self) -> usize {
  509    693   
        self.records
  510    694   
            .values()
  511    695   
            .filter(|record| matches!(&record.residence, DemandResidence::Queued { .. }))
  512    696   
            .count()
  513    697   
    }
  514    698   
  515    699   
    #[cfg(test)]
  516    700   
    pub(super) fn delivering_len(&self) -> usize {
  517    701   
        self.records
  518    702   
            .values()
  519    703   
            .filter(|record| matches!(&record.residence, DemandResidence::Delivering { .. }))
  520    704   
            .count()
  521    705   
    }
  522    706   
  523         -
    /// Checks record residence, FIFO links, length, and head-fence relationships.
         707  +
    /// Checks residence, link, length, group, and fence relationships.
  524    708   
    fn assert_consistent(&self) {
  525         -
        #[cfg(debug_assertions)]
  526         -
        self.assert_consistent_debug();
         709  +
        #[cfg(any(debug_assertions, test))]
         710  +
        {
         711  +
            if std::thread::panicking() {
         712  +
                return;
         713  +
            }
         714  +
            self.assert_consistent_debug();
         715  +
        }
  527    716   
    }
  528    717   
  529         -
    #[cfg(debug_assertions)]
         718  +
    #[cfg(any(debug_assertions, test))]
  530    719   
    fn assert_consistent_debug(&self) {
  531    720   
        let ordered_records = self
  532    721   
            .records
  533    722   
            .values()
  534         -
            .filter(|record| record.residence.links().is_some())
         723  +
            .filter(|record| record.residence.links(DemandOrderView::Origin).is_some())
         724  +
            .count();
         725  +
        let origin_deliveries = self
         726  +
            .records
         727  +
            .values()
         728  +
            .filter(|record| {
         729  +
                matches!(
         730  +
                    record.residence,
         731  +
                    DemandResidence::Delivering {
         732  +
                        view: DeliveryView::Origin,
         733  +
                        ..
         734  +
                    }
         735  +
                )
         736  +
            })
  535    737   
            .count();
  536         -
        let head = self.order.head();
  537         -
        let mut delivering = 0;
  538         -
        self.order.assert_consistent(
         738  +
        assert!(
         739  +
            origin_deliveries <= 1,
         740  +
            "more than one origin delivery fence was active"
         741  +
        );
         742  +
        self.origin_order.assert_consistent(
  539    743   
            ordered_records,
  540    744   
            self.records.len(),
  541         -
            "demand order",
         745  +
            "origin demand order",
  542    746   
            |requesting_partition| {
  543    747   
                let record = self
  544    748   
                    .records
  545    749   
                    .get(&requesting_partition)
  546         -
                    .expect("ordered demand disappeared");
         750  +
                    .expect("origin-ordered demand disappeared");
  547    751   
                match &record.residence {
  548         -
                    DemandResidence::Idle => unreachable!("ordered demand became idle"),
         752  +
                    DemandResidence::Idle => {
         753  +
                        unreachable!("origin-ordered demand became idle")
         754  +
                    }
  549    755   
                    DemandResidence::Queued { demand, .. } => {
  550    756   
                        assert!(record.latest.is_active(), "queued demand became inactive");
  551    757   
                        assert_eq!(
  552    758   
                            record.latest.id, *demand,
  553    759   
                            "queued residence did not match its latest demand"
  554    760   
                        );
  555    761   
                    }
  556         -
                    DemandResidence::Delivering { demand, .. } => {
  557         -
                        delivering += 1;
  558         -
                        assert_eq!(
  559         -
                            Some(requesting_partition),
  560         -
                            head,
  561         -
                            "delivery fence moved away from the order head"
  562         -
                        );
         762  +
                    DemandResidence::Delivering { demand, view, .. } => {
         763  +
                        if *view == DeliveryView::Origin {
         764  +
                            assert_eq!(
         765  +
                                Some(requesting_partition),
         766  +
                                self.origin_order.head(),
         767  +
                                "origin delivery fence moved away from its head"
         768  +
                            );
         769  +
                        }
  563    770   
                        assert!(
  564    771   
                            record.latest.id >= *demand,
  565    772   
                            "delivery fence named a future demand"
  566    773   
                        );
  567    774   
                    }
  568    775   
                }
  569    776   
                *record
  570    777   
                    .residence
  571         -
                    .links()
  572         -
                    .expect("ordered demand lost its links")
         778  +
                    .links(DemandOrderView::Origin)
         779  +
                    .expect("origin-ordered demand lost its links")
  573    780   
            },
  574    781   
        );
  575         -
        assert!(delivering <= 1, "more than one delivery fence was active");
         782  +
         783  +
        let group_ordered_records = self
         784  +
            .records
         785  +
            .values()
         786  +
            .filter(|record| record.residence.links(DemandOrderView::Group).is_some())
         787  +
            .count();
         788  +
        for record in self
         789  +
            .records
         790  +
            .values()
         791  +
            .filter(|record| record.residence.links(DemandOrderView::Group).is_some())
         792  +
        {
         793  +
            let group = record
         794  +
                .group
         795  +
                .as_ref()
         796  +
                .expect("group-ordered demand lost its eligibility group");
         797  +
            assert!(
         798  +
                self.group_orders.contains_key(group),
         799  +
                "ordered demand lost its eligibility-group order"
         800  +
            );
         801  +
        }
         802  +
        assert_eq!(
         803  +
            group_ordered_records,
         804  +
            self.group_orders
         805  +
                .values()
         806  +
                .map(IntrusiveOrder::len)
         807  +
                .sum::<usize>(),
         808  +
            "eligibility-group orders did not contain every ordered demand"
         809  +
        );
         810  +
         811  +
        for (group, order) in &self.group_orders {
         812  +
            let group_deliveries = self
         813  +
                .records
         814  +
                .values()
         815  +
                .filter(|record| {
         816  +
                    record.group.as_ref() == Some(group)
         817  +
                        && matches!(
         818  +
                            record.residence,
         819  +
                            DemandResidence::Delivering {
         820  +
                                view: DeliveryView::Group,
         821  +
                                ..
         822  +
                            }
         823  +
                        )
         824  +
                })
         825  +
                .count();
         826  +
            assert!(
         827  +
                group_deliveries <= 1,
         828  +
                "more than one group delivery fence was active"
         829  +
            );
         830  +
            let expected = self
         831  +
                .records
         832  +
                .values()
         833  +
                .filter(|record| {
         834  +
                    record.residence.links(DemandOrderView::Group).is_some()
         835  +
                        && record.group.as_ref() == Some(group)
         836  +
                })
         837  +
                .count();
         838  +
            order.assert_consistent(
         839  +
                expected,
         840  +
                self.records.len(),
         841  +
                "eligibility-group demand order",
         842  +
                |requesting_partition| {
         843  +
                    let record = self
         844  +
                        .records
         845  +
                        .get(&requesting_partition)
         846  +
                        .expect("group-ordered demand disappeared");
         847  +
                    assert_eq!(
         848  +
                        record.group.as_ref(),
         849  +
                        Some(group),
         850  +
                        "demand occupied the wrong eligibility-group order"
         851  +
                    );
         852  +
                    if matches!(
         853  +
                        record.residence,
         854  +
                        DemandResidence::Delivering {
         855  +
                            view: DeliveryView::Group,
         856  +
                            ..
         857  +
                        }
         858  +
                    ) {
         859  +
                        assert_eq!(
         860  +
                            Some(requesting_partition),
         861  +
                            order.head(),
         862  +
                            "group publication fence moved away from its head"
         863  +
                        );
         864  +
                    }
         865  +
                    *record
         866  +
                        .residence
         867  +
                        .links(DemandOrderView::Group)
         868  +
                        .expect("group-ordered demand lost its links")
         869  +
                },
         870  +
            );
         871  +
        }
         872  +
    }
         873  +
}
         874  +
         875  +
#[cfg(all(test, not(smithy_http_client_loom)))]
         876  +
mod tests {
         877  +
    use super::*;
         878  +
         879  +
    fn partition(index: usize) -> PartitionId {
         880  +
        PartitionId::from_index(index)
         881  +
    }
         882  +
         883  +
    fn active(id: u64, group: EligibilityGroup) -> DemandSnapshot {
         884  +
        DemandSnapshot::active(
         885  +
            DemandId::from_u64(id),
         886  +
            super::super::SnapshotVersion::INITIAL,
         887  +
            ProtocolRequirement::H1Compatible,
         888  +
            group,
         889  +
        )
         890  +
    }
         891  +
         892  +
    #[test]
         893  +
    fn origin_and_group_orders_repair_independently() {
         894  +
        let pool = EligibilityGroup::Pool;
         895  +
        let isolated = EligibilityGroup::Partition(partition(2));
         896  +
        let mut schedule = DemandSchedule::default();
         897  +
        schedule.publish(partition(1), active(1, pool.clone()));
         898  +
        schedule.publish(partition(2), active(2, isolated.clone()));
         899  +
        schedule.publish(partition(3), active(3, pool.clone()));
         900  +
         901  +
        assert_eq!(
         902  +
            Some(partition(1)),
         903  +
            schedule.queued_head().map(|head| head.requesting_partition)
         904  +
        );
         905  +
        assert_eq!(
         906  +
            Some(partition(1)),
         907  +
            schedule
         908  +
                .queued_group_head(&pool)
         909  +
                .map(|head| head.requesting_partition)
         910  +
        );
         911  +
        assert_eq!(
         912  +
            Some(partition(2)),
         913  +
            schedule
         914  +
                .queued_group_head(&isolated)
         915  +
                .map(|head| head.requesting_partition)
         916  +
        );
         917  +
         918  +
        let delivery = DeliveryId(7);
         919  +
        schedule
         920  +
            .reserve_group_head(&pool, &partition(1), DemandId::from_u64(1), delivery)
         921  +
            .expect("pool group head should reserve");
         922  +
        assert!(schedule.queued_group_head(&pool).is_none());
         923  +
        assert_eq!(
         924  +
            Some(partition(2)),
         925  +
            schedule
         926  +
                .queued_group_head(&isolated)
         927  +
                .map(|head| head.requesting_partition)
         928  +
        );
         929  +
        assert!(schedule.queued_head().is_none());
         930  +
         931  +
        schedule.finish_delivery(
         932  +
            delivery,
         933  +
            &partition(1),
         934  +
            DeliveryAckResult::Accepted { successor: None },
         935  +
        );
         936  +
        assert_eq!(
         937  +
            Some(partition(2)),
         938  +
            schedule.queued_head().map(|head| head.requesting_partition)
         939  +
        );
         940  +
        assert_eq!(
         941  +
            Some(partition(3)),
         942  +
            schedule
         943  +
                .queued_group_head(&pool)
         944  +
                .map(|head| head.requesting_partition)
         945  +
        );
         946  +
    }
         947  +
         948  +
    #[test]
         949  +
    fn retry_preserves_both_order_positions() {
         950  +
        let group = EligibilityGroup::Pool;
         951  +
        let mut schedule = DemandSchedule::default();
         952  +
        schedule.publish(partition(1), active(1, group.clone()));
         953  +
        schedule.publish(partition(2), active(2, group.clone()));
         954  +
         955  +
        let delivery = DeliveryId(9);
         956  +
        schedule
         957  +
            .reserve_group_head(&group, &partition(1), DemandId::from_u64(1), delivery)
         958  +
            .expect("group head should reserve");
         959  +
        schedule.finish_delivery(
         960  +
            delivery,
         961  +
            &partition(1),
         962  +
            DeliveryAckResult::RetrySameResidence,
         963  +
        );
         964  +
         965  +
        assert_eq!(
         966  +
            Some(partition(1)),
         967  +
            schedule.queued_head().map(|head| head.requesting_partition)
         968  +
        );
         969  +
        assert_eq!(
         970  +
            Some(partition(1)),
         971  +
            schedule
         972  +
                .queued_group_head(&group)
         973  +
                .map(|head| head.requesting_partition)
         974  +
        );
         975  +
    }
         976  +
         977  +
    #[test]
         978  +
    fn inactive_snapshot_during_group_fence_retires_without_losing_group() {
         979  +
        let group = EligibilityGroup::Pool;
         980  +
        let mut schedule = DemandSchedule::default();
         981  +
        schedule.publish(partition(1), active(1, group.clone()));
         982  +
        let delivery = DeliveryId(11);
         983  +
        schedule
         984  +
            .reserve_group_head(&group, &partition(1), DemandId::from_u64(1), delivery)
         985  +
            .expect("group head should reserve");
         986  +
        schedule.publish(
         987  +
            partition(1),
         988  +
            DemandSnapshot::inactive(
         989  +
                DemandId::from_u64(1),
         990  +
                super::super::SnapshotVersion::INITIAL.next(),
         991  +
            ),
         992  +
        );
         993  +
         994  +
        schedule.finish_delivery(
         995  +
            delivery,
         996  +
            &partition(1),
         997  +
            DeliveryAckResult::RetrySameResidence,
         998  +
        );
         999  +
        assert_eq!(0, schedule.len());
        1000  +
        assert!(schedule.queued_group_head(&group).is_none());
        1001  +
    }
        1002  +
        1003  +
    #[test]
        1004  +
    fn accepted_fence_does_not_retire_a_newer_active_publication() {
        1005  +
        let group = EligibilityGroup::Pool;
        1006  +
        let partition = partition(1);
        1007  +
        let demand = DemandId::from_u64(1);
        1008  +
        let initial = super::super::SnapshotVersion::INITIAL;
        1009  +
        let mut schedule = DemandSchedule::default();
        1010  +
        schedule.publish(
        1011  +
            partition,
        1012  +
            DemandSnapshot::active(
        1013  +
                demand,
        1014  +
                initial,
        1015  +
                ProtocolRequirement::H2Required,
        1016  +
                group.clone(),
        1017  +
            ),
        1018  +
        );
        1019  +
        let delivery = DeliveryId(12);
        1020  +
        schedule
        1021  +
            .reserve_group_head(&group, &partition, demand, delivery)
        1022  +
            .expect("group head should reserve");
        1023  +
        let republished = DemandSnapshot::active(
        1024  +
            demand,
        1025  +
            initial.next().next(),
        1026  +
            ProtocolRequirement::H2Required,
        1027  +
            group.clone(),
        1028  +
        );
        1029  +
        schedule.publish(partition, republished.clone());
        1030  +
        1031  +
        schedule.finish_delivery(
        1032  +
            delivery,
        1033  +
            &partition,
        1034  +
            DeliveryAckResult::Accepted { successor: None },
        1035  +
        );
        1036  +
        1037  +
        assert_eq!(Some(&republished), schedule.latest_for_test(&partition));
        1038  +
        assert_eq!(
        1039  +
            Some(partition),
        1040  +
            schedule.queued_head().map(|head| head.requesting_partition)
        1041  +
        );
        1042  +
        assert_eq!(
        1043  +
            Some(partition),
        1044  +
            schedule
        1045  +
                .queued_group_head(&group)
        1046  +
                .map(|head| head.requesting_partition)
        1047  +
        );
        1048  +
    }
        1049  +
        1050  +
    #[test]
        1051  +
    fn group_reservation_rejects_a_stale_demand_identity() {
        1052  +
        let group = EligibilityGroup::Pool;
        1053  +
        let partition = partition(1);
        1054  +
        let mut schedule = DemandSchedule::default();
        1055  +
        schedule.publish(partition, active(1, group.clone()));
        1056  +
        schedule.publish(partition, active(2, group.clone()));
        1057  +
        1058  +
        assert!(schedule
        1059  +
            .reserve_group_head(&group, &partition, DemandId::from_u64(1), DeliveryId(13),)
        1060  +
            .is_none());
        1061  +
        assert_eq!(
        1062  +
            Some(DemandId::from_u64(2)),
        1063  +
            schedule
        1064  +
                .queued_group_head(&group)
        1065  +
                .map(|queued| queued.demand)
        1066  +
        );
        1067  +
    }
        1068  +
        1069  +
    #[test]
        1070  +
    fn stale_acknowledgement_does_not_close_a_newer_delivery_fence() {
        1071  +
        let group = EligibilityGroup::Pool;
        1072  +
        let partition = partition(1);
        1073  +
        let demand = DemandId::from_u64(1);
        1074  +
        let mut schedule = DemandSchedule::default();
        1075  +
        schedule.publish(partition, active(1, group.clone()));
        1076  +
        schedule
        1077  +
            .reserve_group_head(&group, &partition, demand, DeliveryId(14))
        1078  +
            .expect("group head should reserve");
        1079  +
        1080  +
        schedule.finish_delivery(
        1081  +
            DeliveryId(15),
        1082  +
            &partition,
        1083  +
            DeliveryAckResult::Accepted { successor: None },
        1084  +
        );
        1085  +
        1086  +
        assert!(schedule.delivery_is_current(DeliveryId(14), &partition, demand));
        1087  +
        assert!(!schedule.delivery_is_current(DeliveryId(15), &partition, demand));
        1088  +
        assert_eq!(
        1089  +
            Some(&active(1, group)),
        1090  +
            schedule.latest_for_test(&partition)
        1091  +
        );
        1092  +
    }
        1093  +
        1094  +
    #[test]
        1095  +
    #[should_panic(expected = "ordered demand lost its eligibility-group order")]
        1096  +
    fn consistency_check_rejects_an_orphaned_group_link() {
        1097  +
        let group = EligibilityGroup::Pool;
        1098  +
        let mut schedule = DemandSchedule::default();
        1099  +
        schedule.publish(partition(1), active(1, group.clone()));
        1100  +
        1101  +
        schedule.group_orders.remove(&group);
        1102  +
        schedule.assert_consistent();
  576   1103   
    }
  577   1104   
}