moqtap_proxy/shape/mod.rs
1//! Egress shaping — the configuration a caller writes, and the
2//! pure primitives the scheduler is built from.
3//!
4//! A [`ShapeProfile`] describes what one session's *media* egress is
5//! allowed to do: named token buckets, class rules that aim a [`Matcher`]
6//! at a bucket, one bounded-queue policy, and a [`Discipline`] that
7//! arbitrates between classes competing for the same bucket. Control
8//! streams are never shaped — pacing SUBSCRIBE and ANNOUNCE behind a video
9//! bucket would stall MoQT's normal steady state and make an idle control
10//! stream look like a dead session.
11//!
12//! # Why this module is `pub`
13//!
14//! Unlike the engine internals (`egress`, `exec`, `release_timer`), a
15//! caller *constructs* these types, so they are public and every
16//! item below carries a rustdoc comment.
17//!
18//! # The two constructor shapes, and why they differ
19//!
20//! [`ShapeProfile`] has **private fields and a fallible constructor**. A
21//! mis-typed bucket name would otherwise be a silently inert class —
22//! configuration that looks applied and does nothing, which is exactly the
23//! failure this module exists to make impossible. [`ShapeProfile::try_new`]
24//! rejects it, so an invalid profile cannot reach a session and there is no
25//! runtime *your config was rejected* path to miss.
26//!
27//! The four config structs it is built from — [`BucketConfig`],
28//! [`ClassRule`], [`Matcher`] and [`QueueConfig`] — are the opposite:
29//! all-public fields and `#[non_exhaustive]` **with** a [`Default`],
30//! exactly as [`EgressConfig`](crate::action::EgressConfig) is. The pairing
31//! is load-bearing rather than stylistic: `#[non_exhaustive]` on its own
32//! makes a struct unconstructible outside this crate, because
33//! struct-expression *and* functional-update syntax are both illegal there
34//! — the entire public configuration surface would be unreachable from an
35//! integration-test crate and from every caller's code.
36//!
37//! Note precisely what the `Default` buys, because it is one step less than
38//! it looks: `..Default::default()` is **also** `E0639` outside this crate,
39//! so an outside caller writes `let mut m = Matcher::default();` followed by
40//! per-field assignment — which is what `tests/actions_shaping.rs` does at
41//! every construction site. What the `Default` provides is a *value* to
42//! start from, not a syntax. Inside this crate both forms compile, which is
43//! why the unit tests below use the shorter one; a sentence claiming the
44//! functional-update form works for an outside caller was measured false
45//! (11 × `E0639` out of tree, on all four structs).
46//!
47//! # State of the module
48//!
49//! The types, [`Matcher::matches`], [`ShapeProfile::try_new`]'s validation
50//! and the pure token bucket ([`charge`]) landed first, with their own unit
51//! tests, so the scheduler that consumes them lands against arithmetic that
52//! is already gated.
53//!
54//! A configured [`ShapeProfile`] arms framing on its own
55//! (`ProxySessionConfig::shape`), [`ShapeStats`] is recorded and readable
56//! through
57//! [`ProxySession::shape_stats`](crate::session::ProxySession::shape_stats),
58//! **admission** runs — per-unit classification through [`Matcher`], the
59//! per-stream queue depth and all three [`Overflow`] policies — and so does
60//! **release**: every shaped unit is queued rather than written inline, its
61//! class's token bucket is debited at `PendingQueue::pop_next_due`, the
62//! configured [`Discipline`] arbitrates between classes sharing a bucket,
63//! and [`Expiry`] decides what becomes of a unit that outlives `max_hold`.
64//!
65//! The same figures are kept a second time for a whole proxy. [`ProxyStats`]
66//! — [`LegStats`], [`SessionStats`] and the class rows — is read through
67//! [`ProxyControl::stats`](crate::control::ProxyControl::stats) and covers
68//! every session the proxy has accepted, including the ones that have already
69//! ended, so it is cumulative where `ProxySession::shape_stats` is one
70//! session's own. It is charged by the same writers, forwarded from inside
71//! each one, so no figure can reach a session's rows and miss the proxy's.
72//! The one shape difference is worth knowing before reading a cell: a
73//! session's totals carry a direction only, while a proxy's carry a leg *and*
74//! a direction, because a proxy holds two connections and a byte crosses
75//! both. [`LegStats`] states which cell each measurement lands in.
76//!
77//! Two things are deliberately outside that: **control streams**, which
78//! install no scheduler at all, and **teardown**, which drains ignoring
79//! release times so a bucket can never gate a mirrored reset.
80//!
81//! **An object too large for the framer to buffer is outside it as well, and
82//! says so.** Such an object has no `ObjectMeta`, so no rule can name it, so no
83//! bucket charges it and it is granted unconditionally — one object can
84//! therefore cross a class's rate whole. Measured: a 4 MiB object crossed in
85//! 800 ms against a class whose bucket was configured at zero bytes per second.
86//! The bytes are accounted on [`ShapeStats::unshapeable`] and the session
87//! reports `Impairment{ShapeUnpacedObject}`, once per stream, naming the class
88//! the stream's other units are charged to — because *my 500 kbps cap was
89//! breached by one large segment* is otherwise a hole in the accounting with
90//! nothing to attribute it to.
91//!
92//! **Datagrams are policed rather than paced**, which is a different
93//! operation and not a lesser one. `forward_datagrams` classifies each
94//! datagram through [`Matcher::matches_datagram`], asks its class's bucket
95//! for the bytes, and **discards** what the bucket refuses instead of
96//! queuing it. Nothing on that path delays anything, and nothing should: a
97//! FIFO would impose a delivery order the protocol does not have, and a
98//! datagram has neither a successor written against it nor a stream whose
99//! object IDs would move behind a hole — which is exactly what makes
100//! dropping the arriving unit sound here and unsound for a queued stream
101//! unit.
102//!
103//! What follows from that shape, and is worth knowing before reading a
104//! figure: [`QueueConfig`] is **not consulted** for a datagram. Neither
105//! depth binds it and no [`Overflow`] policy decides it, because it is never
106//! queued — the bucket is the whole of the decision. A profile that shapes
107//! subgroup streams and polices datagrams reads its queue policy for the
108//! first and not for the second.
109//!
110//! **This is settled rather than pending, and the sharp edge is worth stating
111//! outright:** the bucket can answer *not now, but at this instant* — the
112//! same answer that defers a stream unit — and on the datagram path that
113//! answer is discarded like every other refusal. A datagram over a live rate
114//! is dropped where it arrived, not held until the instant its own bucket
115//! named.
116//!
117//! **If what is wanted is smoothing, reach for `quinn-netem`.** It delays,
118//! jitters and reorders at the socket, under the whole connection, which is
119//! the scope a link-level queue has: a bottleneck queues by link, not by
120//! track, and a router does not know which track a datagram belongs to.
121//! Class-aware policing is a real box — an operator rate-limiter drops over
122//! rate — while class-aware smoothing is a scheduler *inside* a router, which
123//! is not a condition a player is ever placed in. So netem's not being
124//! class-aware is the right scope for it rather than a gap in it, and the
125//! division is: **a rate on one track is a class over a bucket, and belongs
126//! here; a congested path is netem.**
127//!
128//! The framed sites keep their `Delay` and `Hold` because a stream **has** a
129//! delivery order — holding object N and then N+1 preserves a guarantee the
130//! protocol makes, where holding two datagrams would manufacture one.
131//!
132//! There is no seam a datagram never reaches: every unmatchable rule reports
133//! from a unit that arrived, through `Scheduler::classify` or its datagram
134//! sibling.
135//!
136//! # What is still owed
137//!
138//! [`BucketConfig::ceil_bps`] is accepted and never borrowed against: a
139//! profile setting it above `rate_bps` measures a flat `rate_bps`.
140//!
141//! Nothing else. What a **hook** does is not reported here and cannot be:
142//! every figure on this page is gated on a configured profile while a hook
143//! needs none, so a unit a hook delayed and an object it truncated are
144//! counted on
145//! [`Counters::units_delayed`](crate::instrument::Counters::units_delayed)
146//! and
147//! [`Counters::objects_truncated`](crate::instrument::Counters::objects_truncated)
148//! instead. No figure here is a `Duration` either; [`ClassStats`] says why
149//! this page carries no duration at all.
150//!
151//! Separately, and not a figure this page owes: of the five figures a
152//! [`DirectionStats`] carries, only `objects_seen` and `bytes_shaped` are
153//! measured at both crossings, so the three event figures read zero in a
154//! **departure** cell of [`ProxyStats::per_leg`] — a zero with no producer
155//! rather than one waiting on work. [`LegStats`] says which cell is which.
156
157mod bucket;
158mod matcher;
159mod scheduler;
160mod stats;
161
162pub use bucket::{charge, BucketConfig, BucketState, Grant};
163pub use matcher::{MatchKind, Matcher, MatcherField, RangeSet};
164pub use stats::{ClassStats, DirectionStats, LegStats, ProxyStats, SessionStats, ShapeStats};
165
166pub(crate) use scheduler::{Acquire, Admission, Class, QueueDepth, Scheduler};
167pub(crate) use stats::{ProxyRecorder, ShapeRecorder};
168
169use std::collections::HashSet;
170use std::time::Duration;
171
172use crate::types::ProxySide;
173
174/// A complete egress shaping configuration for one session.
175///
176/// Constructed only through [`ShapeProfile::try_new`], which validates the
177/// combination — an invalid profile cannot reach a session, so there is no
178/// runtime *your config was rejected* path to miss.
179///
180/// Not `#[non_exhaustive]`: the fields are private, so the attribute would
181/// add nothing a caller could observe.
182///
183/// # Reading one from a file
184///
185/// Under the non-default `serde` feature this type serializes and
186/// deserializes, and the two directions are **not symmetric**. Serializing is
187/// a derive over the private fields, which is safe because writing a profile
188/// out cannot make an invalid one. Deserializing goes
189/// `#[serde(try_from = "ShapeProfileSpec")]`, through a public-field mirror
190/// whose `TryFrom` calls [`ShapeProfile::try_new`].
191///
192/// The detour is the whole point. A derived `Deserialize` would reach these
193/// private fields directly and bypass every one of the seven validations below
194/// — including [`ShapeError::UnknownBucket`], where a file naming a bucket
195/// that does not exist would parse, arm, report shaping and shape nothing.
196/// Routing through the mirror means there is no deserialization path that
197/// skips the constructor, and no consumer has to remember to convert.
198#[derive(Debug, Clone, PartialEq)]
199#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
200#[cfg_attr(feature = "serde", serde(try_from = "ShapeProfileSpec"))]
201pub struct ShapeProfile {
202 buckets: Vec<BucketConfig>,
203 classes: Vec<ClassRule>,
204 queue: QueueConfig,
205 discipline: Discipline,
206}
207
208impl ShapeProfile {
209 /// Validate a profile and build it, or say exactly what is wrong.
210 ///
211 /// The seven rejections, in the order they are checked:
212 ///
213 /// 1. [`ShapeError::EmptyQueue`] — a queue that can hold nothing.
214 /// 2. [`ShapeError::NoClasses`] — a profile with no class rules at all,
215 /// which is a shaping profile that shapes nothing.
216 /// 3. [`ShapeError::DuplicateClassName`] — class names index the
217 /// statistics, so duplicates make them unattributable.
218 /// 4. [`ShapeError::UnknownBucket`] — a class naming a bucket that is
219 /// not in `buckets`; the silently-inert class this constructor
220 /// exists to prevent.
221 /// 5. [`ShapeError::EgressSideInMatcher`] — a matcher keyed on an
222 /// egress side, which no hook site ever sees.
223 /// 6. [`ShapeError::ZeroWeight`] — a zero weight under
224 /// [`Discipline::WeightedRoundRobin`], which is a class that can
225 /// never be scheduled.
226 /// 7. [`ShapeError::InertMatcher`] — a matcher key naming an *empty set
227 /// of values*, which is a class that can never claim a unit.
228 ///
229 /// The second is checked **outside** the loop below, and that is the
230 /// point of it: every other class rule is checked *inside* a
231 /// `for class in &classes`, and a loop over nothing runs no checks at
232 /// all. A profile with no classes was therefore the one shape that could
233 /// pass every rule here by not being subject to any of them.
234 ///
235 /// Duplicate *bucket* names are not an error: two identical entries
236 /// resolve to the same bucket and the first one wins, which is what a
237 /// caller who wrote the name twice meant. Only class names index
238 /// anything.
239 ///
240 /// # What this constructor cannot see, and why the line is there
241 ///
242 /// Every check above is a property of the **configuration alone**. What
243 /// it deliberately does not attempt is anything that depends on the
244 /// draft or on the traffic, and there are two such faults; both are
245 /// reported during the run instead, because a rejection here has to be
246 /// right for *every* session the profile could be used in.
247 ///
248 /// A key the wire does not carry on this draft is
249 /// `Impairment{ShapeRuleUnmatchable}` — `try_new` has no draft. And a
250 /// [`BucketConfig::burst_bytes`] smaller than the objects a class
251 /// actually sees is
252 /// `Impairment{ShapeBurstBelowUnit}` — `try_new` has the burst but not
253 /// the object sizes, and the sizes are what decide. The second is worth
254 /// the attention because its silent form is so plausible: a burst below
255 /// one object makes every unit leave at its `max_hold` clamp, at a
256 /// throughput with no relation to the rate that was configured, and
257 /// before the report existed the only signal was the one an ordinary
258 /// rate-limited class produces.
259 pub fn try_new(
260 buckets: Vec<BucketConfig>,
261 classes: Vec<ClassRule>,
262 queue: QueueConfig,
263 discipline: Discipline,
264 ) -> Result<Self, ShapeError> {
265 if queue.depth_bytes == 0 || queue.depth_objects == 0 {
266 return Err(ShapeError::EmptyQueue);
267 }
268 // Before the loop, because the loop is what every other rule lives
269 // in and an empty list is the one input it cannot judge.
270 if classes.is_empty() {
271 return Err(ShapeError::NoClasses);
272 }
273
274 let mut seen: HashSet<&str> = HashSet::with_capacity(classes.len());
275 for class in &classes {
276 if !seen.insert(class.name.as_str()) {
277 return Err(ShapeError::DuplicateClassName { name: class.name.clone() });
278 }
279 if !buckets.iter().any(|b| b.name == class.bucket) {
280 return Err(ShapeError::UnknownBucket {
281 class: class.name.clone(),
282 bucket: class.bucket.clone(),
283 });
284 }
285 if matches!(
286 class.matcher.side,
287 Some(ProxySide::ProxyToClient) | Some(ProxySide::ProxyToRelay)
288 ) {
289 return Err(ShapeError::EgressSideInMatcher { class: class.name.clone() });
290 }
291 if discipline == Discipline::WeightedRoundRobin && class.weight == 0 {
292 return Err(ShapeError::ZeroWeight { class: class.name.clone() });
293 }
294 if let Some(key) = class.matcher.inert_key() {
295 return Err(ShapeError::InertMatcher { class: class.name.clone(), key });
296 }
297 }
298
299 Ok(Self { buckets, classes, queue, discipline })
300 }
301
302 /// The configured buckets, in the order they were given.
303 pub fn buckets(&self) -> &[BucketConfig] {
304 &self.buckets
305 }
306
307 /// The configured class rules, in the order they were given — which is
308 /// also the order the statistics snapshot reports them in, and the
309 /// order [`Discipline::Fifo`] tie-breaks on.
310 pub fn classes(&self) -> &[ClassRule] {
311 &self.classes
312 }
313
314 /// The per-stream queue policy.
315 pub fn queue(&self) -> &QueueConfig {
316 &self.queue
317 }
318
319 /// How classes competing for one bucket are arbitrated.
320 pub fn discipline(&self) -> Discipline {
321 self.discipline
322 }
323}
324
325/// Why a [`ShapeProfile`] could not be built.
326///
327/// `#[non_exhaustive]` and no `Default`: nobody constructs an error, and a
328/// later release adding a further reason must not be a breaking change.
329#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)]
330#[non_exhaustive]
331pub enum ShapeError {
332 /// Two classes share a name. Class names index the stats, so they must
333 /// be unique or the stats are unattributable.
334 #[error("duplicate class name: {name}")]
335 DuplicateClassName {
336 /// The name that appeared more than once.
337 name: String,
338 },
339 /// A class names a bucket that is not in `buckets`.
340 #[error("class {class} names unknown bucket {bucket}")]
341 UnknownBucket {
342 /// The class holding the dangling reference.
343 class: String,
344 /// The bucket name that matches no [`BucketConfig`].
345 bucket: String,
346 },
347 /// A matcher's `side` is an egress label. Hook sites only ever see
348 /// `ClientToProxy` / `RelayToProxy`, so such a rule matches nothing —
349 /// rejected here rather than left to look like a working rule that
350 /// never fires.
351 #[error("class {class} matches on an egress side, which no hook site sees")]
352 EgressSideInMatcher {
353 /// The class whose matcher named an egress side.
354 class: String,
355 },
356 /// [`Discipline::WeightedRoundRobin`] with a zero weight: a class that
357 /// can never be scheduled.
358 #[error("class {class} has weight 0 under WeightedRoundRobin")]
359 ZeroWeight {
360 /// The class whose weight is zero.
361 class: String,
362 },
363 /// [`QueueConfig::depth_objects`] or [`QueueConfig::depth_bytes`] is
364 /// zero, so the queue could admit nothing.
365 #[error("queue depth is zero in bytes or in objects")]
366 EmptyQueue,
367 /// The profile declares no class rules, so it is a shaping profile that
368 /// shapes nothing.
369 ///
370 /// Every unit such a profile sees falls to `Class::Default`, which is
371 /// unpaced: no bucket claims it, no discipline arbitrates it and the
372 /// queue releases it as soon as it reaches the head. A session
373 /// configured with one therefore frames every object — because a profile
374 /// arms framing on its own — pays for the classification and the queue,
375 /// reports itself as shaping, and delivers at line rate. Nothing in
376 /// [`crate::shape::ShapeStats`] distinguishes it from a profile whose
377 /// classes never matched.
378 ///
379 /// # It would also reach further than the session that carried it
380 ///
381 /// A proxy sizes its class rows once, from the first shaped session it
382 /// accepts, because a class is an index into the class list of the
383 /// scheduler that produced it and rows that could be resized underneath
384 /// a running session would relabel every figure in them. A classless
385 /// profile arriving first would have installed **no rows at all**, and
386 /// every classed session accepted afterwards — for the life of the proxy
387 /// — would have found rows it did not match and charged the default one:
388 /// every number right, every label gone, unrecoverable without a
389 /// restart. That is guarded a second time where the sizing happens, but
390 /// the guard is a repair at the far end of the pipe; this is the profile
391 /// never existing.
392 ///
393 /// A caller who wants the queue policy and no pacing writes one class
394 /// claiming everything, over a bucket with no rate — an explicitly
395 /// unshaped class, which has a name and a row of its own and says in the
396 /// configuration what a missing class list only implied.
397 #[error(
398 "a shaping profile with no classes shapes nothing: every unit falls to the unpaced \
399 default class. Declare a class over a rate-less bucket if that is what was meant"
400 )]
401 NoClasses,
402 /// A [`Matcher`] key names an **empty set of values**, so the class can
403 /// never claim a unit — on any draft, from any traffic.
404 ///
405 /// The three shapes this catches:
406 /// a [`RangeSet`] built from an inverted range (`RangeSet::new` drops
407 /// `start > end`, leaving an empty set whose `contains` is always
408 /// `false`), an empty [`Matcher::priority`] range such as `200..=100`,
409 /// and [`Matcher::every_nth`] with `n == 0`.
410 ///
411 /// Distinct from
412 /// [`ImpairmentKind::ShapeRuleUnmatchable`](crate::event::ImpairmentKind::ShapeRuleUnmatchable),
413 /// and the distinction is *when the answer exists*: a rule keyed on a
414 /// field this draft does not carry can only be judged against a running
415 /// session, so it is reported; an empty value set is a property of the
416 /// configuration by itself, so it is rejected before a session starts.
417 /// Rejecting is strictly the better answer where it is available —
418 /// there is no run to read the report from.
419 #[error("class {class} keys on {key}, which names no value at all")]
420 InertMatcher {
421 /// The class whose matcher can never claim anything.
422 class: String,
423 /// The key that names the empty set, spelled as the
424 /// [`Matcher`] field is: one of `track_alias`, `group_id`,
425 /// `subgroup_id`, `object_id`, `priority`, `every_nth`.
426 key: &'static str,
427 },
428}
429
430/// A matcher plus what to do with what it matches.
431///
432/// `#[non_exhaustive]` *with* a [`Default`] — see the module doc for why
433/// the pairing is required rather than stylistic. The default is an
434/// unnamed class that claims every unit and names no bucket, so it is
435/// always edited before use; note that a defaulted `weight` of zero is
436/// rejected under [`Discipline::WeightedRoundRobin`].
437///
438/// In the written form `name` and `bucket` are required and the other three
439/// default, because the two that are required are the two whose defaults are
440/// wrong rather than merely empty: an unnamed class collides with the next
441/// unnamed class as [`ShapeError::DuplicateClassName`], and a class naming no
442/// bucket at all is [`ShapeError::UnknownBucket`] against the empty string.
443/// Both are refusals about a key the author never wrote.
444#[derive(Debug, Clone, PartialEq, Eq, Default)]
445#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
446#[cfg_attr(feature = "serde", serde(deny_unknown_fields))]
447#[non_exhaustive]
448pub struct ClassRule {
449 /// The class name. Unique across the profile, and the label under
450 /// which this class's statistics are reported.
451 pub name: String,
452 /// Which units this class claims.
453 #[cfg_attr(feature = "serde", serde(default))]
454 pub matcher: Matcher,
455 /// The [`BucketConfig::name`] this class charges against. Several
456 /// classes may share one bucket, which is what makes [`Discipline`]
457 /// mean anything.
458 pub bucket: String,
459 /// [`Discipline::StrictPriority`] orders classes by this; higher wins.
460 ///
461 /// Distinct from MoQT's `publisher_priority`, which
462 /// [`Matcher::priority`] keys on: this one is the scheduler's, and it
463 /// is always present.
464 #[cfg_attr(feature = "serde", serde(default))]
465 pub priority: u8,
466 /// [`Discipline::WeightedRoundRobin`] shares a bucket by this. Zero is
467 /// rejected under that discipline and ignored under the other two.
468 #[cfg_attr(feature = "serde", serde(default))]
469 pub weight: u16,
470}
471
472/// The per-stream queue policy: how deep, how long, and what happens at
473/// each limit.
474///
475/// `#[non_exhaustive]` *with* a [`Default`] — see the module doc.
476///
477/// The `Default` is **hand-written, not derived**: a derived one would
478/// give `depth_bytes == 0` and `depth_objects == 0`, which
479/// [`ShapeProfile::try_new`] rejects as [`ShapeError::EmptyQueue`] — so
480/// `QueueConfig::default()` would be a value that cannot be used, and the
481/// `..Default::default()` idiom this type is built for would fail on every
482/// profile that did not restate both depths.
483///
484/// That hand-written `Default` is also what the written form defaults every
485/// key to, so a serialized profile may omit `queue` entirely or name only the one
486/// knob it cares about. It is the one config in this module where the derived
487/// default would be the unusable value and the written default is therefore
488/// worth having.
489#[derive(Debug, Clone, PartialEq, Eq)]
490#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
491#[cfg_attr(feature = "serde", serde(default, deny_unknown_fields))]
492#[non_exhaustive]
493pub struct QueueConfig {
494 /// Bytes one stream's queue may hold. Defaults to 1 MiB, matching
495 /// [`EgressConfig::max_pending_bytes`](crate::action::EgressConfig::max_pending_bytes).
496 pub depth_bytes: usize,
497 /// Objects one stream's queue may hold. Defaults to 256 — a limit the
498 /// byte depth does not imply, since a queue of small objects reaches
499 /// neither.
500 pub depth_objects: usize,
501 /// Deadline for a queued object, per class. `None` inherits
502 /// [`EgressConfig::max_hold`](crate::action::EgressConfig::max_hold),
503 /// which is 30 s.
504 ///
505 /// Pin this explicitly in any fixture that relies on a class *not*
506 /// delivering: under the default [`Expiry::Deliver`] a starved class
507 /// still delivers at `max_hold`, so an unnamed 30 s is a margin the
508 /// test inherited rather than chose.
509 pub max_hold: Option<Duration>,
510 /// What happens when the queue is full.
511 pub overflow: Overflow,
512 /// What happens when a queued object outlives `max_hold`.
513 pub on_expiry: Expiry,
514}
515
516impl Default for QueueConfig {
517 fn default() -> Self {
518 Self {
519 depth_bytes: 1024 * 1024,
520 depth_objects: 256,
521 max_hold: None,
522 overflow: Overflow::default(),
523 on_expiry: Expiry::default(),
524 }
525 }
526}
527
528/// What happens when a per-stream queue is full.
529///
530/// `#[non_exhaustive]`, with `Block` as the [`Default`]: the only
531/// non-destructive answer is the one a caller gets without asking.
532///
533/// There is deliberately no `DropHead`. Dropping an *already queued* unit
534/// happens after the framer's positional cursor has moved past it, so the
535/// elide fix-up can no longer be armed — and on drafts 14-21 object IDs are
536/// delta-encoded, so the result is not a gap but every successor decoding
537/// with a wrong absolute ID. [`Overflow::DropTail`] is sound for exactly
538/// the reason `DropHead` is not: it discards the *arriving* unit, at
539/// admission time, where the fix-up is still legal.
540///
541/// Written as `"block"`, `"drop-tail"` or `{"reset-stream": {"code": 1}}`.
542#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
543#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
544#[cfg_attr(feature = "serde", serde(rename_all = "kebab-case", deny_unknown_fields))]
545#[non_exhaustive]
546pub enum Overflow {
547 /// Stop reading the source. Non-destructive. **Default.**
548 ///
549 /// Per stream, and it does not reliably stall the *peer*: the
550 /// transport's own receive window absorbs megabytes first.
551 #[default]
552 Block,
553 /// Discard the *arriving* unit. Renumbers via the framer's own elide
554 /// fix-up, so absolute object IDs stay correct on drafts 14-21.
555 ///
556 /// When an elide guard refuses the fix-up the unit is admitted anyway —
557 /// the queue overshoots by one — and the refusal is reported. A shaper
558 /// may not corrupt a stream to honour a depth limit.
559 DropTail,
560 /// Abandon the destination stream.
561 ResetStream {
562 /// The application error code to reset with.
563 code: u64,
564 },
565}
566
567/// What happens when a queued object outlives `max_hold`.
568///
569/// `#[non_exhaustive]`, with `Deliver` as the [`Default`] — which is
570/// exactly today's behaviour, so nothing changes for a session that does
571/// not ask for shaping.
572///
573/// There is deliberately **no** `Drop` variant. An expiry is decided at
574/// release time, long after the framer's positional cursor has advanced
575/// past the object, so the elide fix-up cannot be armed; on drafts 14-21
576/// that corrupts every successor's absolute ID. A variant that is
577/// constructible and always refused is worse than an absent one.
578///
579/// The cost of the pick, stated so nobody rediscovers it: *the relay gave up on
580/// stale media* is expressible only at *stream* granularity, and
581/// `objects_expired` is therefore zero by default, with a producer only on the
582/// [`Expiry::ResetStream`] arm.
583///
584/// Written as `"deliver"` or `{"reset-stream": {"code": 1}}`.
585#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
586#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
587#[cfg_attr(feature = "serde", serde(rename_all = "kebab-case", deny_unknown_fields))]
588#[non_exhaustive]
589pub enum Expiry {
590 /// Clamp the deadline and deliver anyway. Today's behaviour, and the
591 /// reason a starved class still delivers at `max_hold`. **Default.**
592 #[default]
593 Deliver,
594 /// Give up on the stream: drain what is due, then reset.
595 ResetStream {
596 /// The application error code to reset with.
597 code: u64,
598 },
599}
600
601/// How classes competing for the same bucket are arbitrated.
602///
603/// `#[non_exhaustive]`, with `Fifo` as the [`Default`]. The discipline only
604/// decides *between* classes; within one destination stream the queue stays
605/// a single FIFO whatever this says, because object IDs are delta-encoded
606/// on the wire and reordering them corrupts the chain.
607///
608/// Written as `"fifo"`, `"strict-priority"` or `*weighted-round-robin*`.
609#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
610#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
611#[cfg_attr(feature = "serde", serde(rename_all = "kebab-case"))]
612#[non_exhaustive]
613pub enum Discipline {
614 /// Whoever asked first. **Default.**
615 #[default]
616 Fifo,
617 /// Highest [`ClassRule::priority`] first, and only then the rest.
618 StrictPriority,
619 /// Share by [`ClassRule::weight`]. A zero weight is rejected by
620 /// [`ShapeProfile::try_new`] under this discipline.
621 WeightedRoundRobin,
622}
623
624/// Identifies one forwarded stream within a session.
625///
626/// A **session-local monotonic id**, minted from one counter per session at
627/// the moment the session accepts the stream, plus the side it arrived on.
628/// Unique for the session's lifetime and never reused.
629///
630/// Deliberately **not** a media key: `subgroup_id` is absent on eight
631/// drafts and `track_alias` on every fetch stream, so a media-keyed
632/// serialize would silently miss.
633///
634/// Deliberately **not** the transport stream id either. On the WebTransport
635/// arm `SendStream::stream_id()` is the constant `0`
636/// (`moqtap-client/src/transport/mod.rs:133-137`), and the proxy really
637/// does accept WebTransport clients — so a transport-keyed `StreamKey`
638/// collapses every WT stream onto one entry per side. A serialize would
639/// then attach a stream to an arbitrary sibling, or to itself, which is a
640/// self-deadlock that degrades to a `max_hold` stall, and every per-stream
641/// report becomes unattributable.
642///
643/// The transport id is still worth reporting where it means something, so
644/// it stays a separate field on the events that carry both.
645/// `Hash` is hand-written because [`ProxySide`] does not derive it and
646/// lives in a module this type may not edit. It hashes the side's
647/// discriminant, so it agrees with the derived [`PartialEq`] exactly:
648/// equal keys hash equal, which is the whole obligation.
649#[derive(Debug, Clone, Copy, PartialEq, Eq)]
650pub struct StreamKey {
651 /// The direction the stream was accepted on.
652 pub side: ProxySide,
653 /// Session-local monotonic id. Never reused within a session, and not
654 /// comparable across sessions.
655 pub id: u64,
656}
657
658impl std::hash::Hash for StreamKey {
659 fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
660 std::mem::discriminant(&self.side).hash(state);
661 self.id.hash(state);
662 }
663}
664
665/// The written form of a [`ShapeProfile`].
666///
667/// # Why the mirror exists
668///
669/// [`ShapeProfile`]'s four fields are private, and that is the whole of its
670/// safety: [`ShapeProfile::try_new`] is the only way to build one, and it
671/// refuses seven configurations that would otherwise arm and do nothing —
672/// chief among them a class naming a bucket that is not in `buckets`, which
673/// reports shaping and shapes nothing, and a profile with no classes at all,
674/// which matches nothing there is to match.
675///
676/// A derived `Deserialize` on `ShapeProfile` would reach those private fields
677/// directly and bypass all seven. So `ShapeProfile` deserializes
678/// `#[serde(try_from = "ShapeProfileSpec")]` instead: serde builds *this*
679/// type, whose fields are public and whose only job is to be built, and the
680/// [`TryFrom`] impl runs `try_new`. Every deserialization path therefore
681/// validates, and no consumer has to remember to convert — which matters
682/// because the consumer is usually a field on some caller's own
683/// configuration type, reached by someone who never names this one at all.
684///
685/// The reverse direction, [`From<&ShapeProfile>`](ShapeProfileSpec), exists so
686/// a caller can take a profile apart, edit it and rebuild it through the same
687/// validation.
688///
689/// `#[non_exhaustive]` **with** a [`Default`], as the shaping configs are: the
690/// attribute makes struct-expression and functional-update syntax illegal
691/// outside this crate, so the `Default` is what leaves a construction path
692/// open — `let mut spec = ShapeProfileSpec::default();` and then per-field
693/// assignment.
694#[cfg(feature = "serde")]
695#[derive(Debug, Clone, serde::Serialize, serde::Deserialize, PartialEq, Default)]
696#[serde(deny_unknown_fields)]
697#[non_exhaustive]
698pub struct ShapeProfileSpec {
699 /// The token buckets, by name. A class names one of these.
700 pub buckets: Vec<BucketConfig>,
701 /// The class rules, in the order they are matched — which is also the
702 /// order the statistics report them in.
703 pub classes: Vec<ClassRule>,
704 /// The per-stream queue policy. Defaults to
705 /// [`QueueConfig::default`], which is a usable policy rather than a
706 /// placeholder.
707 #[serde(default)]
708 pub queue: QueueConfig,
709 /// How classes sharing a bucket are arbitrated. Defaults to
710 /// [`Discipline::Fifo`].
711 #[serde(default)]
712 pub discipline: Discipline,
713}
714
715#[cfg(feature = "serde")]
716impl TryFrom<ShapeProfileSpec> for ShapeProfile {
717 type Error = ShapeError;
718
719 fn try_from(spec: ShapeProfileSpec) -> Result<Self, Self::Error> {
720 ShapeProfile::try_new(spec.buckets, spec.classes, spec.queue, spec.discipline)
721 }
722}
723
724#[cfg(feature = "serde")]
725impl From<&ShapeProfile> for ShapeProfileSpec {
726 fn from(profile: &ShapeProfile) -> Self {
727 Self {
728 buckets: profile.buckets().to_vec(),
729 classes: profile.classes().to_vec(),
730 queue: profile.queue().clone(),
731 discipline: profile.discipline(),
732 }
733 }
734}
735#[cfg(test)]
736mod tests {
737 use super::*;
738
739 fn bucket(name: &str) -> BucketConfig {
740 BucketConfig {
741 name: name.to_string(),
742 rate_bps: Some(64_000),
743 burst_bytes: 16_000,
744 ..BucketConfig::default()
745 }
746 }
747
748 fn class(name: &str) -> ClassRule {
749 ClassRule {
750 name: name.to_string(),
751 bucket: "b".to_string(),
752 weight: 1,
753 ..ClassRule::default()
754 }
755 }
756
757 /// A profile that `try_new` accepts, so every rejection below is
758 /// attributable to the one field its row edits.
759 fn valid() -> (Vec<BucketConfig>, Vec<ClassRule>, QueueConfig, Discipline) {
760 (
761 vec![bucket("b")],
762 vec![class("audio"), class("video")],
763 QueueConfig::default(),
764 Discipline::Fifo,
765 )
766 }
767
768 #[test]
769 fn a_valid_profile_round_trips_through_its_accessors() {
770 let (buckets, classes, queue, discipline) = valid();
771 let p = ShapeProfile::try_new(buckets, classes, queue.clone(), discipline)
772 .expect("the control profile must be valid or every rejection below is unattributable");
773 assert_eq!(p.buckets().len(), 1);
774 assert_eq!(
775 p.classes().iter().map(|c| c.name.as_str()).collect::<Vec<_>>(),
776 ["audio", "video"]
777 );
778 assert_eq!(p.queue(), &queue);
779 assert_eq!(p.discipline(), Discipline::Fifo);
780 }
781
782 /// The named single case for the egress-side rejection.
783 ///
784 /// Kept beside the table below rather than folded into it: this is the
785 /// one rejection that is about the *proxy's* topology rather than about
786 /// the profile's internal consistency, and it is the rejection a
787 /// caller is most likely to trip.
788 #[test]
789 fn try_new_rejects_an_egress_side() {
790 for side in [ProxySide::ProxyToClient, ProxySide::ProxyToRelay] {
791 let (buckets, mut classes, queue, discipline) = valid();
792 classes[1].matcher.side = Some(side);
793 assert_eq!(
794 ShapeProfile::try_new(buckets, classes, queue, discipline),
795 Err(ShapeError::EgressSideInMatcher { class: "video".to_string() }),
796 "{side:?} is an egress label and no hook site ever sees it"
797 );
798 }
799
800 // The two ingress sides are accepted, so the rejection is about the
801 // direction and not about the field being set at all.
802 for side in [ProxySide::ClientToProxy, ProxySide::RelayToProxy] {
803 let (buckets, mut classes, queue, discipline) = valid();
804 classes[1].matcher.side = Some(side);
805 assert!(ShapeProfile::try_new(buckets, classes, queue, discipline).is_ok());
806 }
807 }
808
809 /// One row per `ShapeError` variant. Seven variants, seven rows, and
810 /// the count is asserted so an eighth variant cannot be added without a
811 /// row.
812 #[test]
813 fn try_new_rejects_every_invalid_profile() {
814 type Edit =
815 fn(&mut Vec<BucketConfig>, &mut Vec<ClassRule>, &mut QueueConfig, &mut Discipline);
816
817 let rows: [(&str, Edit, ShapeError); 7] = [
818 (
819 "two classes share a name",
820 |_b, c, _q, _d| c[1].name = "audio".to_string(),
821 ShapeError::DuplicateClassName { name: "audio".to_string() },
822 ),
823 (
824 "a class names a bucket that is not configured",
825 |_b, c, _q, _d| c[1].bucket = "nope".to_string(),
826 ShapeError::UnknownBucket {
827 class: "video".to_string(),
828 bucket: "nope".to_string(),
829 },
830 ),
831 (
832 "a matcher names an egress side",
833 |_b, c, _q, _d| c[1].matcher.side = Some(ProxySide::ProxyToClient),
834 ShapeError::EgressSideInMatcher { class: "video".to_string() },
835 ),
836 (
837 "weighted round robin with a zero weight",
838 |_b, c, _q, d| {
839 *d = Discipline::WeightedRoundRobin;
840 c[1].weight = 0;
841 },
842 ShapeError::ZeroWeight { class: "video".to_string() },
843 ),
844 (
845 "a queue that can hold nothing",
846 |_b, _c, q, _d| q.depth_objects = 0,
847 ShapeError::EmptyQueue,
848 ),
849 (
850 "a matcher key that names no value at all",
851 |_b, c, _q, _d| c[1].matcher.every_nth = Some((0, 0)),
852 ShapeError::InertMatcher { class: "video".to_string(), key: "every_nth" },
853 ),
854 (
855 "a profile with no class rules at all",
856 |_b, c, _q, _d| c.clear(),
857 ShapeError::NoClasses,
858 ),
859 ];
860
861 for (label, edit, want) in rows {
862 let (mut buckets, mut classes, mut queue, mut discipline) = valid();
863 edit(&mut buckets, &mut classes, &mut queue, &mut discipline);
864 assert_eq!(
865 ShapeProfile::try_new(buckets, classes, queue, discipline),
866 Err(want),
867 "{label}"
868 );
869 }
870
871 // The byte half of `EmptyQueue`, which the row above cannot also
872 // cover without testing two fields in one assertion.
873 let (buckets, classes, mut queue, discipline) = valid();
874 queue.depth_bytes = 0;
875 assert_eq!(
876 ShapeProfile::try_new(buckets, classes, queue, discipline),
877 Err(ShapeError::EmptyQueue)
878 );
879
880 // A zero weight is only an error under WeightedRoundRobin, so the
881 // fourth row is about the pairing and not about the weight.
882 let (buckets, mut classes, queue, discipline) = valid();
883 classes[1].weight = 0;
884 assert!(ShapeProfile::try_new(buckets, classes, queue, discipline).is_ok());
885 }
886
887 /// **`try_new` rejects every class that can never claim anything** —
888 /// one of the silent no-ops this constructor exists to make loud.
889 /// The constructor's own reason for existing is that *a mis-typed bucket
890 /// name would not be a silently inert class — configuration that looks
891 /// applied and does nothing*, and it checked five things, none of which was
892 /// the matcher's own arithmetic. All three rows below were accepted as
893 /// valid profiles and delivered zero bytes in silence; the detector
894 /// (`RangeSet::is_empty`) was already written and already public.
895 ///
896 /// Each row is paired with the **same key holding a real value**, in
897 /// the same body, so a rejection cannot be passing because `try_new`
898 /// rejects any profile that sets that key at all — which would be a far
899 /// worse defect than the one being fixed.
900 ///
901 /// *Ablation, recorded:* delete the `inert_key` check from `try_new` —
902 /// all three rejections redden here, the inert-matcher row of
903 /// `try_new_rejects_every_invalid_profile` reddens with it,
904 /// and each `left` is the accepted profile printed in full, which is the
905 /// point: the thing that ships is a `ShapeProfile` that looks entirely
906 /// ordinary and holds `track_alias: Some(RangeSet { ranges: [] })`.
907 ///
908 /// ```text
909 /// assertion `left == right` failed: an inverted track_alias range is an empty
910 /// set, so this class can never claim a unit
911 /// left: Ok(ShapeProfile { .. track_alias: Some(RangeSet { ranges: [] }) .. })
912 /// right: Err(InertMatcher { class: "video", key: "track_alias" })
913 /// ```
914 #[test]
915 fn try_new_rejects_a_class_that_can_never_claim_a_unit() {
916 // Bound through locals: a literal `9..=1` is a clippy error at the
917 // call site (`reversed_empty_ranges`), and a configuration built at
918 // runtime is exactly where these come from.
919 let (lo, hi) = (9u64, 1u64);
920 let (top, bottom) = (200u8, 100u8);
921
922 let build = |edit: &dyn Fn(&mut Matcher)| {
923 let (buckets, mut classes, queue, discipline) = valid();
924 edit(&mut classes[1].matcher);
925 ShapeProfile::try_new(buckets, classes, queue, discipline)
926 };
927 let inert =
928 |key: &'static str| Err(ShapeError::InertMatcher { class: "video".to_string(), key });
929
930 assert_eq!(
931 build(&|m| m.track_alias = Some(RangeSet::new([lo..=hi]))),
932 inert("track_alias"),
933 "an inverted track_alias range is an empty set, so this class can \
934 never claim a unit"
935 );
936 assert_eq!(
937 build(&|m| m.priority = Some(top..=bottom)),
938 inert("priority"),
939 "an empty priority range contains no value, so no header can satisfy it"
940 );
941 assert_eq!(
942 build(&|m| m.every_nth = Some((0, 0))),
943 inert("every_nth"),
944 "`n == 0` names no units, which `Matcher::matches` answers false for \
945 unconditionally"
946 );
947
948 // The positive controls: the same three keys holding real values
949 // are ordinary working rules.
950 assert!(build(&|m| m.track_alias = Some(RangeSet::new([hi..=lo]))).is_ok());
951 assert!(build(&|m| m.priority = Some(bottom..=top)).is_ok());
952 assert!(build(&|m| m.every_nth = Some((2, 0))).is_ok());
953
954 // And the empty range really is what `RangeSet` stores, so the
955 // rejection is about emptiness and not about the literal.
956 assert!(RangeSet::new([lo..=hi]).is_empty());
957 }
958
959 /// Duplicate *bucket* names are deliberately not an error, unlike
960 /// duplicate class names. Pinned so the asymmetry is a decision rather
961 /// than an oversight.
962 #[test]
963 fn duplicate_bucket_names_are_not_an_error() {
964 let (_, classes, queue, discipline) = valid();
965 assert!(ShapeProfile::try_new(vec![bucket("b"), bucket("b")], classes, queue, discipline)
966 .is_ok());
967 }
968
969 /// The defaults the `..Default::default()` idiom hands a caller are
970 /// usable as they stand — a derived `QueueConfig::default()` would be
971 /// `EmptyQueue` and every profile that did not restate both depths
972 /// would be rejected.
973 #[test]
974 fn the_default_queue_config_is_a_profile_that_builds() {
975 let (buckets, classes, _, discipline) = valid();
976 assert!(ShapeProfile::try_new(buckets, classes, QueueConfig::default(), discipline).is_ok());
977
978 let q = QueueConfig::default();
979 assert_ne!(q.depth_bytes, 0);
980 assert_ne!(q.depth_objects, 0);
981 assert_eq!(q.max_hold, None);
982 assert_eq!(q.overflow, Overflow::Block);
983 assert_eq!(q.on_expiry, Expiry::Deliver);
984 assert_eq!(Discipline::default(), Discipline::Fifo);
985 }
986
987 /// `StreamKey` is session-local and side-scoped: the same id on the two
988 /// sides is two keys, or a registry entry would be shared by the two
989 /// halves of a forwarded pair.
990 #[test]
991 fn a_stream_key_is_scoped_by_side_as_well_as_id() {
992 let a = StreamKey { side: ProxySide::ClientToProxy, id: 1 };
993 let b = StreamKey { side: ProxySide::RelayToProxy, id: 1 };
994 assert_ne!(a, b);
995 assert_eq!(a, StreamKey { side: ProxySide::ClientToProxy, id: 1 });
996
997 let mut set = HashSet::new();
998 assert!(set.insert(a));
999 assert!(set.insert(b));
1000 assert!(!set.insert(a));
1001 }
1002
1003 /// A profile with one bucket and one class, valid by construction.
1004 ///
1005 /// Built with functional-update syntax, which compiles here and would not
1006 /// outside this crate: `#[non_exhaustive]` makes both the struct literal
1007 /// and `..Default::default()` illegal there, so a caller assigns per field
1008 /// on a `::default()` binding instead.
1009 #[cfg(feature = "serde")]
1010 fn profile() -> ShapeProfile {
1011 let bucket = BucketConfig {
1012 name: "video".to_owned(),
1013 rate_bps: Some(500_000),
1014 burst_bytes: 65_536,
1015 ..Default::default()
1016 };
1017
1018 let class = ClassRule {
1019 name: "video".to_owned(),
1020 bucket: "video".to_owned(),
1021 matcher: Matcher {
1022 side: Some(crate::types::ProxySide::ClientToProxy),
1023 track_alias: Some(crate::shape::RangeSet::new([1..=3, 9..=9])),
1024 ..Default::default()
1025 },
1026 ..Default::default()
1027 };
1028
1029 ShapeProfile::try_new(
1030 vec![bucket],
1031 vec![class],
1032 QueueConfig::default(),
1033 Discipline::StrictPriority,
1034 )
1035 .expect("a profile with one class naming its own bucket is valid")
1036 }
1037
1038 /// A `ShapeProfile` serializes into the shape `ShapeProfileSpec` reads
1039 /// back, and the two derives are on different types — one on the private
1040 /// fields, one on the mirror — so nothing but this test holds their field
1041 /// names together. A rename on either side lands here as a parse failure.
1042 #[cfg(feature = "serde")]
1043 #[test]
1044 fn shape_profile_round_trips_through_json() {
1045 let before = profile();
1046 let json = serde_json::to_string(&before).expect("a profile serializes");
1047 let after: ShapeProfile = serde_json::from_str(&json).expect("and reads back");
1048 assert_eq!(before, after, "round trip through {json}");
1049 }
1050
1051 /// One profile per corner of the written form, each valid by
1052 /// construction.
1053 ///
1054 /// `profile()` above is one ordinary configuration, which is the right
1055 /// fixture for *does the round trip work at all* and reaches almost none
1056 /// of the values a file can hold. These rows are picked for the rest: both
1057 /// ends of every integer, every optional field once present and once
1058 /// absent, and every variant of every enum in the written form at least
1059 /// once — [`Discipline`] across the four rows, [`Overflow`] and
1060 /// [`Expiry`] within them, [`MatchKind`] and both ingress sides in the
1061 /// last.
1062 #[cfg(feature = "serde")]
1063 fn corners() -> Vec<(&'static str, ShapeProfile)> {
1064 let build = |buckets, classes, queue, discipline| {
1065 ShapeProfile::try_new(buckets, classes, queue, discipline)
1066 .expect("every row here must be a profile or it proves nothing about writing one")
1067 };
1068
1069 // Every field at the smallest value `try_new` accepts, which is not
1070 // the same as `Default`: a zero queue depth is `EmptyQueue`, so the
1071 // floor is one. The zero weight is legal here and only here — it is
1072 // refused under `WeightedRoundRobin`, which the third row uses.
1073 let minima = build(
1074 vec![BucketConfig { name: "b".to_owned(), burst_bytes: 0, ..Default::default() }],
1075 vec![ClassRule {
1076 name: String::new(),
1077 bucket: "b".to_owned(),
1078 priority: 0,
1079 weight: 0,
1080 matcher: Matcher::default(),
1081 }],
1082 QueueConfig {
1083 depth_bytes: 1,
1084 depth_objects: 1,
1085 max_hold: None,
1086 overflow: Overflow::Block,
1087 on_expiry: Expiry::Deliver,
1088 },
1089 Discipline::Fifo,
1090 );
1091
1092 let maxima = build(
1093 vec![BucketConfig {
1094 name: "b".to_owned(),
1095 rate_bps: Some(u64::MAX),
1096 burst_bytes: u64::MAX,
1097 ceil_bps: Some(u64::MAX),
1098 }],
1099 vec![ClassRule {
1100 name: "everything".to_owned(),
1101 bucket: "b".to_owned(),
1102 priority: u8::MAX,
1103 weight: u16::MAX,
1104 matcher: Matcher {
1105 side: Some(ProxySide::RelayToProxy),
1106 track_alias: Some(RangeSet::new([0..=u64::MAX])),
1107 group_id: Some(RangeSet::new([u64::MAX..=u64::MAX])),
1108 subgroup_id: Some(RangeSet::new([0..=0])),
1109 object_id: Some(RangeSet::new([1..=2, 5..=9])),
1110 priority: Some(0..=u8::MAX),
1111 stream_kind: Some(MatchKind::Subgroup),
1112 every_nth: Some((u64::MAX, u64::MAX)),
1113 },
1114 }],
1115 QueueConfig {
1116 depth_bytes: usize::MAX,
1117 depth_objects: usize::MAX,
1118 max_hold: Some(Duration::MAX),
1119 overflow: Overflow::ResetStream { code: u64::MAX },
1120 on_expiry: Expiry::ResetStream { code: u64::MAX },
1121 },
1122 Discipline::StrictPriority,
1123 );
1124
1125 // A deliberately stopped class: `rate_bps: Some(0)` is legal and is
1126 // not `None`, and a `max_hold` of zero is not an absent one.
1127 let stopped = build(
1128 vec![BucketConfig {
1129 name: "b".to_owned(),
1130 rate_bps: Some(0),
1131 burst_bytes: 1,
1132 ceil_bps: None,
1133 }],
1134 vec![ClassRule {
1135 name: "stopped".to_owned(),
1136 bucket: "b".to_owned(),
1137 priority: 1,
1138 weight: 1,
1139 matcher: Matcher { every_nth: Some((1, 0)), ..Default::default() },
1140 }],
1141 QueueConfig {
1142 max_hold: Some(Duration::ZERO),
1143 overflow: Overflow::DropTail,
1144 ..Default::default()
1145 },
1146 Discipline::WeightedRoundRobin,
1147 );
1148
1149 // One class per stream kind, and the ingress side the first two rows
1150 // do not name.
1151 let kinds = build(
1152 vec![
1153 BucketConfig { name: "a".to_owned(), burst_bytes: 8, ..Default::default() },
1154 BucketConfig { name: "b".to_owned(), burst_bytes: 8, ..Default::default() },
1155 ],
1156 [MatchKind::Subgroup, MatchKind::Fetch, MatchKind::Datagram]
1157 .into_iter()
1158 .enumerate()
1159 .map(|(n, kind)| ClassRule {
1160 name: format!("k{n}"),
1161 bucket: if n == 0 { "a".to_owned() } else { "b".to_owned() },
1162 matcher: Matcher {
1163 side: Some(ProxySide::ClientToProxy),
1164 stream_kind: Some(kind),
1165 ..Default::default()
1166 },
1167 ..Default::default()
1168 })
1169 .collect(),
1170 QueueConfig::default(),
1171 Discipline::Fifo,
1172 );
1173
1174 vec![
1175 ("every field at its floor", minima),
1176 ("every field at its ceiling", maxima),
1177 ("a stopped class and a zero hold", stopped),
1178 ("one class per stream kind", kinds),
1179 ]
1180 }
1181
1182 /// Every value the written form can hold survives being written and read.
1183 ///
1184 /// The two directions share no code — `Serialize` is a derive over
1185 /// `ShapeProfile`'s private fields and `Deserialize` goes through
1186 /// `ShapeProfileSpec`'s public ones — so nothing but a test holds them
1187 /// together, and `shape_profile_round_trips_through_json` above holds them
1188 /// together over one configuration. These rows are the values that one
1189 /// does not reach.
1190 #[cfg(feature = "serde")]
1191 #[test]
1192 fn every_corner_of_the_written_form_round_trips() {
1193 for (label, before) in corners() {
1194 let json = serde_json::to_string(&before).expect("a profile serializes");
1195 let after: ShapeProfile =
1196 serde_json::from_str(&json).expect("and what it wrote is readable");
1197 assert_eq!(before, after, "{label}: round trip through {json}");
1198 }
1199 }
1200
1201 /// A profile and its mirror write the same document.
1202 ///
1203 /// [`ShapeProfileSpec`]'s own `Serialize` is the half of this pair that no
1204 /// code path in this crate runs: everything here reads a profile and
1205 /// nothing writes one, so the derive is compiled and unproven. It matters
1206 /// because the mirror is what a caller edits — take a profile apart,
1207 /// change a bucket, write it back out — and a document the mirror writes
1208 /// that the profile cannot read would be a file that had round-tripped
1209 /// through this crate and come out unusable.
1210 #[cfg(feature = "serde")]
1211 #[test]
1212 fn the_mirror_writes_what_a_profile_writes() {
1213 for (label, profile) in corners() {
1214 let spec = ShapeProfileSpec::from(&profile);
1215 assert_eq!(
1216 serde_json::to_value(&spec).expect("the mirror serializes"),
1217 serde_json::to_value(&profile).expect("and so does the profile it came from"),
1218 "{label}: two writers of one format"
1219 );
1220
1221 let json = serde_json::to_string(&spec).expect("the mirror serializes");
1222 assert_eq!(
1223 serde_json::from_str::<ShapeProfile>(&json)
1224 .expect("and what the mirror wrote is a profile"),
1225 profile,
1226 "{label}: taking a profile apart and writing it back has to be lossless, or the \
1227 documented way to edit one loses the edit's neighbours"
1228 );
1229 }
1230 }
1231
1232 /// The one written form that is **not** a round trip, stated as itself.
1233 ///
1234 /// `ShapeProfileSpec::default()` is the construction path
1235 /// `#[non_exhaustive]` leaves open outside this crate, so it is a value a
1236 /// caller holds; it serializes, and reading what it wrote back as a
1237 /// `ShapeProfile` is [`ShapeError::NoClasses`]. That is the mirror working
1238 /// rather than failing — a spec is a builder and the constructor is the
1239 /// only thing that turns one into a profile — but it means *serialize then
1240 /// deserialize* is not total over this pair, and the refusal is the useful
1241 /// half to pin: it proves reading goes through `try_new` and not around it.
1242 #[cfg(feature = "serde")]
1243 #[test]
1244 fn a_default_mirror_writes_a_document_that_is_not_a_profile() {
1245 let json = serde_json::to_string(&ShapeProfileSpec::default())
1246 .expect("the mirror writes whatever it holds, valid or not");
1247
1248 let err = serde_json::from_str::<ShapeProfile>(&json)
1249 .expect_err("a profile with no classes shapes nothing and is not one");
1250 assert!(
1251 err.to_string().contains("shapes nothing"),
1252 "the refusal has to be the constructor's own, or reading is going around it: {err}"
1253 );
1254
1255 assert_eq!(
1256 serde_json::from_str::<ShapeProfileSpec>(&json).expect("the mirror reads its own back"),
1257 ShapeProfileSpec::default(),
1258 "the asymmetry is between the two types and not inside the mirror"
1259 );
1260 }
1261
1262 /// A `RangeSet` is written as a plain list of ranges and read back through
1263 /// `RangeSet::new`, which sorts and coalesces — so a file listing
1264 /// overlapping ranges in any order produces the same set as one listing
1265 /// them merged, and `contains` cannot be reading an unsorted vector.
1266 #[cfg(feature = "serde")]
1267 #[test]
1268 fn range_sets_coalesce_when_they_are_read() {
1269 let json = r#"[{"start":4,"end":6},{"start":1,"end":3}]"#;
1270 let set: crate::shape::RangeSet = serde_json::from_str(json).expect("ranges parse");
1271 assert_eq!(set.ranges(), &[1..=6], "adjacent ranges are one range after a read");
1272 }
1273}