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