moqtap_proxy/shape/stats.rs
1//! Per-class shaping statistics: the atomic storage and its snapshot.
2//!
3//! The split mirrors [`instrument`](crate::instrument) exactly — a
4//! [`ShapeRecorder`] of atomics that the forwarding tasks write, and a
5//! plain-value [`ShapeStats`] that a reader takes. It is a **sibling** of
6//! `Counters`, not an extension of it: `interest_none.rs` compares
7//! `Counters` whole against `Counters::default()` and `actions_*.rs`
8//! compare it by value, so a field added there would have to be threaded
9//! through every one of those assertions, and `Counters` would lose `Copy`
10//! for a `Vec` that is empty on all but shaped sessions.
11//!
12//! # Why a pre-sized `Vec`, indexed by position
13//!
14//! The class list is fixed for a session's lifetime — live reconfigure is
15//! expected to build a new recorder rather than mutate this one. So the
16//! rows are allocated once at session construction, in
17//! [`ShapeProfile::classes`] order, and every increment on
18//! the data path is one relaxed `fetch_add` at a known index. No lock, no
19//! name lookup, no hot-path allocation — the cost model `instrument.rs`
20//! already established.
21//!
22//! Snapshot order is therefore the configured order, deterministically, and
23//! a class that never saw a unit is present with a zero row rather than
24//! absent. A reader asking "what did `video` do?" gets an answer whether or
25//! not `video` did anything, which is the difference between a starved
26//! class and a mis-typed one.
27//!
28//! # What has a producer today
29//!
30//! Two session totals — `objects_seen` and `bytes_shaped` — are written at the
31//! one place a shaped session's framed unit becomes visible, and they exist
32//! here so that *an unshaped session shaped nothing* is falsifiable rather than
33//! vacuous: something has to move when the path *is* entered, or an all-zero
34//! snapshot cannot tell "not armed" from *armed and did nothing*.
35//! `bytes_shaped` additionally counts what **no rule could see** — a stream
36//! header, an oversized object's passthrough chunk, a bypassed stream's tail —
37//! because the `unshapeable` row is a term of the conservation identity and a
38//! term with nothing on the other side of the equals sign is not a term.
39//! `objects_seen` does not: it is the *classifier's* count, and a header is not
40//! an object.
41//!
42//! Admission adds the three rows a policy can move without a clock:
43//! `objects_dropped` / `bytes_dropped` (`Overflow::DropTail`),
44//! `blocked_episodes` (`Overflow::Block`) and `streams_reset_by_shaping`
45//! (`Overflow::ResetStream`). Every one of them is charged to the class the
46//! classifier resolved, so a per-class figure is an answer about a *rule*
47//! and not about a stream.
48//!
49//! Release adds the rest: `bytes_delivered` / `objects_delivered` on every
50//! granted unit, `tokens_exhausted_episodes` when a class's own bucket is
51//! dry, `starved_behind_other_class` when a *different* class's unit was in
52//! the way, `objects_expired` on the `Expiry::ResetStream` arm, and
53//! `streams_with_mixed_classes` once per stream that carried two classes.
54//!
55//! # Session totals carry a direction
56//!
57//! Every session total is stored twice, once per leg: **uplink** is the
58//! client's traffic on its way to the relay, **downlink** is the relay's on
59//! its way to the client. One recorder serves every forwarding task of a
60//! session, so without the split an author shaping both legs reads a single
61//! figure and cannot tell an uplink stall from a downlink one — a
62//! bidirectional run reports downlink starvation as though the uplink class
63//! had caused it.
64//!
65//! The flat totals stay, and they are **derived**: `bytes_shaped` is
66//! `uplink.bytes_shaped + downlink.bytes_shaped`, summed in
67//! [`ShapeRecorder::snapshot`] rather than accumulated in a third atomic.
68//! The data path therefore costs exactly what it did — one relaxed
69//! `fetch_add` into the arriving leg's row — and the aggregate cannot drift
70//! from its parts, which is not a property any pair of independently
71//! written counters has. What that leaves falsifiable is the *attribution*:
72//! charging every byte to one leg keeps the sum right and both legs wrong,
73//! which is what the tests below and `egress.rs`'s two-leg test aim at.
74//!
75//! Per-class rows are deliberately **not** split. An author who wants a
76//! class figure per leg writes two classes and keys each matcher on
77//! [`Matcher::side`](super::Matcher::side); the totals are the ones no
78//! configuration could separate, which is why they are the ones that carry
79//! the direction themselves.
80//!
81//! # The proxy-wide aggregate, and why it is not a sum over live sessions
82//!
83//! [`ProxyRecorder`] is the second recorder in this module: one per
84//! [`TransparentProxy`](crate::proxy::TransparentProxy), held by its control
85//! plane, and charged by the *same* writers that charge the session
86//! recorder — every `note_*` below forwards, so no call site in `session.rs`
87//! or `egress.rs` knows it exists and no figure can be charged to one
88//! recorder and missed by the other.
89//!
90//! Summing the sessions a control plane lists instead would be wrong three
91//! ways, and only the first is fixable. The registry holds no recorder at
92//! all, so there is nothing to sum. The registration is released by a
93//! `Drop`, so a total taken over it would go *down* when a client
94//! disconnected — a statistic that falls under normal operation cannot be
95//! alerted on. And the list is a snapshot of a proxy that keeps moving, so a
96//! sum walked across it is a consistent read of nothing. A recorder that
97//! outlives every session has none of those problems: a session whose whole
98//! future is dropped mid-flight still contributed at the instant each unit
99//! was charged.
100//!
101//! # Where a proxy-level byte is charged: the measurement point
102//!
103//! [`ProxyStats::per_leg`] is a 2×2 — two legs, two directions — and the
104//! two axes are orthogonal, which is exactly what makes the shape worth
105//! having and exactly what makes it easy to fill in wrongly. A proxy holds
106//! two connections; a byte crosses **both**, read on one leg and written on
107//! the other. So the cell is chosen by where the measurement is taken, not
108//! by a label copied off the arriving side:
109//!
110//! * `per_leg[Client].uplink` — bytes read from the client.
111//! * `per_leg[Upstream].uplink` — bytes written to the relay.
112//! * `per_leg[Upstream].downlink` — bytes read from the relay.
113//! * `per_leg[Client].downlink` — bytes written to the client.
114//!
115//! The alternative — deriving a leg from the side a counting site holds — looks
116//! equivalent and carries no information at all. Every hook site is handed
117//! `ClientToProxy` or `RelayToProxy` and nothing else
118//! ([`ShapeProfile::try_new`] refuses a rule keyed on an egress side), and over
119//! those two values leg and direction are the *same* partition: the client row
120//! would be a verbatim copy of the uplink row, the upstream row of the downlink
121//! row, and two of the four cells would be identically zero. Four numbers
122//! carrying two numbers' worth of information, with a reader who took
123//! `per_leg[Upstream]` for *what this proxy sent upstream* getting the figure
124//! for what it received from the client.
125//!
126//! Charging by measurement point makes the difference between the two
127//! uplink cells the shaper's own retention — bytes it read from the client
128//! and did not write to the relay — which is a number that can be non-zero
129//! and therefore a number that can be asserted.
130//!
131//! Only the flow of units is measured twice. The three event figures on a
132//! [`DirectionStats`] — expiries, streams a policy gave up on, streams that
133//! carried two classes — are decisions taken over traffic that *arrived*,
134//! so they are charged to the arrival cell and the departure cell reports
135//! zero for them. That is stated on [`LegStats`] rather than left for a
136//! reader to infer from a zero.
137//!
138//! [`ProxyStats::sessions`] is the flat rollup, and it sums the two
139//! **arrival** cells — the two places a byte enters this proxy, where each
140//! byte is counted exactly once. Summing all four would count every byte
141//! twice, once read and once written.
142//!
143//! **Every figure on this page has a producer.** Worth stating, because it
144//! was not always true: five fields here snapshotted as a constant zero for
145//! a long time, each documenting its own emptiness, and what settled them
146//! was not writing five producers but noticing that none of the five
147//! belonged here.
148//!
149//! Everything on this page is gated on a configured [`ShapeProfile`] — a
150//! proxy running without one reports `ProxyStats::default()` however many
151//! gigabytes it forwards. Two of the five counted what a **hook** does,
152//! which needs no profile at all, so in this type they could only ever have
153//! been a partial count that read zero for every unprofiled session that
154//! delayed or truncated a thousand objects. They are
155//! [`Counters::units_delayed`](crate::instrument::Counters::units_delayed)
156//! and
157//! [`Counters::objects_truncated`](crate::instrument::Counters::objects_truncated)
158//! now, beside
159//! [`Counters::objects_elided`](crate::instrument::Counters::objects_elided),
160//! which had already settled where a hook's decision is counted.
161//!
162//! The other three were `Duration` totals, and a sum of wall-clock time that
163//! the machine's scheduler moves as much as this code does is reportable and
164//! never assertable. The timing dimension is measured instead by
165//! [`Counters::release_errors`](crate::instrument::Counters::release_errors),
166//! which reports a distribution — p50, p95 and an exact maximum — rather
167//! than a total nobody can calibrate.
168//!
169//! With release and the unshapeable row wired, the conservation identity
170//! `Σ classes(delivered + dropped) + default + unshapeable == bytes_shaped`
171//! holds for a stream that ran to completion. Both sides come from the same
172//! measurement — `raw.len()` for a framed object, `Pending::len()` for a
173//! unit no rule saw — taken at the see-point and charged again, unchanged,
174//! at release, so it is an identity over one number and not an agreement
175//! between two.
176//!
177//! Two things break it, both named rather than papered over. A
178//! `STOP_SENDING`-driven teardown, where `propagate_stop` clears the queue
179//! rather than draining it — those bytes are neither delivered nor dropped,
180//! and `Impairment{QueuedBytesAtTeardown}` is what accounts for them. And a
181//! hook action that changes a unit's size after it was seen: `Replace`,
182//! `ReplacePayload` and `Truncate` are all charged in full on the left and
183//! by what they actually wrote on the right. Both are properties of the
184//! *run*, not of the recorder, which is why the fixture that asserts
185//! the identity takes no hook action and runs its streams to completion
186//! before it reads.
187
188use std::sync::atomic::{AtomicU64, Ordering};
189use std::sync::{Arc, OnceLock};
190
191use super::scheduler::Class;
192use super::ShapeProfile;
193use crate::types::{Leg, ProxySide};
194
195// ── which leg ──────────────────────────────────────────────────────────
196
197/// Which leg of the proxy a shaped unit is travelling on.
198///
199/// Two variants rather than [`ProxySide`]'s four. A unit is charged on the
200/// side it *arrived* on, and the two egress sides never reach a shaping
201/// decision at all — [`ShapeProfile::try_new`] rejects a rule keyed on one
202/// — but both halves of a leg answer the same question anyway, so the
203/// conversion below is total and no call site has an impossible arm to
204/// invent a value for.
205#[derive(Debug, Clone, Copy, PartialEq, Eq)]
206pub(crate) enum Direction {
207 /// Client → relay: what the client publishes and asks for.
208 Uplink,
209 /// Relay → client: what the client is subscribed to.
210 Downlink,
211}
212
213impl Direction {
214 /// This leg's row in [`ShapeRecorder`]'s session totals.
215 ///
216 /// An index rather than a `match` at each writer: the totals are an
217 /// array precisely so that charging one is the same single
218 /// `fetch_add` at a known offset the class rows already are.
219 fn index(self) -> usize {
220 match self {
221 Direction::Uplink => 0,
222 Direction::Downlink => 1,
223 }
224 }
225}
226
227impl From<ProxySide> for Direction {
228 fn from(side: ProxySide) -> Self {
229 match side {
230 ProxySide::ClientToProxy | ProxySide::ProxyToRelay => Direction::Uplink,
231 ProxySide::RelayToProxy | ProxySide::ProxyToClient => Direction::Downlink,
232 }
233 }
234}
235
236/// The connection a side's traffic is travelling over, and the way it is
237/// going — the pair [`ProxyStats::per_leg`] is indexed by.
238///
239/// A [`ProxySide`] is exactly this pair: `transport.rs` says so in prose,
240/// that `ClientToProxy` and `ProxyToClient` are the two directions of the
241/// client leg and `ProxyToRelay` and `RelayToProxy` the two of the upstream
242/// leg, and this is that sentence written as a total function. Written out
243/// arm by arm rather than composed from [`Direction::from`] and a second
244/// mapping, because the four arms are the definition and a reader checking
245/// the attribution should not have to compose two functions to see it.
246fn split(side: ProxySide) -> (Leg, Direction) {
247 match side {
248 ProxySide::ClientToProxy => (Leg::Client, Direction::Uplink),
249 ProxySide::ProxyToRelay => (Leg::Upstream, Direction::Uplink),
250 ProxySide::RelayToProxy => (Leg::Upstream, Direction::Downlink),
251 ProxySide::ProxyToClient => (Leg::Client, Direction::Downlink),
252 }
253}
254
255/// The leg a unit read on `leg` leaves this proxy by.
256///
257/// A proxy holds two connections and forwards between them, so the leg a
258/// unit is written on is always the other one. The *direction* is unchanged
259/// — traffic the client published is uplink on both legs — which is why
260/// this takes a leg rather than a side: naming the departure cell needs the
261/// leg flipped and nothing else.
262fn opposite(leg: Leg) -> Leg {
263 match leg {
264 Leg::Client => Leg::Upstream,
265 Leg::Upstream => Leg::Client,
266 }
267}
268
269/// This leg's row in [`ProxyStats::per_leg`].
270///
271/// An index rather than a `match` at each writer, for the reason
272/// [`Direction::index`] gives: charging a cell stays one relaxed
273/// `fetch_add` at a known offset.
274fn leg_index(leg: Leg) -> usize {
275 match leg {
276 Leg::Client => 0,
277 Leg::Upstream => 1,
278 }
279}
280
281// ── the snapshot types ─────────────────────────────────────────────────
282
283/// Shaping statistics for one session.
284///
285/// Zero-valued on any session with no [`ShapeProfile`], which is what makes
286/// *this session shaped nothing* a falsifiable claim rather than a promise.
287/// Reported separately from `Counters` so that crate's whole-struct `==
288/// Counters::default()` assertions keep meaning what they mean.
289///
290/// Read through
291/// [`ProxySession::shape_stats`](crate::session::ProxySession::shape_stats).
292///
293/// The session totals appear twice: flat, summed over the whole session,
294/// and again under [`Self::uplink`] and [`Self::downlink`] for one leg
295/// each. The flat figure is the sum of the two by construction, so the two
296/// forms can never disagree — pick the leg when the question is which side
297/// stalled, and the aggregate when it is whether the profile ran at all.
298///
299/// **Every field here is written.** Two were not until recently — the
300/// `Duration` totals on [`ClassStats`], which were calibration figures and
301/// are gone; that type says why there is no duration among these at all. A
302/// zero row is still reported, never omitted — a class that saw nothing is
303/// present and empty, which is the difference between a starved class and a
304/// mis-typed one.
305#[derive(Debug, Clone, Default, PartialEq, Eq)]
306pub struct ShapeStats {
307 /// One entry per configured class, in [`ShapeProfile::classes`] order.
308 pub classes: Vec<ClassStats>,
309 /// Units that matched no rule.
310 pub default_class: ClassStats,
311 /// Units with no object metadata at all: subgroup and fetch stream
312 /// headers, oversized passthrough objects, bypassed streams — every
313 /// fetch stream on drafts 15-19, which have no fetch object codec.
314 /// A **separate** row from [`Self::default_class`], and the distinction is
315 /// the point: the default row is *the rules saw this unit and none claimed
316 /// it*, this row is *no rule could have seen it*. Merging them would make a
317 /// mis-aimed matcher indistinguishable from a stream the framer cannot
318 /// address.
319 ///
320 /// These bytes are **not paced**: they charge no bucket, so a class
321 /// rate can be exceeded by exactly one oversized object.
322 pub unshapeable: ClassStats,
323 /// Hook-visible units the classifier saw.
324 ///
325 /// Objects only. A stream header is counted in [`Self::bytes_shaped`]
326 /// and in [`Self::unshapeable`], and not here.
327 pub objects_seen: u64,
328 /// Bytes the shaper accounted for — every byte it saw, whether a bucket
329 /// granted it, a policy dropped it, or it was unshapeable.
330 /// Deliberately **not** *bytes that passed through a bucket*: the
331 /// `unshapeable` row never touches a bucket, and the conservation identity
332 /// this total exists for — `Σ classes(delivered + dropped) + default +
333 /// unshapeable == bytes_shaped` — has to hold across that row too, or bytes
334 /// the shaper declined to shape would vanish from the accounting.
335 pub bytes_shaped: u64,
336 /// Objects whose `max_hold` elapsed under [`Expiry::ResetStream`].
337 /// Zero under the default [`Expiry::Deliver`], which has no producer.
338 ///
339 /// [`Expiry::ResetStream`]: super::Expiry::ResetStream
340 /// [`Expiry::Deliver`]: super::Expiry::Deliver
341 pub objects_expired: u64,
342 /// Destination streams abandoned by an overflow or expiry policy.
343 pub streams_reset_by_shaping: u64,
344 /// Streams on which two units resolved to different classes.
345 ///
346 /// Head-gating means such a stream's throughput is decided by whichever
347 /// class is at the head, so without this count configured shaping and
348 /// head-of-line blocking are indistinguishable from outside.
349 pub streams_with_mixed_classes: u64,
350 /// The same totals for the client's traffic on its way to the relay.
351 ///
352 /// Every flat total above is this plus [`Self::downlink`], term by
353 /// term. Read a leg when the question is *which side stalled*; read the
354 /// aggregate when it is *did the profile do anything at all*. A session
355 /// shaping only one leg reports the other as all zeros, which is an
356 /// answer rather than an absence.
357 pub uplink: DirectionStats,
358 /// The same totals for the relay's traffic on its way to the client.
359 pub downlink: DirectionStats,
360}
361
362/// Five totals over the traffic travelling **one way**.
363///
364/// Reported in two different containers, and reading a figure here means
365/// knowing which one it came out of. As [`ShapeStats::uplink`] and
366/// [`ShapeStats::downlink`] it is one half of a session's own totals, and
367/// the split is by direction alone — a session recorder has no leg axis. As
368/// one cell of [`ProxyStats::per_leg`] it is one *crossing*: a connection
369/// and a direction together, so the same unit appears in two cells, once
370/// where it was read and once where it was written. [`LegStats`] names the
371/// four.
372///
373/// The distinction is not decoration for the three event figures below.
374/// Every one of them is charged where the traffic **arrived**, so in a
375/// departure cell all three read zero, and in an arrival cell they describe
376/// destination streams that physically live on the *other* connection —
377/// they name which of the two flows the shaper acted on, not which socket
378/// the abandoned stream was on. That is deliberate, and stated on each
379/// field, because putting them on the departure cell would separate them
380/// from the `bytes_shaped` that explains them.
381///
382/// Nothing here is per class: an author who wants a class figure per
383/// direction writes two classes and keys each matcher on
384/// [`Matcher::side`](super::Matcher::side), and these are the totals no
385/// configuration could have separated.
386#[derive(Debug, Clone, Default, PartialEq, Eq)]
387pub struct DirectionStats {
388 /// Hook-visible units the classifier saw. Objects only, for the reason
389 /// [`ShapeStats::objects_seen`] gives — including in a departure cell,
390 /// where a released stream header is bytes and still not an object.
391 pub objects_seen: u64,
392 /// Bytes the shaper accounted for — every byte it saw, whether a bucket
393 /// granted it, a policy dropped it, or it was unshapeable.
394 ///
395 /// The one figure besides `objects_seen` that is measured at both
396 /// crossings, which is what makes the difference between the two cells
397 /// of one direction in [`ProxyStats::per_leg`] the shaper's own
398 /// retention.
399 pub bytes_shaped: u64,
400 /// Objects whose `max_hold` elapsed under
401 /// [`Expiry::ResetStream`](super::Expiry::ResetStream).
402 ///
403 /// Charged to the cell the traffic arrived on. **Zero in a departure
404 /// cell of [`ProxyStats::per_leg`]** — an expired object was read and
405 /// never written, so the leg it would have left by never carried it.
406 pub objects_expired: u64,
407 /// Destination streams abandoned by an overflow or expiry policy,
408 /// counted against the flow whose traffic they were carrying.
409 ///
410 /// Charged to the cell the traffic arrived on, which is **not** the
411 /// connection the abandoned stream is on: a stream carrying what the
412 /// client published is written towards the relay, and this figure
413 /// appears in the client leg's uplink cell beside the `bytes_shaped`
414 /// that explains it. **Zero in a departure cell.**
415 pub streams_reset_by_shaping: u64,
416 /// Streams on which two units resolved to different classes.
417 ///
418 /// Charged to the cell the traffic arrived on, on the same terms as
419 /// [`Self::streams_reset_by_shaping`], and **zero in a departure
420 /// cell**. Note that the event beside it,
421 /// [`ProxyEvent::Impairment`](crate::event::ProxyEvent::Impairment)
422 /// carrying
423 /// [`ImpairmentKind::ClassChangedMidStream`](crate::event::ImpairmentKind::ClassChangedMidStream),
424 /// answers the **other** leg for the same occurrence: an event names the
425 /// connection whose write is affected, and a cell here names the flow
426 /// the figure belongs to. Correlating the two means expecting them to
427 /// disagree by exactly one leg.
428 pub streams_with_mixed_classes: u64,
429}
430
431/// Per-class shaping statistics, one entry in [`ShapeStats`].
432///
433/// Every figure is a count, and there is deliberately no duration among
434/// them. Two once were — a blocked total and a hold total — and both would
435/// have been sums of wall-clock time, movable by the machine's scheduler as
436/// much as by this code, so a gate could only ever have reported them. One
437/// of the two would have misdescribed itself as well:
438/// [`Overflow::Block`](super::Overflow::Block) stops *this crate* calling
439/// `read()` on the source stream and does **not** stall the peer, because
440/// the transport's own receive window absorbs megabytes before a publisher
441/// notices anything — so a blocked time measured here would have been how
442/// long this queue refused to read, never how long the publisher was held
443/// up, and a test reading it as publisher backpressure would have been
444/// measuring the wrong thing under a name that invited it. Real backpressure
445/// needs a small
446/// [`TransportProfile::stream_receive_window`](crate::transport::TransportProfile::stream_receive_window),
447/// which is a transport setting and not a shaping one.
448///
449/// What survives is the load-independent companion a gate can read:
450/// [`Self::blocked_episodes`] and [`Self::tokens_exhausted_episodes`].
451#[derive(Debug, Clone, Default, PartialEq, Eq)]
452pub struct ClassStats {
453 /// The [`ClassRule::name`](super::ClassRule::name) this row reports, or
454 /// an empty string for the default and unshapeable rows.
455 pub name: String,
456 /// Bytes released to the destination stream.
457 pub bytes_delivered: u64,
458 /// Bytes discarded by a policy.
459 pub bytes_dropped: u64,
460 /// Objects released to the destination stream.
461 pub objects_delivered: u64,
462 /// Objects discarded by a policy.
463 pub objects_dropped: u64,
464 /// **Edge-triggered**: distinct starvation episodes, not dequeue
465 /// attempts. Level counting would report the runner's read batching
466 /// rather than the shaper.
467 pub tokens_exhausted_episodes: u64,
468 /// Edge-triggered, same reason: distinct episodes of the read side
469 /// being stalled by [`Overflow::Block`](super::Overflow::Block).
470 pub blocked_episodes: u64,
471 /// Units that waited behind a *different* class's unit on the same
472 /// stream. Separates configured shaping from head-of-line blocking;
473 /// conflating it with `tokens_exhausted_episodes` would hide which of
474 /// the two a run actually produced.
475 pub starved_behind_other_class: u64,
476}
477
478// ── the proxy-wide snapshot ────────────────────────────────────────────
479
480/// Shaping statistics for a whole proxy: every session it has accepted,
481/// including the ones that have already ended.
482///
483/// Read through [`ProxyControl::stats`](crate::control::ProxyControl::stats)
484/// and cleared through
485/// [`ProxyControl::reset_stats`](crate::control::ProxyControl::reset_stats).
486/// The figures are **cumulative and monotone** between resets, which is the
487/// property that separates this from
488/// [`ProxyControl::sessions`](crate::control::ProxyControl::sessions): that
489/// list is what is live now and shrinks when a client disconnects, and a
490/// total summed from it would shrink with it. Nothing here is ever removed,
491/// so a session that ended, errored, or had its whole future dropped
492/// mid-flight has already contributed everything it moved.
493///
494/// # Everything here is gated on a configured [`ShapeProfile`]
495///
496/// The writers are the shaping path's, so a proxy running with no profile
497/// reports `ProxyStats::default()` no matter how many gigabytes it forwards.
498/// An all-zero snapshot means **no profile**, not *no traffic*, and the two
499/// are not distinguishable from this type alone — ask
500/// [`ProxyControl::sessions`](crate::control::ProxyControl::sessions) or an
501/// observer's event stream which of the two it is.
502///
503/// # A session driven directly is not in here
504///
505/// A [`ProxySession`](crate::session::ProxySession) constructed by a caller
506/// rather than accepted by a proxy belongs to no control plane, so it has
507/// nowhere to report and keeps only its own
508/// [`ShapeStats`]. That is the same rule
509/// [`ProxyControl::sessions`](crate::control::ProxyControl::sessions)
510/// follows, and for the same reason: a proxy must not claim traffic it never
511/// accepted.
512#[derive(Debug, Clone, Default, PartialEq, Eq)]
513#[non_exhaustive]
514pub struct ProxyStats {
515 /// One row per connection this proxy holds: index `0` is
516 /// [`Leg::Client`], index `1` is [`Leg::Upstream`]. Reach a row by name
517 /// with [`ProxyStats::leg`] rather than by literal index.
518 ///
519 /// A byte crosses **both** legs — read on one, written on the other —
520 /// so a figure here is charged where it was measured and the two legs
521 /// are not two views of one number. See [`LegStats`] for the cell by
522 /// cell statement.
523 pub per_leg: [LegStats; 2],
524 /// The flat rollup over every session, **derived** at snapshot time
525 /// from [`Self::per_leg`] and the class rows rather than accumulated
526 /// into counters of its own.
527 ///
528 /// Derived for the reason a session's own snapshot sums its two legs:
529 /// two independently written counters can disagree, and the disagreement
530 /// surfaces as an identity that fails for no reason a reader could act
531 /// on. A sum taken at read time cannot drift from its parts, which also
532 /// means the rollup identity is a property of this type's shape and not
533 /// something a test could ever falsify.
534 pub sessions: SessionStats,
535 /// One entry per class, in [`ShapeProfile::classes`] order.
536 ///
537 /// Sized **once**, from the first session this proxy accepts that has a
538 /// class to install — a session with no profile, and a session whose
539 /// profile declares no classes, both leave the rows alone — and never
540 /// resized: a `Class::Rule(index)` is an index into the class
541 /// list its own scheduler was built from, so a row set that changed
542 /// shape under a running session would relabel every figure in it. A
543 /// session whose class list does not match, which is what a live
544 /// [`ProxyControl::set_shape`](crate::control::ProxyControl::set_shape)
545 /// with a different set of classes produces, charges
546 /// [`Self::default_class`] instead of a row that would be named for
547 /// somebody else's rule.
548 pub classes: Vec<ClassStats>,
549 /// Units that matched no rule — and units of a session whose class list
550 /// this proxy's rows were not sized for, for the reason
551 /// [`Self::classes`] gives.
552 pub default_class: ClassStats,
553 /// Units no rule could have seen: stream headers, oversized passthrough
554 /// objects, bypassed streams. A separate row from [`Self::default_class`]
555 /// for the reason [`ShapeStats::unshapeable`] gives.
556 pub unshapeable: ClassStats,
557}
558
559impl ProxyStats {
560 /// One leg's row, by name.
561 ///
562 /// [`Self::per_leg`] is an array so the data path can charge a cell at a
563 /// known offset; a reader should not have to remember which offset that
564 /// is, and an index literal at a call site is exactly the kind of
565 /// mistake that reads plausibly forever.
566 pub fn leg(&self, leg: Leg) -> &LegStats {
567 &self.per_leg[leg_index(leg)]
568 }
569}
570
571/// One connection's shaping statistics, split by which way the traffic was
572/// going — one entry of [`ProxyStats::per_leg`].
573///
574/// Two rows and not one. A leg carries traffic both ways, and a leg that
575/// reported a single row would answer *how much crossed this connection* while
576/// refusing "in which direction" — which is the question an author diagnosing a
577/// one-sided stall is actually asking, and the one no configuration could
578/// separate afterwards.
579///
580/// # Which cell a figure lands in
581///
582/// A proxy reads on one leg and writes on the other, so the same unit is
583/// measured twice, once at each crossing:
584///
585/// * `per_leg[Client].uplink` — read from the client.
586/// * `per_leg[Upstream].uplink` — written to the relay.
587/// * `per_leg[Upstream].downlink` — read from the relay.
588/// * `per_leg[Client].downlink` — written to the client.
589///
590/// The difference between the two cells of one direction is what the shaper
591/// kept back: bytes it read and did not write, whether a policy dropped
592/// them, a hook elided them, or a stream was given up on before they went
593/// out.
594///
595/// # Three fields of a departure cell have no producer
596///
597/// Only the flow of units is measured at both crossings. Of the five
598/// figures a [`DirectionStats`] carries, [`DirectionStats::objects_seen`]
599/// and [`DirectionStats::bytes_shaped`] are charged at both;
600/// [`DirectionStats::objects_expired`],
601/// [`DirectionStats::streams_reset_by_shaping`] and
602/// [`DirectionStats::streams_with_mixed_classes`] are decisions taken over
603/// traffic that **arrived**, so they are charged to the arrival cell only
604/// and a departure cell reports zero for all three. Stated here because a
605/// zero that means "no producer" and a zero that means "it did not happen"
606/// are not the same answer.
607///
608/// `objects_seen` stays the classifier's count on both cells: a stream
609/// header is not an object on the way in, and it is still not one on the
610/// way out.
611#[derive(Debug, Clone, Default, PartialEq, Eq)]
612pub struct LegStats {
613 /// This leg's two directions: index `0` is uplink — the client's traffic
614 /// on its way to the relay — and index `1` is downlink. Reach them by
615 /// name with [`LegStats::uplink`] and [`LegStats::downlink`].
616 pub directions: [DirectionStats; 2],
617}
618
619impl LegStats {
620 /// The client's traffic on its way to the relay, on this leg.
621 pub fn uplink(&self) -> &DirectionStats {
622 &self.directions[Direction::Uplink.index()]
623 }
624
625 /// The relay's traffic on its way to the client, on this leg.
626 pub fn downlink(&self) -> &DirectionStats {
627 &self.directions[Direction::Downlink.index()]
628 }
629}
630
631/// What every session this proxy has run did, added up — the flat form of
632/// [`ProxyStats`].
633///
634/// Every field is **derived** at snapshot time, from [`ProxyStats::per_leg`]
635/// or from the class rows, so it cannot drift from the figures beside it.
636/// The two arrival cells are what the totals below are summed from — the two
637/// points a byte enters this proxy, where each byte is counted exactly once.
638/// Summing all four cells would count every byte twice, once where it was
639/// read and once where it was written.
640///
641/// # Every field here has a producer, and the three that did not are gone
642///
643/// Every figure is derived at snapshot time from the per-leg cells and the
644/// class rows, and those are written only by the shaping path. So a figure
645/// counting something a **hook** does — which needs no
646/// [`ShapeProfile`](super::ShapeProfile), while everything here is gated on
647/// one — could not have been a complete count in this type however it was
648/// wired, and two of the three were exactly that. [`Self::objects_dropped`]
649/// below already said where that kind of figure lives: an object a hook
650/// elided is counted on [`Counters`](crate::instrument::Counters), and a
651/// unit a hook delayed and an object it truncated are counted beside it
652/// there now. The third was a `Duration` total; see [`ClassStats`] for why
653/// no statistic on this page is one.
654#[derive(Debug, Clone, Default, PartialEq, Eq)]
655pub struct SessionStats {
656 /// Hook-visible units the classifier saw, over every session. Objects
657 /// only, for the reason [`ShapeStats::objects_seen`] gives.
658 pub objects_seen: u64,
659 /// Objects discarded by a policy, summed over every class row.
660 ///
661 /// Today that means [`Overflow::DropTail`](super::Overflow::DropTail)
662 /// and nothing else: an object a hook elided is the hook's decision
663 /// rather than the shaper's and is counted by
664 /// [`Counters::objects_elided`](crate::instrument::Counters::objects_elided)
665 /// on the session that ran the hook.
666 pub objects_dropped: u64,
667 /// Objects whose `max_hold` elapsed under
668 /// [`Expiry::ResetStream`](super::Expiry::ResetStream). Zero under the
669 /// default [`Expiry::Deliver`](super::Expiry::Deliver), which has no
670 /// producer — correctly, because that arm delivers the object instead.
671 pub objects_expired: u64,
672 /// Destination streams abandoned by an overflow or expiry policy.
673 ///
674 /// Shaping only. A stream reset by a hook, by a peer, or by a mirrored
675 /// teardown is not counted here — those are not decisions this profile
676 /// took, and folding them in would make a configured
677 /// [`Overflow::ResetStream`](super::Overflow::ResetStream) impossible to
678 /// distinguish from a client that went away.
679 pub streams_reset: u64,
680 /// Bytes the shaper accounted for — every byte it saw, whether a bucket
681 /// granted it, a policy dropped it, or it was unshapeable.
682 pub bytes_shaped: u64,
683}
684
685// ── the storage ────────────────────────────────────────────────────────
686
687/// One class's atomic counters — the storage behind one [`ClassStats`].
688///
689/// `name` is immutable for the recorder's lifetime, so it is a plain
690/// `String` rather than anything shared: the class list cannot change
691/// under a running session.
692pub(crate) struct ClassCounters {
693 name: String,
694 bytes_delivered: AtomicU64,
695 bytes_dropped: AtomicU64,
696 objects_delivered: AtomicU64,
697 objects_dropped: AtomicU64,
698 tokens_exhausted_episodes: AtomicU64,
699 blocked_episodes: AtomicU64,
700 starved_behind_other_class: AtomicU64,
701}
702
703impl ClassCounters {
704 /// An all-zero row labelled `name`.
705 fn named(name: String) -> Self {
706 Self {
707 name,
708 bytes_delivered: AtomicU64::new(0),
709 bytes_dropped: AtomicU64::new(0),
710 objects_delivered: AtomicU64::new(0),
711 objects_dropped: AtomicU64::new(0),
712 tokens_exhausted_episodes: AtomicU64::new(0),
713 blocked_episodes: AtomicU64::new(0),
714 starved_behind_other_class: AtomicU64::new(0),
715 }
716 }
717
718 /// Read this row.
719 fn snapshot(&self) -> ClassStats {
720 ClassStats {
721 name: self.name.clone(),
722 bytes_delivered: self.bytes_delivered.load(Ordering::Relaxed),
723 bytes_dropped: self.bytes_dropped.load(Ordering::Relaxed),
724 objects_delivered: self.objects_delivered.load(Ordering::Relaxed),
725 objects_dropped: self.objects_dropped.load(Ordering::Relaxed),
726 tokens_exhausted_episodes: self.tokens_exhausted_episodes.load(Ordering::Relaxed),
727 blocked_episodes: self.blocked_episodes.load(Ordering::Relaxed),
728 starved_behind_other_class: self.starved_behind_other_class.load(Ordering::Relaxed),
729 }
730 }
731
732 /// Zero every counter, keeping the row's name.
733 ///
734 /// Only [`ProxyRecorder`] resets: a session's figures are the session's
735 /// for its whole life. Seven relaxed stores, and deliberately not one
736 /// atomic swap of the whole row — there is no such primitive, and a
737 /// reset that raced traffic would land between two `fetch_add`s whatever
738 /// it was written with. What that costs is a snapshot straddling a reset
739 /// that reports a row part-cleared, which is why the reset is a caller's
740 /// verb and not something this crate does on its own.
741 fn reset(&self) {
742 self.bytes_delivered.store(0, Ordering::Relaxed);
743 self.bytes_dropped.store(0, Ordering::Relaxed);
744 self.objects_delivered.store(0, Ordering::Relaxed);
745 self.objects_dropped.store(0, Ordering::Relaxed);
746 self.tokens_exhausted_episodes.store(0, Ordering::Relaxed);
747 self.blocked_episodes.store(0, Ordering::Relaxed);
748 self.starved_behind_other_class.store(0, Ordering::Relaxed);
749 }
750}
751
752/// One leg's session totals — the storage behind one [`DirectionStats`].
753struct DirectionCounters {
754 objects_seen: AtomicU64,
755 bytes_shaped: AtomicU64,
756 objects_expired: AtomicU64,
757 streams_reset_by_shaping: AtomicU64,
758 streams_with_mixed_classes: AtomicU64,
759}
760
761impl DirectionCounters {
762 /// An all-zero leg.
763 fn new() -> Self {
764 Self {
765 objects_seen: AtomicU64::new(0),
766 bytes_shaped: AtomicU64::new(0),
767 objects_expired: AtomicU64::new(0),
768 streams_reset_by_shaping: AtomicU64::new(0),
769 streams_with_mixed_classes: AtomicU64::new(0),
770 }
771 }
772
773 /// Read this leg.
774 fn snapshot(&self) -> DirectionStats {
775 DirectionStats {
776 objects_seen: self.objects_seen.load(Ordering::Relaxed),
777 bytes_shaped: self.bytes_shaped.load(Ordering::Relaxed),
778 objects_expired: self.objects_expired.load(Ordering::Relaxed),
779 streams_reset_by_shaping: self.streams_reset_by_shaping.load(Ordering::Relaxed),
780 streams_with_mixed_classes: self.streams_with_mixed_classes.load(Ordering::Relaxed),
781 }
782 }
783
784 /// Zero every counter. [`ClassCounters::reset`] states the terms.
785 fn reset(&self) {
786 self.objects_seen.store(0, Ordering::Relaxed);
787 self.bytes_shaped.store(0, Ordering::Relaxed);
788 self.objects_expired.store(0, Ordering::Relaxed);
789 self.streams_reset_by_shaping.store(0, Ordering::Relaxed);
790 self.streams_with_mixed_classes.store(0, Ordering::Relaxed);
791 }
792}
793
794// ── the proxy-wide storage ─────────────────────────────────────────────
795
796/// Proxy-scoped shaping-counter storage — the storage behind [`ProxyStats`].
797///
798/// One per [`TransparentProxy`](crate::proxy::TransparentProxy), owned by
799/// its control plane and handed to each session's [`ShapeRecorder`] as that
800/// session is attached, so it lives for as long as the proxy does and no
801/// session's ending takes anything out of it.
802///
803/// Every counter here is charged by the same call that charges the session
804/// recorder — [`ShapeRecorder`]'s `note_*` methods forward — which is what
805/// makes the two recorders unable to disagree. The data path pays one extra
806/// relaxed `fetch_add` per figure and one `Option` test, on a path that has
807/// already done a `write_all`.
808pub(crate) struct ProxyRecorder {
809 /// The four cells of [`ProxyStats::per_leg`]: `legs[leg][direction]`,
810 /// indexed by [`leg_index`] then [`Direction::index`].
811 legs: [[DirectionCounters; 2]; 2],
812 /// The class rows, installed by the first session to attach with a class
813 /// to install — see [`Self::adopt_classes`] for the two kinds of session
814 /// that have none and must not win this.
815 ///
816 /// A [`OnceLock`] and not a `Mutex`, because the list must be **fixed**:
817 /// a `Class::Rule(index)` is an index into the class list of the
818 /// scheduler that produced it, so rows that could be resized under a
819 /// running session would silently relabel every figure in them. First
820 /// writer wins; a later session whose class list differs charges
821 /// [`Self::default_class`], which [`Self::row`] does and
822 /// [`ProxyStats::classes`] states.
823 ///
824 /// Empty until the first shaped session, so a proxy that has accepted
825 /// nothing, or only unshaped sessions, reports no class rows rather than
826 /// rows invented from a profile nothing ran under.
827 classes: OnceLock<Vec<ClassCounters>>,
828 default_class: ClassCounters,
829 unshapeable: ClassCounters,
830}
831
832impl ProxyRecorder {
833 /// A recorder with four all-zero cells and no class rows yet.
834 pub(crate) fn new() -> Self {
835 Self {
836 legs: [
837 [DirectionCounters::new(), DirectionCounters::new()],
838 [DirectionCounters::new(), DirectionCounters::new()],
839 ],
840 classes: OnceLock::new(),
841 // Unnamed by contract, exactly as `ShapeRecorder::for_profile`
842 // leaves them: an empty `name` is what distinguishes these two
843 // rows from a class somebody wrote.
844 default_class: ClassCounters::named(String::new()),
845 unshapeable: ClassCounters::named(String::new()),
846 }
847 }
848
849 /// Size the class rows from `names` if nothing has yet, and answer
850 /// whether the installed rows are `names` — which is whether a session
851 /// running that class list may charge them by index.
852 ///
853 /// The comparison is over names **and order**, the same pair
854 /// `same_classes` compares in `session.rs` and for the same reason: that
855 /// pair is what makes a `Class::Rule(index)` mean the same thing to the
856 /// scheduler that produced it and to the row it is charged to. The same
857 /// names in a different order would charge every class to another one's
858 /// row without a single count going missing.
859 ///
860 /// # An empty list never sizes anything
861 ///
862 /// A list with no rows in it names no row, so letting it win the
863 /// `OnceLock` would install an empty row set that nothing can ever match
864 /// again: every classed session accepted for the rest of the proxy's life
865 /// would find rows it did not match and charge [`Self::default_class`],
866 /// with [`ProxyStats::classes`] empty forever. Every figure right, every
867 /// label gone — which is precisely the relabelling this whole sizing rule
868 /// exists to prevent, and it would arrive silently and be unrecoverable
869 /// without restarting the proxy.
870 ///
871 /// The one way an empty list could reach here is a [`ShapeProfile`] with
872 /// no classes, and [`ShapeProfile::try_new`] refuses that outright as
873 /// [`ShapeError::NoClasses`](super::ShapeError::NoClasses), so no
874 /// profile can carry an empty list to this call and the branch below is
875 /// not reachable from any public path.
876 ///
877 /// It stays because of what it costs against what it prevents: two lines
878 /// and a comparison that is already being made, against a proxy-wide,
879 /// silent, restart-only failure. Answering `false` costs a caller nothing
880 /// in any case — a session with no `Class::Rule` to charge is unaffected
881 /// by the flag, and `Class::Default` and `Class::Unshapeable` are
882 /// unaffected by it always.
883 fn adopt_classes(&self, names: &[String]) -> bool {
884 if names.is_empty() {
885 return false;
886 }
887 let rows =
888 self.classes.get_or_init(|| names.iter().cloned().map(ClassCounters::named).collect());
889 rows.len() == names.len() && rows.iter().zip(names).all(|(row, name)| &row.name == name)
890 }
891
892 /// The class rows, or an empty slice before any shaped session attached.
893 fn classes(&self) -> &[ClassCounters] {
894 self.classes.get().map_or(&[], Vec::as_slice)
895 }
896
897 /// The cell a unit **read** on `side` is charged to.
898 fn arrival(&self, side: ProxySide) -> &DirectionCounters {
899 let (leg, direction) = split(side);
900 &self.legs[leg_index(leg)][direction.index()]
901 }
902
903 /// The cell a unit read on `side` is charged to when it is **written**:
904 /// the other leg, the same direction.
905 fn departure(&self, side: ProxySide) -> &DirectionCounters {
906 let (leg, direction) = split(side);
907 &self.legs[leg_index(opposite(leg))][direction.index()]
908 }
909
910 /// The row one class's units are charged to.
911 ///
912 /// `sized` is whether the charging session's class list is the one these
913 /// rows were sized from. When it is not, a `Class::Rule` index names a
914 /// row belonging to a different rule, so the unit goes to the default
915 /// row instead — the same answer [`ShapeRecorder::row`] gives an
916 /// impossible index, and for the stronger of the two reasons: here the
917 /// index is not impossible, it is *plausible and wrong*.
918 ///
919 /// [`Class::Unshapeable`] is unaffected either way. That row is not
920 /// named for a rule, so no reconfiguration can make it mean something
921 /// else.
922 fn row(&self, class: Class, sized: bool) -> &ClassCounters {
923 match class {
924 Class::Unshapeable => &self.unshapeable,
925 Class::Rule(index) if sized => self.classes().get(index).unwrap_or(&self.default_class),
926 _ => &self.default_class,
927 }
928 }
929
930 /// Read every counter.
931 ///
932 /// Allocates: one `Vec` and one `String` per class row. A reader's call —
933 /// the control plane's — never the data path's.
934 ///
935 /// [`ProxyStats::sessions`] is **computed here**, from the two arrival
936 /// cells and the class rows, which is why no writer maintains it.
937 pub(crate) fn snapshot(&self) -> ProxyStats {
938 let per_leg = [self.leg_stats(Leg::Client), self.leg_stats(Leg::Upstream)];
939 let classes: Vec<ClassStats> = self.classes().iter().map(ClassCounters::snapshot).collect();
940 let default_class = self.default_class.snapshot();
941 let unshapeable = self.unshapeable.snapshot();
942
943 // The two cells traffic *enters* by. Every other cell reports what
944 // left, and a byte that entered and left would be counted twice.
945 let from_client = per_leg[leg_index(Leg::Client)].uplink();
946 let from_relay = per_leg[leg_index(Leg::Upstream)].downlink();
947 let objects_dropped = classes
948 .iter()
949 .chain([&default_class, &unshapeable])
950 .fold(0u64, |sum, row| sum.saturating_add(row.objects_dropped));
951
952 let sessions = SessionStats {
953 objects_seen: from_client.objects_seen.saturating_add(from_relay.objects_seen),
954 objects_dropped,
955 objects_expired: from_client.objects_expired.saturating_add(from_relay.objects_expired),
956 streams_reset: from_client
957 .streams_reset_by_shaping
958 .saturating_add(from_relay.streams_reset_by_shaping),
959 bytes_shaped: from_client.bytes_shaped.saturating_add(from_relay.bytes_shaped),
960 };
961
962 ProxyStats { per_leg, sessions, classes, default_class, unshapeable }
963 }
964
965 /// One leg's two cells.
966 fn leg_stats(&self, leg: Leg) -> LegStats {
967 let rows = &self.legs[leg_index(leg)];
968 LegStats { directions: [rows[0].snapshot(), rows[1].snapshot()] }
969 }
970
971 /// Zero every counter this proxy holds, keeping the class rows and their
972 /// names.
973 ///
974 /// The rows survive because they are what a `Class::Rule(index)` means:
975 /// dropping them would let the next session install a different class
976 /// list and charge figures under it, which is precisely the relabelling
977 /// [`Self::adopt_classes`] exists to prevent. A reset moves the counters
978 /// to zero, not the schema.
979 pub(crate) fn reset(&self) {
980 for leg in &self.legs {
981 for cell in leg {
982 cell.reset();
983 }
984 }
985 for row in self.classes() {
986 row.reset();
987 }
988 self.default_class.reset();
989 self.unshapeable.reset();
990 }
991}
992
993impl std::fmt::Debug for ProxyRecorder {
994 /// Prints the snapshot, not the atomics.
995 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
996 f.debug_tuple("ProxyRecorder").field(&self.snapshot()).finish()
997 }
998}
999
1000/// Session-scoped shaping-counter storage.
1001///
1002/// One per `ProxySession`, held behind an `Arc` and cloned into every
1003/// forwarding task exactly as `Recorder` is, so two sessions running
1004/// side by side in one test binary cannot see each other's increments.
1005///
1006/// **Always constructed**, including for a session with no profile, on the
1007/// same reasoning `StreamRegistry` is. An unshaped session's
1008/// recorder has an empty class list and no writer, so it snapshots as
1009/// `ShapeStats::default()`; making construction conditional would replace
1010/// one always-zero allocation with an `Option` on the hot context and prove
1011/// nothing extra.
1012pub(crate) struct ShapeRecorder {
1013 classes: Vec<ClassCounters>,
1014 default_class: ClassCounters,
1015 unshapeable: ClassCounters,
1016 /// The session totals, one row per leg, indexed by [`Direction`].
1017 ///
1018 /// One recorder serves every forwarding task of a session, so a single
1019 /// row would answer "how much was shaped" and refuse to answer "on
1020 /// which side" — the question an author shaping both legs is actually
1021 /// asking. Two rows and an index cost the writers nothing: charging a
1022 /// total is still one relaxed `fetch_add` at a known offset.
1023 totals: [DirectionCounters; 2],
1024 /// This session's proxy's counters, or `None` when it has no proxy.
1025 ///
1026 /// Forwarded to by every writer below rather than reached by a separate
1027 /// call at each site. One call charges both recorders or neither, so
1028 /// there is no way to add a producer to a session figure and forget the
1029 /// proxy figure beside it — which is the failure a second set of call
1030 /// sites would make inevitable and silent.
1031 ///
1032 /// `None` is not a degraded mode. A session constructed directly belongs
1033 /// to no proxy, so there is no aggregate for it to be part of, and it
1034 /// keeps reporting its own [`ShapeStats`] exactly as it always did.
1035 proxy: Option<Arc<ProxyRecorder>>,
1036 /// Whether this session's class list is the one the proxy's rows were
1037 /// sized from, which decides whether a `Class::Rule(index)` may address
1038 /// them. Resolved once at attach; see [`ProxyRecorder::row`].
1039 proxy_sized: bool,
1040}
1041
1042impl ShapeRecorder {
1043 /// Pre-size the rows for `profile`'s classes, in configured order.
1044 ///
1045 /// `None` gives a recorder with no class rows — the unshaped session's
1046 /// shape, whose snapshot is `ShapeStats::default()`.
1047 ///
1048 /// The recorder this builds reports to no proxy. That is the right
1049 /// answer for a session driven directly, and
1050 /// [`Self::attached`] is what the accept loop
1051 /// uses instead.
1052 pub(crate) fn for_profile(profile: Option<&ShapeProfile>) -> Self {
1053 let classes = profile
1054 .map(|p| p.classes().iter().map(|c| ClassCounters::named(c.name.clone())).collect())
1055 .unwrap_or_default();
1056 Self {
1057 classes,
1058 // The default and unshapeable rows are unnamed by contract:
1059 // an empty `name` is what distinguishes them from a class the
1060 // user wrote, and a user-written class name is unique by
1061 // `ShapeError::DuplicateClassName`, so there is no collision.
1062 default_class: ClassCounters::named(String::new()),
1063 unshapeable: ClassCounters::named(String::new()),
1064 totals: [DirectionCounters::new(), DirectionCounters::new()],
1065 proxy: None,
1066 proxy_sized: false,
1067 }
1068 }
1069
1070 /// The same recorder, additionally reporting into `proxy`.
1071 ///
1072 /// Built at the one moment a session is attached to a control plane —
1073 /// after it is constructed and before it runs — so the recorder that
1074 /// forwards is the recorder every forwarding task will clone, and no
1075 /// figure is charged before the forwarding target is in place.
1076 ///
1077 /// An unshaped session does **not** size the proxy's class rows. It has
1078 /// no classes to install, and installing its empty list would mean the
1079 /// first shaped session to arrive afterwards found rows it did not match
1080 /// and charged the default row forever. Nothing is lost by skipping it:
1081 /// an unshaped session's writers are all behind a configured profile, so
1082 /// it charges nothing at all.
1083 ///
1084 /// A **shaped** session cannot be in the same position any more:
1085 /// [`ShapeProfile::try_new`] refuses a profile with no classes, so
1086 /// `Some(profile)` always carries at least one class name to install.
1087 /// [`ProxyRecorder::adopt_classes`] keeps its own guard against an empty
1088 /// list all the same, and says there why.
1089 pub(crate) fn attached(profile: Option<&ShapeProfile>, proxy: Arc<ProxyRecorder>) -> Self {
1090 let mut recorder = Self::for_profile(profile);
1091 recorder.proxy_sized = match profile {
1092 Some(profile) => {
1093 let names: Vec<String> = profile.classes().iter().map(|c| c.name.clone()).collect();
1094 proxy.adopt_classes(&names)
1095 }
1096 None => false,
1097 };
1098 recorder.proxy = Some(proxy);
1099 recorder
1100 }
1101
1102 /// This session's proxy row for `class`, when it has a proxy.
1103 fn proxy_row(&self, class: Class) -> Option<&ClassCounters> {
1104 self.proxy.as_ref().map(|proxy| proxy.row(class, self.proxy_sized))
1105 }
1106
1107 /// Snapshot every counter for this session.
1108 ///
1109 /// Allocates: one `Vec` and one `String` per class row. Called by a
1110 /// reader (a test, or a control plane), never on the data path.
1111 ///
1112 /// The flat session totals are **computed here** from the two legs,
1113 /// which is why no writer maintains them. Two independently written
1114 /// counters can disagree, and the disagreement would surface as a
1115 /// conservation identity that fails for no reason a reader could act
1116 /// on; a sum taken at read time cannot.
1117 pub(crate) fn snapshot(&self) -> ShapeStats {
1118 let uplink = self.leg(Direction::Uplink).snapshot();
1119 let downlink = self.leg(Direction::Downlink).snapshot();
1120 ShapeStats {
1121 classes: self.classes.iter().map(ClassCounters::snapshot).collect(),
1122 default_class: self.default_class.snapshot(),
1123 unshapeable: self.unshapeable.snapshot(),
1124 objects_seen: uplink.objects_seen.saturating_add(downlink.objects_seen),
1125 bytes_shaped: uplink.bytes_shaped.saturating_add(downlink.bytes_shaped),
1126 objects_expired: uplink.objects_expired.saturating_add(downlink.objects_expired),
1127 streams_reset_by_shaping: uplink
1128 .streams_reset_by_shaping
1129 .saturating_add(downlink.streams_reset_by_shaping),
1130 streams_with_mixed_classes: uplink
1131 .streams_with_mixed_classes
1132 .saturating_add(downlink.streams_with_mixed_classes),
1133 uplink,
1134 downlink,
1135 }
1136 }
1137
1138 /// One framed unit entered the shaping path, carrying `bytes` on the
1139 /// wire.
1140 ///
1141 /// Two relaxed `fetch_add`s, called from the object arm of
1142 /// `pipe_data_framed` **behind `ForwardCtx::shaping_enabled`** — so a
1143 /// session with no profile never reaches it, and `interest_none.rs`
1144 /// stays at an all-zero `Counters` and an all-zero `ShapeStats`
1145 /// together.
1146 ///
1147 /// This is the *seen* count, not a delivery count: it is taken where
1148 /// the unit becomes visible to the shaper, before any classification or
1149 /// bucket exists to say what became of it. `bytes_shaped` is therefore
1150 /// the left-hand side of the conservation identity from the start, and
1151 /// the units that add the per-class rows are adding the right-hand
1152 /// side rather than re-defining this one.
1153 /// `side` is the side the unit **arrived** on, which the caller already
1154 /// holds as the side it reports every other event for. Charging it here
1155 /// rather than deriving it later is what keeps *the downlink stalled*
1156 /// separable from "the uplink did", and it is a side rather than a
1157 /// [`Direction`] because the proxy-wide rows need the leg as well — a
1158 /// caller that passed a direction would have thrown away exactly the half
1159 /// of the answer [`ProxyStats::per_leg`] exists for.
1160 pub(crate) fn note_object_seen(&self, side: ProxySide, bytes: u64) {
1161 let leg = self.leg(Direction::from(side));
1162 leg.objects_seen.fetch_add(1, Ordering::Relaxed);
1163 leg.bytes_shaped.fetch_add(bytes, Ordering::Relaxed);
1164 if let Some(proxy) = &self.proxy {
1165 let cell = proxy.arrival(side);
1166 cell.objects_seen.fetch_add(1, Ordering::Relaxed);
1167 cell.bytes_shaped.fetch_add(bytes, Ordering::Relaxed);
1168 }
1169 }
1170
1171 /// `bytes` no rule could see entered the shaping path.
1172 /// The sibling of [`Self::note_object_seen`] for a unit with no
1173 /// `ObjectMeta` — a stream header, an oversized object's passthrough chunk,
1174 /// a bypassed stream's tail. One relaxed `fetch_add`, and deliberately
1175 /// **not** two: `objects_seen` counts what the *classifier* saw, and a
1176 /// header is not an object. Bumping it here would make the count that every
1177 /// fixture anchors on (*wait until all twelve objects have been
1178 /// classified*) depend on how many stream headers happened to arrive first.
1179 ///
1180 /// `bytes_shaped` does move, because it is the left-hand side of the
1181 /// conservation identity and the `unshapeable` row is one of that
1182 /// identity's right-hand terms. The row itself is charged on release,
1183 /// from the same `unit.len()`.
1184 pub(crate) fn note_unshapeable_seen(&self, side: ProxySide, bytes: u64) {
1185 self.leg(Direction::from(side)).bytes_shaped.fetch_add(bytes, Ordering::Relaxed);
1186 if let Some(proxy) = &self.proxy {
1187 proxy.arrival(side).bytes_shaped.fetch_add(bytes, Ordering::Relaxed);
1188 }
1189 }
1190
1191 /// One unit of `bytes` was discarded by [`Overflow::DropTail`], charged
1192 /// to the class that claimed it.
1193 ///
1194 /// No side, because nothing this moves is per leg: a drop is a fact
1195 /// about a *rule*, and the leg it happened on is already in the
1196 /// difference between the two cells of that direction — the bytes were
1197 /// charged where they arrived and are never charged where they would
1198 /// have left.
1199 ///
1200 /// [`Overflow::DropTail`]: super::Overflow::DropTail
1201 pub(crate) fn note_dropped(&self, class: Class, bytes: u64) {
1202 for row in [Some(self.row(class)), self.proxy_row(class)].into_iter().flatten() {
1203 row.objects_dropped.fetch_add(1, Ordering::Relaxed);
1204 row.bytes_dropped.fetch_add(bytes, Ordering::Relaxed);
1205 }
1206 }
1207
1208 /// One **episode** of the read side being stalled by
1209 /// [`Overflow::Block`] began.
1210 ///
1211 /// Edge-triggered by the caller, which owns the per-stream latch: level
1212 /// counting here would report the runner's read batching rather than
1213 /// the shaper. Charged to the class of the last unit classified on the
1214 /// stream — the one whose admission filled the queue — because a stall
1215 /// is a property of a stream and a stream has no single class.
1216 ///
1217 /// [`Overflow::Block`]: super::Overflow::Block
1218 pub(crate) fn note_blocked(&self, class: Class) {
1219 for row in [Some(self.row(class)), self.proxy_row(class)].into_iter().flatten() {
1220 row.blocked_episodes.fetch_add(1, Ordering::Relaxed);
1221 }
1222 }
1223
1224 /// A destination stream carrying traffic that arrived on `side` was
1225 /// abandoned by a shaping policy.
1226 ///
1227 /// Charged to the **arrival** cell, like every other decision figure,
1228 /// and not to the leg the abandoned stream is physically on. What a
1229 /// reader wants from it is which of the two flows the profile gave up
1230 /// on, and the flow is named by where its traffic came from; splitting
1231 /// this one figure the other way would put it in a different cell from
1232 /// the `bytes_shaped` that explains it.
1233 pub(crate) fn note_stream_reset_by_shaping(&self, side: ProxySide) {
1234 self.leg(Direction::from(side)).streams_reset_by_shaping.fetch_add(1, Ordering::Relaxed);
1235 if let Some(proxy) = &self.proxy {
1236 proxy.arrival(side).streams_reset_by_shaping.fetch_add(1, Ordering::Relaxed);
1237 }
1238 }
1239
1240 /// One unit of `bytes` was released to the destination stream.
1241 ///
1242 /// The right-hand term the conservation identity was missing: with this
1243 /// producer in place, `Σ classes(delivered + dropped) + default +
1244 /// unshapeable == bytes_shaped` holds for a stream that ran to
1245 /// completion. It is charged from the same `raw.len()` `note_object_seen`
1246 /// took, so the identity is over one measurement rather than two.
1247 /// A teardown flush counts here too. `write_all` returning is the only
1248 /// definition of *released to the destination stream* this side of the
1249 /// transport, and the bytes it could not vouch for are separately reported
1250 /// as `QueuedBytesAtTeardown` — an over-report there is a diagnosable
1251 /// nuisance, a byte missing from both is a hole in the identity.
1252 ///
1253 /// `side` is still the side the unit **arrived** on — every caller holds
1254 /// that one and only that one — and it is the proxy-wide rows that turn
1255 /// it around: a release is the second and last time a unit is measured,
1256 /// on the leg it is leaving by, so [`ProxyRecorder::departure`] is what
1257 /// this charges. The session's own rows are untouched by that, because
1258 /// they have no leg axis to be wrong about.
1259 ///
1260 /// The proxy's departure cell counts an object only for a unit that
1261 /// carried one. [`Class::Unshapeable`] is the tag every unit no rule
1262 /// could see is pushed with — a stream header, an oversized passthrough
1263 /// chunk — and `objects_seen` is the *classifier's* count on both cells
1264 /// or it is not one figure at all: a header is not an object on the way
1265 /// in and it is not one on the way out.
1266 pub(crate) fn note_delivered(&self, side: ProxySide, class: Class, bytes: u64) {
1267 for row in [Some(self.row(class)), self.proxy_row(class)].into_iter().flatten() {
1268 row.objects_delivered.fetch_add(1, Ordering::Relaxed);
1269 row.bytes_delivered.fetch_add(bytes, Ordering::Relaxed);
1270 }
1271 if let Some(proxy) = &self.proxy {
1272 let cell = proxy.departure(side);
1273 cell.bytes_shaped.fetch_add(bytes, Ordering::Relaxed);
1274 if class != Class::Unshapeable {
1275 cell.objects_seen.fetch_add(1, Ordering::Relaxed);
1276 }
1277 }
1278 }
1279
1280 /// One **episode** of a class's own bucket being dry began.
1281 /// Edge-triggered by the caller, which owns the per-stream latch, for the
1282 /// same reason [`Self::note_blocked`] is: a level count would report how
1283 /// often the release branch woke rather than how often the shaper ran out.
1284 /// Deliberately **not** bumped when a [`Discipline`](super::Discipline) is
1285 /// what held the unit back — that is [`Self::note_starved`], and conflating
1286 /// the two would make *my bucket is too small* and *another class is ahead
1287 /// of me* one unactionable number.
1288 pub(crate) fn note_tokens_exhausted(&self, class: Class) {
1289 for row in [Some(self.row(class)), self.proxy_row(class)].into_iter().flatten() {
1290 row.tokens_exhausted_episodes.fetch_add(1, Ordering::Relaxed);
1291 }
1292 }
1293
1294 /// One unit waited behind a unit of a *different* class.
1295 ///
1296 /// Per unit, once — the queue marks a unit as it charges it. Both causes
1297 /// of head-of-line waiting land here: a discipline holding this class
1298 /// back behind a rival, and a same-stream head of another class that the
1299 /// single per-stream FIFO cannot be reordered around.
1300 pub(crate) fn note_starved(&self, class: Class) {
1301 for row in [Some(self.row(class)), self.proxy_row(class)].into_iter().flatten() {
1302 row.starved_behind_other_class.fetch_add(1, Ordering::Relaxed);
1303 }
1304 }
1305
1306 /// One queued object outlived its clamp under
1307 /// [`Expiry::ResetStream`](super::Expiry::ResetStream).
1308 ///
1309 /// Zero under the default [`Expiry::Deliver`](super::Expiry::Deliver),
1310 /// which delivers the object instead of expiring it — that arm has no
1311 /// producer here and deliberately so.
1312 ///
1313 /// Charged to `side`'s **arrival** cell: the object was read and never
1314 /// written, so the leg it would have left by never carried it.
1315 pub(crate) fn note_expired(&self, side: ProxySide) {
1316 self.leg(Direction::from(side)).objects_expired.fetch_add(1, Ordering::Relaxed);
1317 if let Some(proxy) = &self.proxy {
1318 proxy.arrival(side).objects_expired.fetch_add(1, Ordering::Relaxed);
1319 }
1320 }
1321
1322 /// One stream carried units of two different classes.
1323 ///
1324 /// Once per stream, latched by the caller. Head-gating means such a
1325 /// stream's throughput is decided by whichever class is at its head, so
1326 /// without this count a caller cannot tell configured shaping
1327 /// from head-of-line blocking.
1328 pub(crate) fn note_mixed_class_stream(&self, side: ProxySide) {
1329 self.leg(Direction::from(side)).streams_with_mixed_classes.fetch_add(1, Ordering::Relaxed);
1330 if let Some(proxy) = &self.proxy {
1331 proxy.arrival(side).streams_with_mixed_classes.fetch_add(1, Ordering::Relaxed);
1332 }
1333 }
1334
1335 /// The row one class's units are charged to.
1336 ///
1337 /// A [`Class::Rule`] index out of range cannot happen — the indices
1338 /// come from the same profile the rows were pre-sized from — but this
1339 /// is on a forwarding task, where a total function beats a panicking
1340 /// one: an impossible index charges the default row rather than killing
1341 /// the stream.
1342 fn row(&self, class: Class) -> &ClassCounters {
1343 match class {
1344 Class::Rule(index) => self.classes.get(index).unwrap_or(&self.default_class),
1345 Class::Default => &self.default_class,
1346 Class::Unshapeable => &self.unshapeable,
1347 }
1348 }
1349
1350 /// The session totals one leg's units are charged to.
1351 fn leg(&self, direction: Direction) -> &DirectionCounters {
1352 &self.totals[direction.index()]
1353 }
1354}
1355
1356impl std::fmt::Debug for ShapeRecorder {
1357 /// Prints the snapshot, not the atomics.
1358 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1359 f.debug_tuple("ShapeRecorder").field(&self.snapshot()).finish()
1360 }
1361}
1362
1363#[cfg(test)]
1364mod tests {
1365 use super::*;
1366 use crate::shape::{BucketConfig, ClassRule, Discipline, Matcher, QueueConfig};
1367
1368 /// A valid profile whose classes are `names`, in that order. Struct
1369 /// literals rather than field assignment: `#[non_exhaustive]` does not
1370 /// apply inside the defining crate, and `..Default::default()` is what
1371 /// clippy's `field_reassign_with_default` asks for.
1372 fn profile(names: &[&str]) -> ShapeProfile {
1373 let bucket = BucketConfig { name: "b".to_string(), ..BucketConfig::default() };
1374 let classes = names
1375 .iter()
1376 .map(|n| ClassRule {
1377 name: (*n).to_string(),
1378 bucket: "b".to_string(),
1379 matcher: Matcher::default(),
1380 ..ClassRule::default()
1381 })
1382 .collect();
1383 ShapeProfile::try_new(vec![bucket], classes, QueueConfig::default(), Discipline::Fifo)
1384 .expect("the fixture names its own bucket and its classes are unique")
1385 }
1386
1387 /// No profile means no rows and an all-default snapshot — the claim
1388 /// `interest_none.rs` rests on, checked without a session.
1389 /// *Ablation, recorded:* give the `None` arm of `for_profile` one row
1390 /// (`unwrap_or_else(|| vec![ClassCounters::named(String::new())])`). Every
1391 /// counter is still zero and the compare still reddens, on `classes:
1392 /// [ClassStats { .. }]` against `classes: []` — which is the point: a
1393 /// `ShapeStats` that is *all zeros but shaped* is not `default()`, and the
1394 /// integration gate compares the whole struct.
1395 #[test]
1396 fn an_unshaped_recorder_snapshots_as_default() {
1397 let rec = ShapeRecorder::for_profile(None);
1398 assert_eq!(rec.snapshot(), ShapeStats::default());
1399 }
1400
1401 /// Rows are pre-sized and ordered by the configured class order, not by
1402 /// name and not by first use — a class that never saw a unit is present
1403 /// with a zero row.
1404 ///
1405 /// Order is not decoration: it is what lets the data path address a row
1406 /// by index instead of by name, which is the whole reason the storage
1407 /// is a `Vec` and not a map. The two class names are deliberately in
1408 /// non-alphabetical order, so a snapshot that sorted would redden here
1409 /// too.
1410 ///
1411 /// *Ablation, recorded:* `.rev()` the class iteration in `for_profile`
1412 /// — `left: ["audio", "video"]`, `right: ["video", "audio"]`.
1413 #[test]
1414 fn rows_are_pre_sized_in_configured_order() {
1415 let rec = ShapeRecorder::for_profile(Some(&profile(&["video", "audio"])));
1416 let stats = rec.snapshot();
1417 assert_eq!(
1418 stats.classes.iter().map(|c| c.name.as_str()).collect::<Vec<_>>(),
1419 vec!["video", "audio"],
1420 "snapshot order is the configured order, so a reader can index it"
1421 );
1422 assert!(
1423 stats
1424 .classes
1425 .iter()
1426 .all(|c| *c == ClassStats { name: c.name.clone(), ..ClassStats::default() }),
1427 "a class that saw nothing reports a zero row rather than being absent"
1428 );
1429 }
1430
1431 /// The see-point producer moves exactly two totals and leaves every
1432 /// class row alone.
1433 ///
1434 /// Two calls with different byte counts, so a `bytes_shaped` that
1435 /// counted calls rather than bytes, or that overwrote rather than
1436 /// accumulated, is separable from one that adds: only addition gives
1437 /// `2` and `42` together.
1438 ///
1439 /// The three zero assertions are the boundary the release side moves.
1440 /// Attribution is charged when a unit is *released*, from the same
1441 /// measurement taken here, so a class row moving at the see-point would
1442 /// mean the same byte counted twice on the right-hand side of the
1443 /// conservation identity.
1444 ///
1445 /// *Ablation, recorded:* have `note_object_seen` also bump
1446 /// `default_class.objects_delivered` —
1447 /// `left: ClassStats { .., objects_delivered: 2, .. }` against
1448 /// `right: ClassStats { .., objects_delivered: 0, .. }`.
1449 #[test]
1450 fn note_object_seen_moves_the_session_totals_only() {
1451 let rec = ShapeRecorder::for_profile(Some(&profile(&["video"])));
1452 rec.note_object_seen(ProxySide::ClientToProxy, 40);
1453 rec.note_object_seen(ProxySide::ClientToProxy, 2);
1454 let stats = rec.snapshot();
1455 assert_eq!(stats.objects_seen, 2);
1456 assert_eq!(stats.bytes_shaped, 42);
1457 assert_eq!(stats.classes[0].bytes_delivered, 0, "release is what attributes bytes");
1458 assert_eq!(stats.default_class, ClassStats::default());
1459 assert_eq!(stats.unshapeable, ClassStats::default());
1460 }
1461
1462 /// One recorder, both legs busy: every session total lands on the side
1463 /// it was charged from, and the flat total is the two sides added.
1464 ///
1465 /// This is the shape of a session shaping in both directions, which is
1466 /// the case a single set of totals cannot report: the recorder is
1467 /// shared by every forwarding task, so before the split "5 objects
1468 /// seen" was compatible with 5/0, 0/5 and anything between, and an
1469 /// author diagnosing a stall could not tell which leg had it.
1470 ///
1471 /// Both legs are compared **whole** rather than field by field, so a
1472 /// figure leaking into a neighbouring total on the correct leg reddens
1473 /// too. Every quantity differs between the legs — 2 objects against 3,
1474 /// 49 bytes against 605, one reset against three — so swapping the two
1475 /// rows fails on all of them rather than on none.
1476 ///
1477 /// The sums are stated because they are the guarantee a reader relies on
1478 /// when mixing the two forms; they hold by construction of `snapshot`,
1479 /// which sums the legs rather than keeping a third counter. What the
1480 /// per-leg equalities above them falsify is the attribution, and that
1481 /// is the part no arithmetic guarantees.
1482 ///
1483 /// *Ablation, recorded:* swap the two legs in `snapshot`, so each row is
1484 /// reported under the other's name. Every sum below still passes —
1485 /// addition does not care which order it is given — and the uplink
1486 /// compare reddens with `left: DirectionStats { objects_seen: 3,
1487 /// bytes_shaped: 605, objects_expired: 2, streams_reset_by_shaping: 3,
1488 /// streams_with_mixed_classes: 0 }` against `right: DirectionStats {
1489 /// objects_seen: 2, bytes_shaped: 49, objects_expired: 0,
1490 /// streams_reset_by_shaping: 1, streams_with_mixed_classes: 1 }`. That
1491 /// is the whole case for the per-leg equalities: they are the part the
1492 /// arithmetic cannot check.
1493 ///
1494 /// *Ablation, recorded:* collapse the split instead — have
1495 /// `ShapeRecorder::leg` ignore its argument and always answer
1496 /// `&self.totals[0]`. The uplink compare reddens with `left:
1497 /// DirectionStats { objects_seen: 5, bytes_shaped: 654, objects_expired:
1498 /// 2, streams_reset_by_shaping: 4, streams_with_mixed_classes: 1 }`, and
1499 /// `note_object_seen_moves_the_session_totals_only` goes with it on
1500 /// `left: 4 right: 2` — the aggregate counts the one surviving row
1501 /// twice.
1502 #[test]
1503 fn a_bidirectional_run_charges_each_leg_and_the_legs_sum_to_the_aggregate() {
1504 let rec = ShapeRecorder::for_profile(Some(&profile(&["video", "audio"])));
1505
1506 // Uplink: two objects, a stream header no rule could see, one
1507 // mixed-class stream and one stream the policy gave up on.
1508 rec.note_object_seen(ProxySide::ClientToProxy, 40);
1509 rec.note_object_seen(ProxySide::ClientToProxy, 2);
1510 rec.note_unshapeable_seen(ProxySide::ClientToProxy, 7);
1511 rec.note_mixed_class_stream(ProxySide::ClientToProxy);
1512 rec.note_stream_reset_by_shaping(ProxySide::ClientToProxy);
1513
1514 // Downlink: more of everything, and two expiries the uplink has
1515 // none of.
1516 rec.note_object_seen(ProxySide::RelayToProxy, 100);
1517 rec.note_object_seen(ProxySide::RelayToProxy, 200);
1518 rec.note_object_seen(ProxySide::RelayToProxy, 300);
1519 rec.note_unshapeable_seen(ProxySide::RelayToProxy, 5);
1520 rec.note_expired(ProxySide::RelayToProxy);
1521 rec.note_expired(ProxySide::RelayToProxy);
1522 rec.note_stream_reset_by_shaping(ProxySide::RelayToProxy);
1523 rec.note_stream_reset_by_shaping(ProxySide::RelayToProxy);
1524 rec.note_stream_reset_by_shaping(ProxySide::RelayToProxy);
1525
1526 let stats = rec.snapshot();
1527 assert_eq!(
1528 stats.uplink,
1529 DirectionStats {
1530 objects_seen: 2,
1531 bytes_shaped: 49,
1532 objects_expired: 0,
1533 streams_reset_by_shaping: 1,
1534 streams_with_mixed_classes: 1,
1535 },
1536 "the uplink reports what the uplink was charged, and nothing else"
1537 );
1538 assert_eq!(
1539 stats.downlink,
1540 DirectionStats {
1541 objects_seen: 3,
1542 bytes_shaped: 605,
1543 objects_expired: 2,
1544 streams_reset_by_shaping: 3,
1545 streams_with_mixed_classes: 0,
1546 },
1547 "the downlink's two expiries are its own: a starved downlink must \
1548 not read as a starved uplink"
1549 );
1550
1551 assert_eq!(stats.uplink.objects_seen + stats.downlink.objects_seen, stats.objects_seen);
1552 assert_eq!(stats.uplink.bytes_shaped + stats.downlink.bytes_shaped, stats.bytes_shaped);
1553 assert_eq!(
1554 stats.uplink.objects_expired + stats.downlink.objects_expired,
1555 stats.objects_expired
1556 );
1557 assert_eq!(
1558 stats.uplink.streams_reset_by_shaping + stats.downlink.streams_reset_by_shaping,
1559 stats.streams_reset_by_shaping
1560 );
1561 assert_eq!(
1562 stats.uplink.streams_with_mixed_classes + stats.downlink.streams_with_mixed_classes,
1563 stats.streams_with_mixed_classes
1564 );
1565
1566 // Splitting the totals must not have started attributing anything:
1567 // the class rows are release's, and nothing here released a unit.
1568 assert!(
1569 stats
1570 .classes
1571 .iter()
1572 .all(|c| *c == ClassStats { name: c.name.clone(), ..ClassStats::default() }),
1573 "no class row moved, on either leg"
1574 );
1575 assert_eq!(stats.default_class, ClassStats::default());
1576 assert_eq!(stats.unshapeable, ClassStats::default());
1577 }
1578
1579 /// Every ingress side maps to the leg its traffic is on, and each
1580 /// egress side maps to the same leg as the ingress side it pairs with.
1581 ///
1582 /// The mapping is total because the recorder has no fifth row to put a
1583 /// surprise on: a `ProxySide` that fell through would have to invent a
1584 /// leg, and an invented leg is a byte silently attributed to the wrong
1585 /// side of the session.
1586 #[test]
1587 fn every_side_maps_to_the_leg_its_traffic_travels_on() {
1588 assert_eq!(Direction::from(ProxySide::ClientToProxy), Direction::Uplink);
1589 assert_eq!(Direction::from(ProxySide::ProxyToRelay), Direction::Uplink);
1590 assert_eq!(Direction::from(ProxySide::RelayToProxy), Direction::Downlink);
1591 assert_eq!(Direction::from(ProxySide::ProxyToClient), Direction::Downlink);
1592 }
1593
1594 /// Every side names one connection and one way along it, and the four
1595 /// pairs are all different.
1596 ///
1597 /// The pair is what [`ProxyStats::per_leg`] is indexed by, so a mapping
1598 /// that sent two sides to the same cell would merge two flows with
1599 /// nothing going red anywhere else — the aggregate would still be right.
1600 /// Written as four equalities against the four pairs rather than as a
1601 /// round trip, because the claim is *which* pair, not that some pair
1602 /// exists.
1603 #[test]
1604 fn every_side_names_one_cell_of_the_two_by_two() {
1605 assert_eq!(split(ProxySide::ClientToProxy), (Leg::Client, Direction::Uplink));
1606 assert_eq!(split(ProxySide::ProxyToRelay), (Leg::Upstream, Direction::Uplink));
1607 assert_eq!(split(ProxySide::RelayToProxy), (Leg::Upstream, Direction::Downlink));
1608 assert_eq!(split(ProxySide::ProxyToClient), (Leg::Client, Direction::Downlink));
1609
1610 // Four sides, four cells, no collisions: the property the four lines
1611 // above are for, stated so a fifth arm added later cannot quietly
1612 // land on a cell that is already taken.
1613 let mut cells: Vec<_> = [
1614 ProxySide::ClientToProxy,
1615 ProxySide::ProxyToRelay,
1616 ProxySide::RelayToProxy,
1617 ProxySide::ProxyToClient,
1618 ]
1619 .into_iter()
1620 .map(|side| {
1621 let (leg, direction) = split(side);
1622 (leg_index(leg), direction.index())
1623 })
1624 .collect();
1625 cells.sort_unstable();
1626 cells.dedup();
1627 assert_eq!(cells.len(), 4, "two sides charge the same cell");
1628 }
1629
1630 /// A proxy recorder and one session reporting into it, on the class list
1631 /// `names`.
1632 fn attached(names: &[&str]) -> (Arc<ProxyRecorder>, ShapeRecorder) {
1633 let proxy = Arc::new(ProxyRecorder::new());
1634 let recorder = ShapeRecorder::attached(Some(&profile(names)), Arc::clone(&proxy));
1635 (proxy, recorder)
1636 }
1637
1638 /// A bidirectional run over the class `video`, in which all four
1639 /// crossings carry a different number of bytes and a different number of
1640 /// objects.
1641 ///
1642 /// Uplink: two objects (40 and 2) and a 7-byte stream header arrive from
1643 /// the client; one 40-byte object is dropped by policy, so a 30-byte
1644 /// object and the header go out to the relay. Downlink: four objects
1645 /// (100, 200, 300, 400) and a 5-byte header arrive from the relay, one
1646 /// expires and takes its stream with it, and three objects and the
1647 /// header go out to the client.
1648 ///
1649 /// Every figure differs from every other, in both directions and on both
1650 /// legs, so no swap of two cells and no collapse of either axis can
1651 /// cancel out.
1652 fn bidirectional_run(rec: &ShapeRecorder) {
1653 let video = Class::Rule(0);
1654
1655 // Read from the client.
1656 rec.note_object_seen(ProxySide::ClientToProxy, 40);
1657 rec.note_object_seen(ProxySide::ClientToProxy, 2);
1658 rec.note_unshapeable_seen(ProxySide::ClientToProxy, 7);
1659 rec.note_mixed_class_stream(ProxySide::ClientToProxy);
1660 rec.note_dropped(video, 12);
1661 // Written to the relay. Still the arrival side: turning it around is
1662 // the recorder's job.
1663 rec.note_delivered(ProxySide::ClientToProxy, video, 30);
1664 rec.note_delivered(ProxySide::ClientToProxy, Class::Unshapeable, 7);
1665
1666 // Read from the relay.
1667 rec.note_object_seen(ProxySide::RelayToProxy, 100);
1668 rec.note_object_seen(ProxySide::RelayToProxy, 200);
1669 rec.note_object_seen(ProxySide::RelayToProxy, 300);
1670 rec.note_object_seen(ProxySide::RelayToProxy, 400);
1671 rec.note_unshapeable_seen(ProxySide::RelayToProxy, 5);
1672 rec.note_expired(ProxySide::RelayToProxy);
1673 rec.note_stream_reset_by_shaping(ProxySide::RelayToProxy);
1674 rec.note_dropped(video, 400);
1675 // Written to the client.
1676 rec.note_delivered(ProxySide::RelayToProxy, video, 100);
1677 rec.note_delivered(ProxySide::RelayToProxy, video, 200);
1678 rec.note_delivered(ProxySide::RelayToProxy, video, 300);
1679 rec.note_delivered(ProxySide::RelayToProxy, Class::Unshapeable, 5);
1680 }
1681
1682 /// The four crossings of a proxy land in four different cells, and each
1683 /// cell reports its own crossing.
1684 ///
1685 /// This is the attribution claim, and it is the one no arithmetic
1686 /// checks. A proxy holds two connections and a byte crosses both — read
1687 /// on one leg, written on the other — so leg and direction are
1688 /// genuinely two axes and the surface has four cells to fill. Two
1689 /// mistakes fill them wrongly while leaving every total intact, and
1690 /// either one hides the other: charging every unit to the direction the
1691 /// first arm happened to name, and deriving the leg from the arriving
1692 /// side so that both legs report the same row. Comparing all four rows
1693 /// **whole** is what separates them; comparing one, or comparing a sum,
1694 /// separates neither.
1695 ///
1696 /// The two upstream figures are what the shape exists for. The uplink
1697 /// pair, 49 bytes in against 37 out, is the shaper's own retention on
1698 /// that direction — bytes it read from the client and did not write to
1699 /// the relay — and it is a number that can only be non-zero because the
1700 /// two cells are measured at two different crossings.
1701 ///
1702 /// The three event figures move on the arrival cell only, and the zeros
1703 /// for them on the two departure cells are asserted rather than elided:
1704 /// an expiry is a decision over traffic that came in and never went out,
1705 /// so a departure cell that reported one would be claiming a crossing
1706 /// that did not happen.
1707 ///
1708 /// *Ablation, recorded:* collapse the leg axis — have `leg_index`
1709 /// answer `0` for both legs, so every figure lands on the client row.
1710 /// The first compare reddens with `left: DirectionStats { objects_seen:
1711 /// 3, bytes_shaped: 86, objects_expired: 0, streams_reset_by_shaping: 0,
1712 /// streams_with_mixed_classes: 1 }` against `right: DirectionStats {
1713 /// objects_seen: 2, bytes_shaped: 49, objects_expired: 0,
1714 /// streams_reset_by_shaping: 0, streams_with_mixed_classes: 1 }` — what
1715 /// was written to the relay piled on top of what was read from the
1716 /// client.
1717 ///
1718 /// *Ablation, recorded:* derive the leg from the arriving side instead —
1719 /// name `Leg::Client` in both upstream arms of `split`, which is what
1720 /// that derivation amounts to over the two sides a hook site can hold.
1721 /// The upstream downlink compare reddens with `left: DirectionStats {
1722 /// objects_seen: 3, bytes_shaped: 605, objects_expired: 0,
1723 /// streams_reset_by_shaping: 0, streams_with_mixed_classes: 0 }` against
1724 /// `right: DirectionStats { objects_seen: 4, bytes_shaped: 1005,
1725 /// objects_expired: 1, streams_reset_by_shaping: 1,
1726 /// streams_with_mixed_classes: 0 }` — the two downlink cells swapped,
1727 /// so what this proxy read from the relay is reported as what it sent to
1728 /// the client. Every total still adds up, and every sum still passes.
1729 ///
1730 /// *Ablation, recorded:* collapse the direction axis instead — name
1731 /// `Direction::Uplink` in both downlink arms of `split`. The first
1732 /// compare reddens with `left: DirectionStats { objects_seen: 5,
1733 /// bytes_shaped: 654, objects_expired: 0, streams_reset_by_shaping: 0,
1734 /// streams_with_mixed_classes: 1 }` against `right: DirectionStats {
1735 /// objects_seen: 2, bytes_shaped: 49, objects_expired: 0,
1736 /// streams_reset_by_shaping: 0, streams_with_mixed_classes: 1 }` — both
1737 /// directions piled into row 0.
1738 ///
1739 /// *Ablation, recorded:* keep both axes and drop the turn-around — have
1740 /// `ProxyRecorder::departure` answer `self.arrival(side)`. The first
1741 /// compare reddens with `left: DirectionStats { objects_seen: 3,
1742 /// bytes_shaped: 86, objects_expired: 0, streams_reset_by_shaping: 0,
1743 /// streams_with_mixed_classes: 1 }` against the same right-hand side as
1744 /// the leg collapse above — the two mutations are different mistakes
1745 /// with one observable, because a proxy that cannot tell its legs apart
1746 /// and a proxy that never turns a release around both file a write where
1747 /// the read went.
1748 #[test]
1749 fn each_crossing_charges_its_own_cell_of_the_two_by_two() {
1750 let (proxy, rec) = attached(&["video"]);
1751 bidirectional_run(&rec);
1752 let stats = proxy.snapshot();
1753
1754 assert_eq!(
1755 *stats.leg(Leg::Client).uplink(),
1756 DirectionStats {
1757 objects_seen: 2,
1758 bytes_shaped: 49,
1759 objects_expired: 0,
1760 streams_reset_by_shaping: 0,
1761 streams_with_mixed_classes: 1,
1762 },
1763 "what this proxy read from the client"
1764 );
1765 assert_eq!(
1766 *stats.leg(Leg::Upstream).uplink(),
1767 DirectionStats {
1768 objects_seen: 1,
1769 bytes_shaped: 37,
1770 objects_expired: 0,
1771 streams_reset_by_shaping: 0,
1772 streams_with_mixed_classes: 0,
1773 },
1774 "what it wrote to the relay — 12 bytes short of what it read, \
1775 because a policy dropped an object"
1776 );
1777 assert_eq!(
1778 *stats.leg(Leg::Upstream).downlink(),
1779 DirectionStats {
1780 objects_seen: 4,
1781 bytes_shaped: 1005,
1782 objects_expired: 1,
1783 streams_reset_by_shaping: 1,
1784 streams_with_mixed_classes: 0,
1785 },
1786 "what it read from the relay"
1787 );
1788 assert_eq!(
1789 *stats.leg(Leg::Client).downlink(),
1790 DirectionStats {
1791 objects_seen: 3,
1792 bytes_shaped: 605,
1793 objects_expired: 0,
1794 streams_reset_by_shaping: 0,
1795 streams_with_mixed_classes: 0,
1796 },
1797 "what it wrote to the client"
1798 );
1799 }
1800
1801 /// A stream header crosses both legs as bytes and as no object at all.
1802 ///
1803 /// `objects_seen` is the classifier's count, and the classifier runs
1804 /// where traffic arrives — so on the way in a header is charged to
1805 /// `bytes_shaped` and not to `objects_seen`. The departure cell has to
1806 /// hold the same line or the field means two different things in two
1807 /// cells of the same array, and the difference between the two would
1808 /// read as objects this proxy invented.
1809 ///
1810 /// *Ablation, recorded:* drop the `class != Class::Unshapeable` guard in
1811 /// `note_delivered`, so every released unit counts as an object. The
1812 /// compare reddens with `left: 2` against `right: 1` — the header
1813 /// counted as an object on the way out and not on the way in.
1814 #[test]
1815 fn a_released_header_is_bytes_on_the_departure_cell_and_not_an_object() {
1816 let (proxy, rec) = attached(&["video"]);
1817 rec.note_unshapeable_seen(ProxySide::ClientToProxy, 7);
1818 rec.note_object_seen(ProxySide::ClientToProxy, 40);
1819 rec.note_delivered(ProxySide::ClientToProxy, Class::Unshapeable, 7);
1820 rec.note_delivered(ProxySide::ClientToProxy, Class::Rule(0), 40);
1821
1822 let stats = proxy.snapshot();
1823 let out = stats.leg(Leg::Upstream).uplink();
1824 assert_eq!(out.bytes_shaped, 47, "both units crossed, and both are bytes");
1825 assert_eq!(out.objects_seen, 1, "a header is not an object on the way out either");
1826 assert_eq!(
1827 stats.unshapeable.objects_delivered, 1,
1828 "the unshapeable row still counts the unit it released"
1829 );
1830 }
1831
1832 /// The flat rollup counts a byte where it **entered**, once.
1833 ///
1834 /// Derived rather than accumulated, so it cannot drift from the cells
1835 /// beside it — but *which* cells it is derived from is a real choice and
1836 /// the wrong one is invisible. A byte crosses two legs, so the obvious
1837 /// summation, all four cells, reports every byte twice and looks
1838 /// entirely plausible: it is monotone, it is proportional to the
1839 /// traffic, and it is about double.
1840 ///
1841 /// The compare is over the **whole struct** rather than the fields this
1842 /// test has an opinion about, so a field added to [`SessionStats`] later
1843 /// and derived from the wrong cells cannot slip past it.
1844 ///
1845 /// *Ablation, re-run:* sum all four cells in `ProxyRecorder::snapshot`
1846 /// instead of the two arrival cells. The compare reddens with `left:
1847 /// SessionStats { objects_seen: 10, objects_dropped: 2, objects_expired:
1848 /// 1, streams_reset: 1, bytes_shaped: 1696 }` against `right:
1849 /// SessionStats { objects_seen: 6, .., bytes_shaped: 1054 }` — every byte
1850 /// counted twice, and nothing else out of place.
1851 #[test]
1852 fn the_flat_rollup_counts_a_byte_where_it_entered() {
1853 let (proxy, rec) = attached(&["video"]);
1854 bidirectional_run(&rec);
1855
1856 assert_eq!(
1857 proxy.snapshot().sessions,
1858 SessionStats {
1859 objects_seen: 6,
1860 objects_dropped: 2,
1861 objects_expired: 1,
1862 streams_reset: 1,
1863 bytes_shaped: 1054,
1864 },
1865 "49 + 1005 bytes in, not 49 + 37 + 1005 + 605 crossings"
1866 );
1867 }
1868
1869 /// A session reports both to itself and to its proxy, and the two agree
1870 /// about everything they both hold.
1871 ///
1872 /// The reason the forwarding lives inside the recorder rather than at a
1873 /// second set of call sites: one call charges both or neither, so a
1874 /// producer cannot be added to a session figure and forgotten beside it.
1875 /// The comparison is over the figures the two types share — the flat
1876 /// session totals and the class rows — because those are the ones a
1877 /// reader would put side by side and expect to match.
1878 ///
1879 /// *Ablation, recorded:* drop the proxy forward from
1880 /// `note_object_seen`. The compare reddens with `left: 0` against
1881 /// `right: 6` — the session still reports six objects and the proxy
1882 /// reports none of them.
1883 #[test]
1884 fn one_session_charges_its_own_rows_and_its_proxys_together() {
1885 let (proxy, rec) = attached(&["video"]);
1886 bidirectional_run(&rec);
1887
1888 let session = rec.snapshot();
1889 let aggregate = proxy.snapshot();
1890 assert_eq!(aggregate.sessions.objects_seen, session.objects_seen);
1891 assert_eq!(aggregate.sessions.bytes_shaped, session.bytes_shaped);
1892 assert_eq!(aggregate.sessions.objects_expired, session.objects_expired);
1893 assert_eq!(aggregate.sessions.streams_reset, session.streams_reset_by_shaping);
1894 assert_eq!(aggregate.classes, session.classes);
1895 assert_eq!(aggregate.unshapeable, session.unshapeable);
1896 }
1897
1898 /// A session with no proxy charges no proxy.
1899 ///
1900 /// `ShapeRecorder::for_profile` is what a session constructed directly
1901 /// gets, and such a session belongs to no proxy: it never went through
1902 /// an accept loop, so no [`ProxyStats`] should claim its traffic. The
1903 /// proxy recorder here is built and driven past — the run below charges
1904 /// a recorder that was never attached to it — and it must still snapshot
1905 /// as default.
1906 #[test]
1907 fn a_session_with_no_proxy_leaves_the_aggregate_alone() {
1908 let proxy = Arc::new(ProxyRecorder::new());
1909 let rec = ShapeRecorder::for_profile(Some(&profile(&["video"])));
1910 bidirectional_run(&rec);
1911
1912 assert_ne!(rec.snapshot(), ShapeStats::default(), "the session recorded its own run");
1913 assert_eq!(proxy.snapshot(), ProxyStats::default());
1914 }
1915
1916 /// Class rows are sized by the first shaped session, and a later session
1917 /// on a different class list charges the default row rather than a row
1918 /// named for somebody else's rule.
1919 ///
1920 /// A `Class::Rule(index)` is an index into the class list of the
1921 /// scheduler that produced it. A live `set_shape` that installs a
1922 /// different list reaches sessions accepted afterwards, so the second
1923 /// session here is that session — and if it charged by index anyway,
1924 /// every figure it produced would be reported under the first profile's
1925 /// names, monotone and plausible and wrong. The default row is where a
1926 /// unit whose row cannot be named belongs, which is the answer
1927 /// `ShapeRecorder::row` already gives an index it cannot place.
1928 ///
1929 /// *Ablation, recorded:* have `ProxyRecorder::row` ignore `sized`. The
1930 /// compare reddens with `left: 500` against `right: 100` — the second
1931 /// session's bytes filed under `video`, a class it never had.
1932 #[test]
1933 fn a_session_on_another_class_list_charges_the_default_row() {
1934 let proxy = Arc::new(ProxyRecorder::new());
1935 let first = ShapeRecorder::attached(Some(&profile(&["video"])), Arc::clone(&proxy));
1936 let second = ShapeRecorder::attached(Some(&profile(&["screen"])), Arc::clone(&proxy));
1937
1938 first.note_delivered(ProxySide::ClientToProxy, Class::Rule(0), 100);
1939 second.note_delivered(ProxySide::ClientToProxy, Class::Rule(0), 400);
1940
1941 let stats = proxy.snapshot();
1942 assert_eq!(
1943 stats.classes.iter().map(|c| c.name.as_str()).collect::<Vec<_>>(),
1944 vec!["video"],
1945 "the rows are the first shaped session's, and are never resized"
1946 );
1947 assert_eq!(stats.classes[0].bytes_delivered, 100, "only the matching session's bytes");
1948 assert_eq!(stats.default_class.bytes_delivered, 400, "the mismatched session's go here");
1949 }
1950
1951 /// An unshaped session does not size the proxy's class rows.
1952 ///
1953 /// It has no classes to install, and installing its empty list would
1954 /// leave every shaped session that arrived afterwards mismatched
1955 /// forever — charging the default row for the life of the proxy because
1956 /// the first connection happened to be one nobody configured shaping
1957 /// for. Nothing is lost by skipping it: an unshaped session's writers
1958 /// are all behind a configured profile, so it charges nothing at all.
1959 ///
1960 /// *Ablation, recorded:* let the `None` arm of `ShapeRecorder::attached`
1961 /// adopt an empty class list too. The compare reddens with `left: []`
1962 /// against `right: ["video"]` — the proxy sized itself from a session
1963 /// that had no classes, and the shaped session behind it has no row.
1964 #[test]
1965 fn an_unshaped_session_does_not_size_the_proxys_class_rows() {
1966 let proxy = Arc::new(ProxyRecorder::new());
1967 let _unshaped = ShapeRecorder::attached(None, Arc::clone(&proxy));
1968 let shaped = ShapeRecorder::attached(Some(&profile(&["video"])), Arc::clone(&proxy));
1969
1970 shaped.note_delivered(ProxySide::ClientToProxy, Class::Rule(0), 64);
1971
1972 let stats = proxy.snapshot();
1973 assert_eq!(
1974 stats.classes.iter().map(|c| c.name.as_str()).collect::<Vec<_>>(),
1975 vec!["video"]
1976 );
1977 assert_eq!(stats.classes[0].bytes_delivered, 64);
1978 }
1979
1980 /// An empty class list does not size the proxy's rows either.
1981 ///
1982 /// The sibling of the test above, and the case that one does not cover.
1983 /// The guard beside it keys on there being no profile; this one is about
1984 /// a list that is present and holds nothing.
1985 ///
1986 /// **It reaches [`ProxyRecorder::adopt_classes`] directly, and that is
1987 /// the point of this version of the row.** The only thing that could
1988 /// carry an empty list there is a `ShapeProfile` with no classes, and
1989 /// [`ShapeProfile::try_new`] refuses that outright as
1990 /// `ShapeError::NoClasses`, so no public path reaches this branch. The
1991 /// guard is kept anyway, and so is this row: what it prevents is
1992 /// proxy-wide, silent and unrecoverable without a restart, and the check
1993 /// is a comparison that is being made in any case.
1994 ///
1995 /// An empty row set matches no later class list, so the proxy's rows
1996 /// would stay empty and every classed session accepted for the rest of
1997 /// its life would charge the default row — every figure right, every
1998 /// label gone, with nothing anywhere reporting it.
1999 ///
2000 /// The 100 bytes are asserted on the row rather than on the total,
2001 /// because a total is exactly what survives the bug: the bytes are never
2002 /// lost, only misfiled.
2003 ///
2004 /// *Ablation, run:* drop the `names.is_empty()` guard from
2005 /// `ProxyRecorder::adopt_classes`. The first assertion reddens with
2006 ///
2007 /// ```text
2008 /// thread 'shape::stats::tests::an_empty_class_list_does_not_size_the_proxys_class_rows'
2009 /// (63164) panicked at crates\moqtap-proxy\src\shape\stats.rs:
2010 /// a list with no rows in it names no row, so there is nothing for it to
2011 /// charge and nothing is lost by declining it
2012 /// ```
2013 ///
2014 /// The three assertions under it are not reached, so that is the whole of
2015 /// what the mutation was seen to produce; the misfiling they describe is
2016 /// what the empty row set leaves behind once the sizing has been lost.
2017 /// A probe run without the guard prints `classes=[] default_bytes=100`:
2018 /// every byte counted, every label gone.
2019 #[test]
2020 fn an_empty_class_list_does_not_size_the_proxys_class_rows() {
2021 let proxy = Arc::new(ProxyRecorder::new());
2022 assert!(
2023 !proxy.adopt_classes(&[]),
2024 "a list with no rows in it names no row, so there is nothing for it to charge and \
2025 nothing is lost by declining it"
2026 );
2027 let classed = ShapeRecorder::attached(Some(&profile(&["video"])), Arc::clone(&proxy));
2028
2029 classed.note_delivered(ProxySide::ClientToProxy, Class::Rule(0), 100);
2030
2031 let stats = proxy.snapshot();
2032 assert_eq!(
2033 stats.classes.iter().map(|c| c.name.as_str()).collect::<Vec<_>>(),
2034 vec!["video"],
2035 "the classed session's rows are the proxy's rows: a list with nothing in it must not \
2036 install itself"
2037 );
2038 assert_eq!(stats.classes[0].bytes_delivered, 100, "and its bytes are charged by name");
2039 assert_eq!(
2040 stats.default_class.bytes_delivered, 0,
2041 "not to a row named for nobody, which is where they land once an empty list has won \
2042 the sizing"
2043 );
2044 }
2045
2046 /// A reset zeroes every counter and keeps the class rows, and traffic
2047 /// after it accumulates from zero.
2048 ///
2049 /// The rows survive because they are what a `Class::Rule(index)` means:
2050 /// a reset that dropped them would let the next session install a
2051 /// different class list, which is the relabelling the sizing rule exists
2052 /// to prevent. So the observable is a class row that is present, still
2053 /// named, and empty.
2054 ///
2055 /// *Ablation, recorded:* have `ProxyRecorder::reset` skip the class rows
2056 /// — reset the four cells and stop. The compare reddens with `left:
2057 /// 630` against `right: 0`: the legs read as a proxy that has done
2058 /// nothing while the class rows still hold the whole run.
2059 #[test]
2060 fn a_reset_zeroes_the_counters_and_keeps_the_rows() {
2061 let (proxy, rec) = attached(&["video"]);
2062 bidirectional_run(&rec);
2063 assert_ne!(proxy.snapshot(), ProxyStats::default(), "there was something to clear");
2064
2065 proxy.reset();
2066 let cleared = proxy.snapshot();
2067 assert_eq!(cleared.classes[0].bytes_delivered, 0);
2068 assert_eq!(cleared.classes[0].name, "video", "the row is still the row it was");
2069 assert_eq!(
2070 cleared,
2071 ProxyStats {
2072 classes: vec![ClassStats { name: "video".to_string(), ..ClassStats::default() }],
2073 ..ProxyStats::default()
2074 },
2075 "nothing but the row's name survives a reset"
2076 );
2077
2078 rec.note_object_seen(ProxySide::ClientToProxy, 11);
2079 assert_eq!(proxy.snapshot().sessions.bytes_shaped, 11, "counting resumes from zero");
2080 }
2081}