Skip to main content

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