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