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}