Skip to main content

moqtap_proxy/
event.rs

1//! Proxy event types emitted by the inline parser.
2
3use std::net::SocketAddr;
4use std::time::{Duration, Instant};
5
6use bytes::Bytes;
7use moqtap_codec::dispatch::{
8    AnyControlMessage, AnyDatagramHeader, AnyFetchHeader, AnyObjectHeader, AnySubgroupHeader,
9};
10use moqtap_codec::version::DraftVersion;
11
12use crate::capability::{ActionKind, Refusal, Site};
13use crate::shape::StreamKey;
14use crate::types::{Leg, ObjectMeta};
15
16pub use crate::types::ProxySide;
17
18/// Unique session identifier (monotonic counter assigned by the proxy).
19#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
20pub struct SessionId(pub u64);
21
22/// The kind of data stream header parsed from a unidirectional stream.
23#[derive(Debug, Clone)]
24pub enum DataStreamHeaderKind {
25    /// Subgroup stream header.
26    Subgroup(AnySubgroupHeader),
27    /// Fetch response stream header.
28    Fetch(AnyFetchHeader),
29}
30
31/// Events emitted by the proxy during stream forwarding.
32///
33/// Marked `#[non_exhaustive]`: observers must carry a catch-all arm, so
34/// that adding an event is not a breaking change.
35///
36/// # Asserting on events — destructure, do not compare
37///
38/// `ProxyEvent` derives **`Debug` and `Clone` only**. It carries decoded
39/// codec types ([`AnyControlMessage`], [`AnyDatagramHeader`], …) that do
40/// not implement `PartialEq`, so there is no `assert_eq!(event,
41/// ProxyEvent::Something { .. })` to write and there will not be one:
42/// deriving `PartialEq` here would require it on every draft's message
43/// enum.
44///
45/// Every assertion in this workspace therefore has the same two-step
46/// shape — **count with `matches!`, then destructure and compare the
47/// payload**:
48///
49/// ```
50/// use moqtap_proxy::event::{Effect, ProxyEvent};
51///
52/// fn exactly_one_replacement(events: &[ProxyEvent]) {
53///     // 1. count the variant with `matches!`
54///     let applied: Vec<&ProxyEvent> = events
55///         .iter()
56///         .filter(|e| matches!(e, ProxyEvent::ActionApplied { .. }))
57///         .collect();
58///     assert_eq!(applied.len(), 1);
59///
60///     // 2. destructure, then compare the payload with `assert_eq!`
61///     let ProxyEvent::ActionApplied { effect, .. } = applied[0] else {
62///         unreachable!("filtered above")
63///     };
64///     assert_eq!(*effect, Effect::Replaced { bytes: 4 });
65/// }
66/// # let _ = exactly_one_replacement;
67/// ```
68///
69/// Step 2 works because the *payload* types added in 0.4.0 — [`Effect`],
70/// [`ImpairmentKind`], [`Refusal`], [`Site`] and [`ActionKind`] — do
71/// derive `PartialEq` and `Eq`. The boundary is exactly at the event: the
72/// event is destructured, what comes out of it is compared.
73///
74/// `ProxyEvent` and every one of those payload enums is
75/// `#[non_exhaustive]`, so a `match` over any of them from outside this
76/// crate needs a catch-all arm; `matches!` supplies one for free, which
77/// is the other reason it is the recommended spelling.
78#[derive(Debug, Clone)]
79#[non_exhaustive]
80pub enum ProxyEvent {
81    /// A new client connected and a session was created.
82    SessionStarted {
83        /// The session identifier.
84        session_id: SessionId,
85        /// The client's remote address.
86        client_addr: SocketAddr,
87        /// The transport the client chose via ALPN — e.g. `"QUIC"` or
88        /// `"WebTransport"`. Observers use this to label per-client
89        /// sessions; the proxy itself accepts either simultaneously.
90        client_transport: String,
91    },
92
93    /// A setup message (CLIENT_SETUP or SERVER_SETUP) was observed.
94    SetupMessage {
95        /// The session identifier.
96        session_id: SessionId,
97        /// Which side sent the message.
98        side: ProxySide,
99        /// QUIC stream ID the message arrived on, on `side`'s connection; `None` if unknown.
100        stream_id: Option<u64>,
101        /// The decoded setup message.
102        message: AnyControlMessage,
103    },
104
105    /// A control message was parsed from the forwarded byte stream.
106    ControlMessage {
107        /// The session identifier.
108        session_id: SessionId,
109        /// Which side sent the message.
110        side: ProxySide,
111        /// QUIC stream ID the message arrived on, on `side`'s connection; `None` if unknown.
112        stream_id: Option<u64>,
113        /// The decoded control message.
114        message: AnyControlMessage,
115    },
116
117    /// A data stream header was parsed from a unidirectional stream.
118    DataStreamHeader {
119        /// The session identifier.
120        session_id: SessionId,
121        /// Which side opened the stream.
122        side: ProxySide,
123        /// QUIC stream ID on `side`'s connection.
124        stream_id: u64,
125        /// The parsed header.
126        header: DataStreamHeaderKind,
127        /// The header's wire bytes, from the stream type through its last
128        /// field. No transport framing.
129        raw: Bytes,
130    },
131
132    /// An object header was parsed on a data stream.
133    ///
134    /// Never emitted: `AnyObjectHeader` has no variants past draft-13, so
135    /// this could only ever report objects on the oldest drafts, and even
136    /// there it reported a header without consuming the payload behind it.
137    /// [`ProxyEvent::Object`] replaces it and covers every draft.
138    #[deprecated(since = "0.3.0", note = "superseded by ProxyEvent::Object")]
139    ObjectHeader {
140        /// The session identifier.
141        session_id: SessionId,
142        /// Which side sent the object.
143        side: ProxySide,
144        /// The parsed object header.
145        header: AnyObjectHeader,
146    },
147
148    /// A complete object was framed on a data stream.
149    ///
150    /// Emitted once per object, in stream order, on every draft.
151    /// Objects the framer could not address individually — one larger than
152    /// its buffer cap, or a stream it stopped parsing — produce no event;
153    /// their bytes are still forwarded unchanged.
154    Object {
155        /// The session identifier.
156        session_id: SessionId,
157        /// Which side sent the object.
158        side: ProxySide,
159        /// QUIC stream ID on `side`'s connection.
160        stream_id: u64,
161        /// The object's identity and framing, without its payload.
162        meta: ObjectMeta,
163    },
164
165    /// A datagram arrived and its header was parsed.
166    ///
167    /// An **observation of what came in**, emitted where the decode happens
168    /// — before any hook is consulted and before the datagram is handed to
169    /// the far transport. It is deliberately not a delivery receipt, and
170    /// calling it one would be wrong in two directions at once: a hook that
171    /// drops or replaces the datagram still produces this event, and the
172    /// forward itself can be refused, which is reported separately as
173    /// [`ImpairmentKind::DatagramNotSent`] or, when somebody acted on it,
174    /// as [`ProxyEvent::ActionFailed`].
175    ///
176    /// It sits with [`ProxyEvent::Object`] and
177    /// [`ProxyEvent::ControlMessage`] rather than with the action events:
178    /// all three say a unit was seen and understood, and none of them says
179    /// where it ended up. Withholding it until after the send would make an
180    /// undecodable-or-dropped datagram invisible, which is exactly the case
181    /// an observer is usually watching for.
182    ///
183    /// Emitted only when an observer is attached — the decode is skipped
184    /// entirely otherwise — and only when the header actually decoded. A
185    /// datagram whose header this crate cannot read produces no event and is
186    /// still forwarded.
187    Datagram {
188        /// The session identifier.
189        session_id: SessionId,
190        /// Which side sent the datagram.
191        side: ProxySide,
192        /// The parsed datagram header.
193        header: AnyDatagramHeader,
194        /// The datagram's bytes before its payload (all of them for a status
195        /// datagram). No transport framing.
196        header_raw: Bytes,
197        /// Size of the datagram payload in bytes.
198        payload_len: usize,
199    },
200
201    /// A bidirectional stream was opened or accepted.
202    BiStreamOpened {
203        /// The session identifier.
204        session_id: SessionId,
205        /// Which side opened the stream.
206        side: ProxySide,
207        /// QUIC stream ID on `side`'s connection.
208        stream_id: u64,
209    },
210
211    /// A unidirectional stream was opened or accepted.
212    UniStreamOpened {
213        /// The session identifier.
214        session_id: SessionId,
215        /// Which side opened the stream.
216        side: ProxySide,
217        /// QUIC stream ID on `side`'s connection.
218        stream_id: u64,
219    },
220
221    /// Inline parse failed (non-fatal — bytes are still forwarded).
222    ParseError {
223        /// The session identifier.
224        session_id: SessionId,
225        /// Which side the error occurred on.
226        side: ProxySide,
227        /// Description of the parse error.
228        error: String,
229    },
230
231    /// A stream direction ended cleanly — the peer sent a FIN and every
232    /// byte it wrote was forwarded.
233    ///
234    /// An abnormal end is reported as [`ProxyEvent::StreamReset`]
235    /// instead, never as this event.
236    StreamClosed {
237        /// The session identifier.
238        session_id: SessionId,
239        /// Which side closed.
240        side: ProxySide,
241        /// QUIC stream ID on `side`'s connection.
242        stream_id: u64,
243    },
244
245    /// A peer tore a stream down abnormally — either `RESET_STREAM` from
246    /// the sender or `STOP_SENDING` from the receiver.
247    ///
248    /// This reports the *observation*. The proxy mirrors the teardown
249    /// onto the opposite stream with the same application error code;
250    /// when that stream has already gone away the mirror is a no-op, and
251    /// the event is still emitted so the teardown is never invisible.
252    ///
253    /// Distinct from [`ProxyEvent::StreamClosed`], which reports an
254    /// orderly FIN. An observer that sees this knows the stream was
255    /// abandoned and any data on it may be truncated.
256    ///
257    /// Not emitted for streams the proxy itself tears down at session
258    /// shutdown — those end with a FIN.
259    StreamReset {
260        /// The session identifier.
261        session_id: SessionId,
262        /// The side the teardown was observed on. For a `RESET_STREAM`
263        /// this is the ingress side the bytes were arriving on; for a
264        /// `STOP_SENDING` it is the egress side they were leaving on.
265        side: ProxySide,
266        /// QUIC stream ID on `side`'s connection.
267        stream_id: u64,
268        /// The peer's application error code, forwarded verbatim.
269        code: u64,
270    },
271
272    /// The session ended.
273    SessionEnded {
274        /// The session identifier.
275        session_id: SessionId,
276        /// Reason for session termination.
277        reason: String,
278    },
279
280    // ── 0.4.0: the action engine ───────────────────────────────────────
281    /// An action was executed and the wire changed.
282    ///
283    /// Emitted once the engine has **admitted** the action and produced the
284    /// bytes or the plan for it: every guard has run, every refusal has
285    /// already been taken, and nothing between here and the wire can decline
286    /// it on this proxy's behalf. `effect` is what was decided, down to the
287    /// byte count.
288    ///
289    /// # It is not a delivery receipt, and one case makes that visible
290    ///
291    /// The engine decides; the forwarding task then hands the result to the
292    /// transport. That transport can still say no — a replacement datagram
293    /// above the path MTU is the case that exists — and when it does, this
294    /// event has already been emitted and is followed by
295    /// [`ProxyEvent::ActionFailed`] naming the same site and action. **Both
296    /// events are emitted for that attempt**, in that order, and the pair is
297    /// the whole truth: the action was admitted, and it did not reach the
298    /// peer.
299    ///
300    /// Folding the two into one report was rejected twice over. Withholding
301    /// this event until the write returned would mean an action that was
302    /// admitted, queued behind a `Delay`, and lost at teardown reported
303    /// nothing at all — and it is precisely the admitted-then-lost case that
304    /// [`ImpairmentKind::QueuedBytesAtTeardown`] is paired against. Reporting
305    /// only the failure would lose which action was taken, since a refusal
306    /// and a rejection name different things.
307    ///
308    /// So the reading is: this event means *the engine did it*. An observer
309    /// that needs *the peer got it* must also watch for `ActionFailed`, and
310    /// for the impairments that report queued bytes.
311    ///
312    /// A proxy-initiated reset appears here, **not** as
313    /// [`ProxyEvent::StreamReset`], which continues to mean an observed
314    /// peer teardown.
315    ///
316    /// # Cardinality under `Delay` and `Hold`
317    ///
318    /// A deferred action emits **two** `ActionApplied` events, not one,
319    /// and they are distinguishable by `action`:
320    ///
321    /// 1. at the decision, `{ action: Delay | Hold, effect: Queued {
322    ///    release_at } }` — the modifier was accepted and the unit is in
323    ///    the queue;
324    /// 2. at the release, `{ action: <the inner kind>, effect: <what the
325    ///    inner action did> }` — e.g. `{ action: Replace, effect:
326    ///    Replaced { bytes } }`.
327    ///
328    /// One event would force a choice between reporting the delay and reporting
329    /// the effect, and a queued unit that is later lost at teardown would have
330    /// reported a `Replaced` that never happened. Two events keep *the engine
331    /// accepted this* and "the wire changed" separately falsifiable, which is
332    /// what [`ImpairmentKind::QueuedBytesAtTeardown`] is paired against.
333    ///
334    /// So the count to assert is *exactly one `ActionApplied` per
335    /// (attempt, phase)*: one for a non-deferred action, two for a
336    /// deferred one.
337    ActionApplied {
338        /// The session identifier.
339        session_id: SessionId,
340        /// The side the unit arrived on.
341        side: ProxySide,
342        /// The source stream, when there is one.
343        stream_id: Option<u64>,
344        /// Where the decision was taken.
345        site: Site,
346        /// What was executed.
347        action: ActionKind,
348        /// What actually happened.
349        effect: Effect,
350    },
351
352    /// An action could not be executed. Emitted **per attempt**, and the
353    /// unit is forwarded unchanged.
354    ///
355    /// The engine declined *before* anything reached the transport: the
356    /// wire carries exactly what it would have carried with no hook at
357    /// all. This is the pre-admission half of the two failure reports —
358    /// [`ProxyEvent::ActionFailed`] is the post-admission one, where the
359    /// engine accepted the action and the transport rejected it.
360    ActionRefused {
361        /// The session identifier.
362        session_id: SessionId,
363        /// The side the unit arrived on.
364        side: ProxySide,
365        /// The source stream, when there is one.
366        stream_id: Option<u64>,
367        /// Where the decision was taken.
368        site: Site,
369        /// What was attempted.
370        action: ActionKind,
371        /// Why it was refused.
372        refusal: Refusal,
373    },
374
375    /// An action was admitted but the transport rejected it. The session
376    /// survives.
377    ///
378    /// Emitted **per failed attempt**. The unit is not forwarded — the
379    /// transport already declined it — and forwarding continues on
380    /// everything else.
381    ///
382    /// Reaching this means the capability table admitted the action and
383    /// the path disagreed; a replacement datagram above the path MTU is
384    /// the case that exists in 0.4.0. Neither
385    /// [`ProxyEvent::ActionApplied`] nor [`ProxyEvent::ActionRefused`]
386    /// can carry it on its own: the action was admitted, so refusing it
387    /// after the fact would be a lie, and it did not reach the peer, so
388    /// reporting only that it applied would be a bigger one.
389    ///
390    /// It is **paired with** `ActionApplied`, not exclusive of it. The
391    /// engine admits, emits `ActionApplied`, and hands the bytes on; the
392    /// transport then declines them and this follows. An attempt that
393    /// reaches here therefore contributes two events, in that order. It is
394    /// exclusive of [`ProxyEvent::ActionRefused`], which is the other
395    /// half of the same split: a refused unit is forwarded unchanged and a
396    /// transport failure on *those* bytes is
397    /// [`ImpairmentKind::DatagramNotSent`], because nobody's action was in
398    /// flight.
399    ///
400    /// Distinct from [`ImpairmentKind::DatagramNotSent`], which is the
401    /// same transport failure on a unit **nobody acted on** — there is no
402    /// `site` and no `action` to name there, and this variant requires
403    /// both.
404    ///
405    /// Connection-level errors (`ConnectionLost`, `Connection(_)`) are
406    /// **not** reported here: they still end the session.
407    ActionFailed {
408        /// The session identifier.
409        session_id: SessionId,
410        /// The side the unit arrived on.
411        side: ProxySide,
412        /// Where the decision was taken.
413        site: Site,
414        /// What was executed.
415        action: ActionKind,
416        /// The transport's error.
417        error: String,
418    },
419
420    /// Something reduced what the proxy can do, with no action involved.
421    ///
422    /// Every [`ImpairmentKind`] states its own emission cardinality;
423    /// read it there before asserting a count.
424    ///
425    /// # This is emitted after the thing it reports, never before
426    ///
427    /// Whatever the report describes has already happened by the time the
428    /// observer is called: the reset has been handed to the transport, the
429    /// datagram has been refused, the queue has been abandoned. Nothing in
430    /// this crate emits one of these on the way *into* an operation that
431    /// could still be declined.
432    ///
433    /// That ordering is what makes the event stream a record rather than an
434    /// intention. Reversed, an impairment raised before a step that a guard
435    /// then refuses is an observer told about a loss that did not occur —
436    /// and there is no later event that retracts it, because this variant
437    /// has no counterpart to [`ProxyEvent::ActionFailed`]. An observer may
438    /// therefore treat every one of these as a fact about the past.
439    ///
440    /// The one place the rule is visibly *not* the same is
441    /// [`ProxyEvent::ActionApplied`], which is emitted when the engine
442    /// admits an action and before the caller hands the bytes to the
443    /// transport; that pairing is covered in its own rustdoc and is why
444    /// `ActionFailed` exists.
445    Impairment {
446        /// The session identifier.
447        session_id: SessionId,
448        /// The side the reporting task was forwarding *from* — the
449        /// direction bytes were arriving on, not necessarily the direction
450        /// the impairment was felt in. Read `leg` for that.
451        side: ProxySide,
452        /// Which of the proxy's two connections the report is about, or
453        /// `None` when it is about neither.
454        ///
455        /// # Not derivable from `side`, which is why it is here
456        /// `side` names the direction the reporting task reads from, so it
457        /// answers *where did these bytes come from*. Most of what
458        /// [`ImpairmentKind`] reports is a failure to *write*: a queue that
459        /// could not be flushed, a datagram the far transport refused, a
460        /// destination stream that had to be reset. Those belong to the
461        /// **opposite** connection from the one the bytes arrived on, and a
462        /// reader who mapped `side` to a connection would attribute every one
463        /// of them to the wrong leg — silently, and in a way that looks
464        /// entirely plausible in a log.
465        ///
466        /// So the two are split, exactly as
467        /// [`ProxyStats`](crate::shape::ProxyStats) splits what a leg read
468        /// from what it wrote:
469        ///
470        /// * **the arriving connection** for
471        ///   [`ImpairmentKind::FramerBypass`],
472        ///   [`ImpairmentKind::ObjectNotAddressable`] and
473        ///   [`ImpairmentKind::ControlFrameNotDecodable`] — all three are a
474        ///   parser giving up on bytes that came *in*, and none of them
475        ///   says anything about what could be written;
476        /// * **the departing connection** for
477        ///   [`ImpairmentKind::EgressQueueFull`],
478        ///   [`ImpairmentKind::HoldClamped`],
479        ///   [`ImpairmentKind::QueuedBytesAtTeardown`],
480        ///   [`ImpairmentKind::DatagramNotSent`],
481        ///   [`ImpairmentKind::ControlStreamTruncated`],
482        ///   [`ImpairmentKind::ElideFixupLost`],
483        ///   [`ImpairmentKind::ShapeUnpacedObject`],
484        ///   [`ImpairmentKind::ClassChangedMidStream`] and
485        ///   [`ImpairmentKind::SerializeTargetUnknown`] — every one of them
486        ///   is about bytes the proxy was trying to place on the far side;
487        /// * **`None`** for
488        ///   [`ImpairmentKind::CoarseReleaseTimer`],
489        ///   [`ImpairmentKind::ShapeRuleUnmatchable`] and
490        ///   [`ImpairmentKind::ShapeBurstBelowUnit`]. These three compare a
491        ///   *profile* against the sizes it is asked to pace, or the
492        ///   process against its host. None of them is a property of a
493        ///   connection, and all three are equally true of both legs.
494        ///   `None` says that in the type instead of picking whichever leg
495        ///   the reporting task happened to be on.
496        ///
497        /// A caller that wants *which connection is unhealthy* reads this field
498        /// and skips the `None`s. A caller that wants *which direction was
499        /// being forwarded when this was noticed* reads `side`.
500        leg: Option<Leg>,
501        /// What happened.
502        kind: ImpairmentKind,
503    },
504
505    /// The egress shaper acted on a unit **as configured**.
506    ///
507    /// This is the product working, not an impairment: a configured drop is
508    /// the tool doing what it was told, where an
509    /// [`ImpairmentKind`] is the tool declining to. It is also the only
510    /// event that can carry a class label, because a class is a shaping
511    /// concept and nothing else in this enum has one.
512    ///
513    /// **Cardinality: once per stream per distinct `outcome`.** Running
514    /// totals live in [`ShapeStats`](crate::shape::ShapeStats) — a
515    /// per-object event would drown an observer at line rate, which is the
516    /// same reason [`ImpairmentKind::ObjectNotAddressable`] carries a total
517    /// instead of firing per object.
518    Shaped {
519        /// The session identifier.
520        session_id: SessionId,
521        /// The side the unit arrived on.
522        side: ProxySide,
523        /// Session-local identity.
524        key: StreamKey,
525        /// Transport stream id, for correlation with the other events in
526        /// this enum; `key` is what identifies.
527        stream_id: u64,
528        /// The class the unit resolved to, or an empty string for a unit
529        /// that matched no rule and for an outcome that is about the stream
530        /// rather than about a unit.
531        class: String,
532        /// What the shaper did.
533        outcome: ShapeOutcome,
534    },
535
536    /// The egress shaper discarded a **datagram** as configured.
537    ///
538    /// [`Self::Shaped`]'s datagram sibling, and separate from it because a
539    /// datagram belongs to no stream: `Shaped` carries a [`StreamKey`] and a
540    /// transport stream id, and there is nothing honest to put in either.
541    /// Merging the two by making those fields optional would put an
542    /// `Option` on the far commoner event to describe the rarer one.
543    ///
544    /// **Cardinality: once per forwarding direction per distinct
545    /// `outcome`**, which is the same rule [`Self::Shaped`] states per
546    /// stream — the direction is a datagram's whole scope. Running totals
547    /// live in [`ShapeStats`](crate::shape::ShapeStats).
548    ShapedDatagram {
549        /// The session identifier.
550        session_id: SessionId,
551        /// The side the datagram arrived on.
552        side: ProxySide,
553        /// The class the datagram resolved to, or an empty string for one
554        /// that matched no rule.
555        class: String,
556        /// What the shaper did.
557        outcome: ShapeOutcome,
558    },
559}
560
561/// What [`ProxyEvent::Shaped`] reports the shaper did.
562///
563/// `#[non_exhaustive]`; derives `PartialEq`/`Eq` like [`Effect`] and
564/// [`ImpairmentKind`], so a test destructures the event and compares this.
565#[derive(Debug, Clone, PartialEq, Eq)]
566#[non_exhaustive]
567pub enum ShapeOutcome {
568    /// A unit was discarded by
569    /// [`Overflow::DropTail`](crate::shape::Overflow::DropTail).
570    ///
571    /// The unit went through the framer's elide path, so absolute object
572    /// IDs on drafts 14-23 stay correct; a unit an elide guard refused was
573    /// admitted instead of dropped and is reported as a refusal, not here.
574    Dropped,
575    /// A queued unit outlived `max_hold` under
576    /// [`Expiry::ResetStream`](crate::shape::Expiry::ResetStream).
577    ///
578    /// Decided at release time rather than at admission: the queue notices
579    /// the overrun when it next looks at its head, replaces everything it
580    /// was holding with the reset the policy asked for, and reports this.
581    /// So the event says the whole destination stream was abandoned, not
582    /// that one unit was — which is why it carries an empty `class` for the
583    /// same reason [`Self::StreamReset`] does.
584    ///
585    /// **Emitted once per stream.** The queue is gone after the first one,
586    /// so there is nothing left to expire.
587    ///
588    /// This variant read "no producer yet" for as long as expiry was decided
589    /// nowhere; it now has one, and asserting on it is
590    /// `actions_shaping.rs`'s business rather than a promise waiting to be
591    /// kept.
592    Expired,
593    /// A datagram was discarded because its class's bucket had no tokens
594    /// for it.
595    ///
596    /// **Policing rather than shaping**, and the difference is that there is
597    /// no queue: a datagram arriving over its class's configured rate is
598    /// dropped at admission rather than held until the tokens arrive. That
599    /// is the only sound answer for this carrier — a queue would impose an
600    /// order the protocol does not have, and a datagram has no successor
601    /// whose framing depends on it, which is what makes discarding one
602    /// harmless where discarding a queued stream unit is not.
603    ///
604    /// Carried only by [`ProxyEvent::ShapedDatagram`]. [`Self::Dropped`] is
605    /// the stream carrier's answer and says something different: that an
606    /// overflow policy discarded a unit the queue had no room for.
607    Policed,
608    /// The destination stream was abandoned by an overflow or expiry
609    /// policy.
610    StreamReset {
611        /// The application error code it was reset with.
612        code: u64,
613    },
614}
615
616/// What an executed action actually did.
617///
618/// Unlike [`ProxyEvent`] this derives `PartialEq` and `Eq`: destructure
619/// the event, then `assert_eq!` on what comes out. See
620/// [`ProxyEvent`]'s own rustdoc for the shape.
621#[derive(Debug, Clone, PartialEq, Eq)]
622#[non_exhaustive]
623pub enum Effect {
624    /// The original bytes were forwarded.
625    ForwardedVerbatim,
626    /// Replacement bytes were forwarded.
627    Replaced {
628        /// How many bytes were written.
629        bytes: usize,
630    },
631    /// An object was removed. `renumbered_successor` is `true` when the next
632    /// object on the stream has to be re-encoded against the one now in
633    /// front of it: its leading ID varint rewritten on a subgroup stream of
634    /// drafts 14-23, its whole framing re-encoded on a fetch stream of
635    /// drafts 15-23.
636    ///
637    /// It says a fix-up is **owed**, not that the bytes moved. A survivor
638    /// that already stated everything it needed comes through unchanged, and
639    /// the debt is settled all the same.
640    Elided {
641        /// Whether a successor fix-up is now pending.
642        renumbered_successor: bool,
643    },
644    /// The unit was queued for later release.
645    Queued {
646        /// When it is due.
647        release_at: Instant,
648    },
649    /// A prefix was written and the stream reset.
650    ///
651    /// `forwarded` counts bytes **handed to the transport**, not bytes the
652    /// peer observed. quinn clears the peer's receive assembler when the
653    /// reset is processed, so the peer may observe fewer — never more.
654    Truncated {
655        /// Bytes of this unit written before the reset.
656        forwarded: usize,
657        /// The reset code, as the action stated it.
658        code: u64,
659        /// `false` when the draft defines no stream-reset code vocabulary
660        /// (drafts 07-10), so `code` is a choice rather than a claim.
661        code_defined: bool,
662    },
663    /// The destination stream was reset.
664    StreamReset {
665        /// The reset code.
666        code: u64,
667        /// `false` when the draft defines no stream-reset code vocabulary
668        /// (drafts 07-10), so `code` is a choice rather than a claim.
669        code_defined: bool,
670    },
671    /// The stream was never forwarded.
672    StreamRejected {
673        /// The code the source was stopped with.
674        code: u64,
675    },
676    /// A session close was requested.
677    SessionClosing {
678        /// The session termination code.
679        code: u32,
680    },
681    /// The unit was not forwarded.
682    Dropped,
683}
684
685/// A reduction in what the proxy can do or observe.
686///
687/// **Every variant states its emission cardinality**, because the counts
688/// differ by an order of magnitude between them — some fire once per
689/// session, some once per stream, and some once per unit — and a test
690/// that assumes "exactly one" against a per-unit variant is a test that
691/// goes red for the wrong reason. Where a variant is capped below its
692/// natural rate, a counter on
693/// [`crate::instrument::Counters`] carries the running total instead.
694///
695/// Like [`Effect`], this derives `PartialEq` and `Eq`; the event that
696/// carries it does not.
697///
698/// # Every variant here has a producer
699///
700/// None of these is reserved, aspirational or waiting for a call site: each
701/// one is emitted by code in this crate, and the emission happens **after**
702/// what it reports — see [`ProxyEvent::Impairment`] for why that ordering is
703/// the whole value of the surface.
704///
705/// One is harder to reach than the rest and says so on itself:
706/// [`Self::CoarseReleaseTimer`] needs `MOQTAP_RELEASE_TIMER=condvar`,
707/// because every default release backend is high-resolution. That is a
708/// diagnostic override rather than a fallback, and the variant exists so a
709/// run made under it cannot quote delays it was unable to honour.
710///
711/// A variant added here is forced to answer one more question before it
712/// compiles: which of the proxy's two connections it is about. The mapping
713/// that answers it matches exhaustively with no catch-all, so a new report
714/// stops the crate building rather than defaulting to a leg somebody else
715/// chose.
716#[derive(Debug, Clone, PartialEq, Eq)]
717#[non_exhaustive]
718pub enum ImpairmentKind {
719    /// The framer stopped parsing a stream, so no object on it is
720    /// addressable. Emitted exactly once per stream.
721    FramerBypass {
722        /// The source stream.
723        stream_id: u64,
724        /// The draft it was parsed as.
725        draft: DraftVersion,
726        /// Why parsing stopped.
727        reason: crate::types::BypassReason,
728    },
729    /// An object exceeded the framer's buffer cap and was streamed through
730    /// without being addressable.
731    ///
732    /// Emitted **at most once per stream**, on the first such object, and
733    /// carries the running count. Every other `ImpairmentKind` states its
734    /// cardinality; this one did not, and it fires per object — a stream
735    /// of large objects under a low `max_buffered_object_bytes` would
736    /// otherwise emit one event per object and swamp an observer written
737    /// against the once-per-stream cardinality every neighbouring variant
738    /// documents. The count keeps the information: `total` is the number
739    /// of unaddressable objects seen on this stream **at the moment of
740    /// emission**, i.e. `1`, and
741    /// [`crate::instrument::Counters::objects_not_addressable`] is the
742    /// running total that stays accurate afterwards.
743    ObjectNotAddressable {
744        /// The source stream.
745        stream_id: u64,
746        /// Unaddressable objects on this stream so far.
747        total: u64,
748    },
749    /// A control frame's body was refused by the decoder, so the proxy
750    /// forwarded a message it could not read.
751    ///
752    /// The frame's declared length was intact — that is what let the parser
753    /// find the frame behind it — and only the message inside it failed to
754    /// decode. The bytes reach the peer regardless, in the position they
755    /// held: on the observation-only control pipe they were forwarded
756    /// before anything was parsed, and on the mutating pipe, where the
757    /// parser owns the forwarding path, they are written verbatim without
758    /// a hook being consulted. So what a refusal costs is this proxy's
759    /// account of a message, never the message.
760    ///
761    /// That account is the whole product, which is why the loss is
762    /// reported. Without this event a control message the proxy could not
763    /// read is indistinguishable from one the peer never sent, and the two
764    /// call for opposite conclusions. It is not a per-draft hazard: a
765    /// Message Type the configured draft does not assign, a frame whose
766    /// body does not match its declared length, and anything an extension
767    /// adds all take the same path on all the drafts.
768    ///
769    /// # Cardinality: at most once per control stream direction
770    ///
771    /// Emitted on the first refused frame, carrying the count as it stood
772    /// when the report went out. A peer repeating an unassigned Message
773    /// Type would otherwise emit one event per frame, against neighbouring
774    /// variants an observer has been told fire once per stream — the same
775    /// argument [`Self::ObjectNotAddressable`] makes for the same shape of
776    /// hazard.
777    ///
778    /// The running figure that stays accurate afterwards is
779    /// [`crate::instrument::Counters::control_frames_not_decodable`]. That
780    /// counter and this event count different things on purpose: one frame
781    /// refused and forty refused produce one event each and differ by
782    /// thirty-nine there.
783    ControlFrameNotDecodable {
784        /// The Message Type varint the **first** refused frame declared.
785        ///
786        /// The type is read from the frame header, which decoded; nothing
787        /// inside the frame did, so this is the whole of what the refused
788        /// message can still say about itself. A later refusal of a
789        /// different type is counted and not named — see the cardinality
790        /// note above.
791        type_id: u64,
792        /// Frames refused on this direction when the report went out.
793        total: u64,
794    },
795    /// A stream's pending queue reached
796    /// [`crate::action::EgressConfig::max_pending_bytes`], so delay has
797    /// become backpressure. Emitted once per stream, on the transition
798    /// into backpressure — a queue that drains and fills again does not
799    /// report twice.
800    EgressQueueFull {
801        /// The source stream.
802        stream_id: u64,
803    },
804    /// A delay, or a shaped release, was clamped to
805    /// [`crate::action::EgressConfig::max_hold`].
806    ///
807    /// Emitted **once per clamped unit**, not once per stream: a hook
808    /// that returns an over-long `Delay` for every object emits one event
809    /// per object. It is a property of the decision, and the decision is
810    /// taken again for the next unit.
811    ///
812    /// # `requested: None` is an unbounded wait, not a missing figure
813    ///
814    /// A hook's own `Delay { by }` names a duration, so it reports
815    /// `Some(by)`. A shaped release frequently names none: a class whose
816    /// rate is zero, or whose burst is smaller than the unit at the head of
817    /// its queue, has **no** refill instant, and the clamp is the only
818    /// thing that will ever release that unit. `None` is that case, and it
819    /// is the honest report — there is no duration to quote and inventing a
820    /// finite one would be a fabrication.
821    ///
822    /// The unbounded case was once reported as `Duration::MAX`, which
823    /// reaches a log as `18446744073709551615.999999999s`. Two things went
824    /// wrong with that. A reader sees a 584-billion-year request beside a
825    /// 2 s applied one and takes it for an encoding fault in whatever
826    /// rendered it, filing against the wrong component; and the natural
827    /// assertion that a clamp *reduced* the request — `requested > applied`
828    /// — cannot tell that sentinel from a real thirty-second request, so it
829    /// passes either way and certifies nothing.
830    HoldClamped {
831        /// What was asked for, or `None` when the wait was unbounded.
832        requested: Option<Duration>,
833        /// What was applied: the `max_hold` ceiling.
834        applied: Duration,
835    },
836    /// A **control** stream's destination ended with a FIN part-way
837    /// through a message. The proxy does not synthesize a reset there —
838    /// that would be a session-level protocol violation — and it does not
839    /// finish the message either, because completing it would mean
840    /// inventing control-stream bytes neither peer wrote. On a data stream
841    /// the equivalent failure synthesizes a reset, which is what makes the
842    /// truncation visible to the peer; here only this report carries it.
843    ///
844    /// Two things produce it, and `error` says which:
845    ///
846    /// 1. a non-reset read failure on the source half, which is the last
847    ///    read that stream will ever do;
848    /// 2. the drain window of a requested session close expiring while a
849    ///    message was part-written — see
850    ///    [`ProxyControl::close_session`](crate::control::ProxyControl::close_session).
851    ///
852    /// Emitted **at most once per control stream direction**. The two
853    /// producers cannot both fire on one direction: the first returns from
854    /// the pipe, so the second is unreachable afterwards. A session with
855    /// both control directions affected emits two, one per `side`.
856    ///
857    /// The second producer says nothing at all when the proxy cannot tell
858    /// where the message boundaries are — a control stream whose framing
859    /// could not be followed reports no truncation rather than a guessed
860    /// one.
861    ControlStreamTruncated {
862        /// The failure that caused it.
863        error: String,
864    },
865    /// Queued bytes were still pending when the session tore down. They
866    /// were flushed best-effort; `bytes` may not have reached the peer.
867    ///
868    /// Emitted **at most once per stream**, at teardown, and only for a
869    /// stream that still had something queued when teardown began — a
870    /// stream that drained at its release times reports nothing.
871    ///
872    /// `bytes` counts everything the teardown flush could not vouch for:
873    /// what it could not write **and** what it wrote into a transport the
874    /// session was already closing. The second half is not pedantry — a
875    /// `write_all` into quinn returns `Ok` as soon as the bytes are
876    /// buffered, and `Connection::close` discards that buffer, so counting
877    /// only the residue reports zero for precisely the case that loses
878    /// data. A peer that did receive the bytes gets a spurious impairment;
879    /// that is the safe side to err on.
880    QueuedBytesAtTeardown {
881        /// The source stream.
882        stream_id: u64,
883        /// How many bytes could not be confirmed delivered.
884        bytes: usize,
885    },
886    /// A datagram could not be handed to the transport, and the session
887    /// survived.
888    ///
889    /// Emitted **once per rejected datagram** — this is a per-unit
890    /// variant, so a source steadily sending datagrams above the path MTU
891    /// produces one event each. Nothing caps it, because unlike
892    /// [`Self::ObjectNotAddressable`] there is no stream to attribute a
893    /// running total to.
894    ///
895    /// Distinct from [`ProxyEvent::ActionFailed`], which requires a `site`
896    /// and an `action`: this is the un-hooked path, where nobody acted
897    /// and there is nothing to name. Both exist because
898    /// `forward_datagrams` drops the `?` on **every** `send_datagram`
899    /// call, not only the ones behind a hook.
900    DatagramNotSent {
901        /// The transport's error.
902        error: String,
903    },
904    /// The release wheel is not high-resolution on this host, so
905    /// [`crate::action::Action::Delay`] cannot resolve below the ~15.6 ms
906    /// system tick and every delay shorter than it is really a tick.
907    ///
908    /// The default backend is high-resolution on every platform, so in
909    /// 0.4.0 the only way to reach this is
910    /// `MOQTAP_RELEASE_TIMER=condvar` on Windows — the diagnostic
911    /// override, not a fallback. It is reported rather than assumed away
912    /// because a forced backend is still a run whose delays were not
913    /// honoured.
914    ///
915    /// Emitted **once per session**, on that session's first deferred
916    /// release. The measured shortfall is in
917    /// [`crate::instrument::Counters::release_errors`]; this event exists
918    /// so a report cannot quote "5 ms delay applied" while silently
919    /// having been unable to apply it.
920    CoarseReleaseTimer {
921        /// The backend the release thread resolved to.
922        backend: crate::instrument::TimerBackend,
923        /// Reserved for a future backend that can fail to initialise.
924        /// **Always `None` in 0.4.0**: the only coarse backend is the one
925        /// `MOQTAP_RELEASE_TIMER` forces, and nothing failed for it to
926        /// describe.
927        detail: Option<String>,
928    },
929    /// The framer stopped parsing a stream while an elide fix-up was
930    /// still owed, so the remainder of the stream cannot be renumbered.
931    ///
932    /// The destination stream is **reset** rather than forwarded, because
933    /// the alternative is delivering bytes that are known to decode to
934    /// the wrong Object IDs. This is the residual case of the elide
935    /// mechanism, and it is reachable: a mid-stream decode error or an
936    /// unmeasurable oversized object can latch bypass at any point after
937    /// an elide.
938    ///
939    /// Emitted **at most once per stream**, and it is that stream's last
940    /// event: the reset has already been asked for by the time this
941    /// arrives, and the forwarding task returns immediately afterwards.
942    /// The order is that way round on purpose — `code` names the value the
943    /// destination *was* reset with, so an event raised above the reset
944    /// would describe a wire change that had not been made, and this arm
945    /// has no later event to correct it with. It is the
946    /// `fixup_owed == true` half of the one `FramerOut::Bypassed` a
947    /// stream can produce — the half that reports [`Self::FramerBypass`]
948    /// is the other.
949    ElideFixupLost {
950        /// The source stream.
951        stream_id: u64,
952        /// Why the framer gave up.
953        reason: crate::types::BypassReason,
954        /// The code the destination was reset with (`0x0`).
955        code: u64,
956    },
957    /// A [`StreamAction::SerializeAfter`](crate::action::StreamAction::SerializeAfter)
958    /// named a stream there is nothing to wait for, so the stream it was
959    /// returned on proceeded immediately.
960    ///
961    /// Three cases reach it and they are deliberately one report: the target
962    /// never existed, the target had already ended, or the target is this
963    /// stream itself. All three are a hook naming a stream that cannot end
964    /// later than now, and in all three the honest engine behaviour is to
965    /// proceed — a serialize that silently held forever would be a
966    /// `max_hold` stall attributed to the wrong thing.
967    ///
968    /// Emitted **once per stream**. A stream takes at most one serialize
969    /// decision at each of the two stream sites, and a decision that names a
970    /// live target reports nothing at all.
971    ///
972    /// Both fields are [`StreamKey`]s, the identity a hook names a stream by.
973    SerializeTargetUnknown {
974        /// The stream that asked to be serialized.
975        key: StreamKey,
976        /// The target it named.
977        target: StreamKey,
978    },
979    /// A [`ClassRule`](crate::shape::ClassRule) keys on a field the wire
980    /// does not carry on this draft and stream kind, so it can never claim
981    /// a unit and everything it was aimed at falls to the default class.
982    /// This is the report that keeps the rule **a key the wire does not carry
983    /// never matches** from being a silent no-op. `Discipline::StrictPriority`
984    /// is specified on `publisher_priority`, which is `None` on drafts 15-19
985    /// under the default-priority bit — so without this, *starve video while
986    /// audio flows* is a silent no-op on the five newest drafts and the run
987    /// reports success. A rule aimed at
988    /// [`MatchKind::Datagram`](crate::shape::MatchKind::Datagram) reports
989    /// through the same seam, from the datagram forwarder rather than from the
990    /// framer, and reports the one key a datagram can be missing: its priority,
991    /// which drafts 15 and later let a type byte leave off. The key no datagram
992    /// *ever* carries is refused before the run instead — see
993    /// `Capabilities::admit_class`.
994    ///
995    /// Emitted **once per session per `(class, field)`**, on the first unit
996    /// that reaches the rule. Not once per stream and not once per unit: it
997    /// is a statement about a *profile* against a *draft*, both of which are
998    /// fixed for the session. The falsifiable companion an author reads is
999    /// `ShapeStats::default_class`, which is where the units went.
1000    ///
1001    /// A rule whose key is *present but out of range* reports nothing —
1002    /// that is a rule working.
1003    ShapeRuleUnmatchable {
1004        /// The [`ClassRule::name`](crate::shape::ClassRule::name) that
1005        /// cannot fire.
1006        class: String,
1007        /// The key it named that the wire did not carry.
1008        field: crate::shape::MatcherField,
1009        /// The draft the session is running as, which is half of why the
1010        /// key is absent.
1011        draft: DraftVersion,
1012    },
1013    /// A class's [`BucketConfig::burst_bytes`](crate::shape::BucketConfig::burst_bytes)
1014    /// is smaller than the units it is being asked to pace, so its
1015    /// configured rate never binds and every unit leaves at its `max_hold`
1016    /// clamp instead.
1017    ///
1018    /// A token bucket can never grant a unit larger than the whole bucket —
1019    /// there is no amount of refilling that covers it — so a burst below one
1020    /// object turns a rate into a metronome running at
1021    /// `queue depth / max_hold`. Measured: `rate_bps` of 1 000 000 with
1022    /// `burst_bytes` of 100 delivered 1000-byte objects at exactly the
1023    /// clamp, a figure the configuration never mentions.
1024    ///
1025    /// **This is the report that makes that case distinguishable.** Without
1026    /// it the symptoms are `HoldClamped` on every unit and a rising
1027    /// `tokens_exhausted_episodes` — which is *also* precisely what a class
1028    /// that is genuinely rate-limited produces, so an author who wrote a rate
1029    /// and left the burst at its default read a plausible-looking starved
1030    /// class and no indication that their number had been ignored.
1031    ///
1032    /// A class whose `rate_bps` is `Some(0)` never reaches this report. That
1033    /// is a class configured to stop, doing what it was asked; only a rate
1034    /// that was asked for and cannot be applied is a fault.
1035    ///
1036    /// It cannot be rejected when the profile is built:
1037    /// [`ShapeProfile::try_new`](crate::shape::ShapeProfile::try_new) has the
1038    /// burst but not the object sizes, and the sizes are half the comparison.
1039    ///
1040    /// Emitted **once per session per class**. The burst is a property of the
1041    /// profile, so every stream carrying the class reproduces it and every
1042    /// unit of it re-triggers it; the running figures beside this report are
1043    /// the class's own `tokens_exhausted_episodes` and one `HoldClamped` per
1044    /// clamped unit.
1045    ShapeBurstBelowUnit {
1046        /// The [`ClassRule::name`](crate::shape::ClassRule::name) whose rate
1047        /// is not being applied.
1048        class: String,
1049        /// The bucket cap, as configured.
1050        burst_bytes: u64,
1051        /// The wire size of the unit it could not cover — the other half of
1052        /// the comparison, so the report is actionable without a second
1053        /// measurement.
1054        unit_bytes: u64,
1055    },
1056    /// An object too large for the framer to buffer was forwarded without
1057    /// passing any token bucket, so the named class's configured rate was
1058    /// exceeded by exactly that object.
1059    ///
1060    /// Shaping is per unit and a unit is classified from its `ObjectMeta`.
1061    /// An object beyond
1062    /// [`FramerConfig::max_buffered_object_bytes`](crate::framer::FramerConfig::max_buffered_object_bytes)
1063    /// has none — the framer streams it through rather than measuring it —
1064    /// so no rule can claim it, no bucket can charge it, and the release seam
1065    /// grants it unconditionally. Measured: a 4 MiB object crossed in 800 ms
1066    /// against a class holding a bucket configured at zero bytes per second.
1067    ///
1068    /// This report is what keeps that from being a silent breach. `class` is
1069    /// the class the stream's *classified* units are charged to, which is the
1070    /// rate the escaping object was nominally under — an empty string on a
1071    /// stream that has not classified anything yet, matching the unnamed rows
1072    /// on [`ShapeStats`](crate::shape::ShapeStats). The bytes themselves are
1073    /// accounted on
1074    /// [`ShapeStats::unshapeable`](crate::shape::ShapeStats::unshapeable), so
1075    /// the conservation identity still closes; what was missing was anything
1076    /// naming the class whose ceiling they went over.
1077    ///
1078    /// Paired with, and deliberately distinct from,
1079    /// [`ObjectNotAddressable`](Self::ObjectNotAddressable): that one says
1080    /// the *hook* cannot address the object, this one says the *shaper* did
1081    /// not pace it. A session with no
1082    /// [`ShapeProfile`](crate::shape::ShapeProfile) emits the first and never
1083    /// the second, because there is no rate to exceed.
1084    ///
1085    /// Emitted **at most once per stream**, on the first such object, for the
1086    /// reason `ObjectNotAddressable` is capped the same way: a stream of
1087    /// large objects would otherwise emit one event each.
1088    ShapeUnpacedObject {
1089        /// The [`ClassRule::name`](crate::shape::ClassRule::name) this
1090        /// stream's classified units are charged to, or an empty string when
1091        /// no rule has claimed one yet.
1092        class: String,
1093        /// The source stream.
1094        stream_id: u64,
1095        /// Wire bytes of the unpaced chunk that triggered the report.
1096        bytes: u64,
1097    },
1098    /// Two units on one destination stream resolved to **different** shaping
1099    /// classes, so that stream's throughput is decided by whichever class is
1100    /// at its head rather than by any one class's bucket.
1101    ///
1102    /// This is not a defect and not a refusal — it is what per-unit
1103    /// classification over a single per-stream FIFO *means*. Reordering the
1104    /// queue by class is forbidden outright: object IDs are delta-encoded on
1105    /// the wire on drafts 14-23, and the framer's only re-encoding primitive
1106    /// handles removal, not reordering. So the head gates everything behind
1107    /// it whatever class those units are, and this report is what keeps that
1108    /// from being mistaken for the shaping the author configured.
1109    ///
1110    /// Emitted **once per stream**, on the first disagreement. The running
1111    /// figures beside it are
1112    /// [`ShapeStats::streams_with_mixed_classes`](crate::shape::ShapeStats::streams_with_mixed_classes)
1113    /// and, per class,
1114    /// [`ClassStats::starved_behind_other_class`](crate::shape::ClassStats::starved_behind_other_class)
1115    /// — which is deliberately *not*
1116    /// [`ClassStats::tokens_exhausted_episodes`](crate::shape::ClassStats::tokens_exhausted_episodes):
1117    /// two causes of waiting, two counters.
1118    ///
1119    /// **This report's `leg` and the proxy-wide counter's cell disagree by
1120    /// exactly one leg, on purpose.** This event answers the departing
1121    /// connection, because the stream whose throughput is now shared is the
1122    /// one being written to; the same occurrence is charged to the *arrival*
1123    /// cell of
1124    /// [`ProxyStats::per_leg`](crate::shape::ProxyStats::per_leg), so it sits
1125    /// beside the `bytes_shaped` that explains it. Correlating an event with
1126    /// a cell means expecting the two labels to differ — see
1127    /// [`DirectionStats::streams_with_mixed_classes`](crate::shape::DirectionStats::streams_with_mixed_classes),
1128    /// which states it from the other side.
1129    ///
1130    /// Keyed on [`StreamKey`], as
1131    /// [`SerializeTargetUnknown`](Self::SerializeTargetUnknown) is. The
1132    /// transport id rides along because it is what correlates this report
1133    /// with every other event in this enum.
1134    ClassChangedMidStream {
1135        /// Session-local identity of the destination stream carrying both
1136        /// classes.
1137        key: StreamKey,
1138        /// Transport stream id, for correlation with the other events in
1139        /// this enum; `key` is what identifies.
1140        stream_id: u64,
1141    },
1142}
1143
1144/// The connection an [`ImpairmentKind`] is about, given the side the task
1145/// that raised it was forwarding from.
1146///
1147/// The whole mapping lives here, in one exhaustive `match`, rather than at
1148/// the twenty-odd sites that raise these reports. Two reasons, and the
1149/// second is the one that matters.
1150///
1151/// A raising site knows only the direction it reads from. Working out that
1152/// a queue it could not flush belongs to the *other* connection is a turn
1153/// each site would have to make for itself, and a site that got it wrong
1154/// would produce a number that is correct under a wrong label — the failure
1155/// this whole surface is written to avoid, and one no test at that site
1156/// would notice, because the event would still arrive and still carry a
1157/// plausible leg.
1158///
1159/// And the turn is a property of the **kind**, not of the site. Whether a
1160/// report is about what came in or about what could not go out is decided
1161/// by what the report *says*, so the decision belongs beside the enum that
1162/// says it. A variant added to [`ImpairmentKind`] stops this function
1163/// compiling — the `match` has no catch-all on purpose — which is the one
1164/// place a new report reliably gets asked which leg it means.
1165///
1166/// # `arrived_on` must be an ingress side
1167///
1168/// Every [`Reporter`](crate::exec::Reporter) in this crate is built with the
1169/// side its pipe *reads* from, so `arrived_on` is `ClientToProxy` or
1170/// `RelayToProxy` in practice. The function is still total over all four,
1171/// because `ProxySide` has four variants and a panicking forwarding path is
1172/// worse than a defensible answer: an egress side is read as naming its own
1173/// connection, which is what it does.
1174pub(crate) fn impairment_leg(kind: &ImpairmentKind, arrived_on: ProxySide) -> Option<Leg> {
1175    let here = connection_of(arrived_on);
1176    match kind {
1177        // The parser gave up on bytes that arrived. Nothing here is a claim
1178        // about what could be written, so the far leg is not implicated.
1179        ImpairmentKind::FramerBypass { .. }
1180        | ImpairmentKind::ObjectNotAddressable { .. }
1181        | ImpairmentKind::ControlFrameNotDecodable { .. } => Some(here),
1182
1183        // Every one of these is a failure to place bytes on the far side —
1184        // a queue that filled in front of it, a release it clamped, bytes it
1185        // could not vouch for, a datagram it refused, a stream it had to
1186        // reset or could not renumber for, an object that crossed it
1187        // unpaced, a destination whose throughput two classes now share, a
1188        // first write that was held. The connection they are about is the
1189        // one being written to, which is the opposite of the one they
1190        // arrived on.
1191        ImpairmentKind::EgressQueueFull { .. }
1192        | ImpairmentKind::HoldClamped { .. }
1193        | ImpairmentKind::QueuedBytesAtTeardown { .. }
1194        | ImpairmentKind::DatagramNotSent { .. }
1195        | ImpairmentKind::ControlStreamTruncated { .. }
1196        | ImpairmentKind::ElideFixupLost { .. }
1197        | ImpairmentKind::ShapeUnpacedObject { .. }
1198        | ImpairmentKind::ClassChangedMidStream { .. }
1199        | ImpairmentKind::SerializeTargetUnknown { .. } => Some(other_connection(here)),
1200
1201        // A profile compared against the sizes it was asked to pace, or the
1202        // process compared against its host. All three are equally true of
1203        // both connections and none of them stops being true if one leg goes
1204        // away, so naming a leg would be naming whichever pipe noticed first.
1205        ImpairmentKind::CoarseReleaseTimer { .. }
1206        | ImpairmentKind::ShapeRuleUnmatchable { .. }
1207        | ImpairmentKind::ShapeBurstBelowUnit { .. } => None,
1208    }
1209}
1210
1211/// The connection a direction of travel belongs to.
1212fn connection_of(side: ProxySide) -> Leg {
1213    match side {
1214        ProxySide::ClientToProxy | ProxySide::ProxyToClient => Leg::Client,
1215        ProxySide::ProxyToRelay | ProxySide::RelayToProxy => Leg::Upstream,
1216    }
1217}
1218
1219/// The proxy's other connection. There are exactly two.
1220fn other_connection(leg: Leg) -> Leg {
1221    match leg {
1222        Leg::Client => Leg::Upstream,
1223        Leg::Upstream => Leg::Client,
1224    }
1225}