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}