diff --git a/doc/bin/relay/http.md b/doc/bin/relay/http.md index 2a2efc21471..110933dad63 100644 --- a/doc/bin/relay/http.md +++ b/doc/bin/relay/http.md @@ -42,10 +42,23 @@ listen = "127.0.0.1:9101" ### GET /metrics -Prometheus text: bytes, frames, groups, announces, subscriptions, viewers, requests, -and sessions split by `tier` and `role`, plus accept-loop counters per TCP listener. Traffic -counters accumulate for the node's lifetime and leave out the stats feed's own -traffic. Host CPU and memory belong to a node exporter. +Prometheus text: payload bytes, frames, groups, announces, subscriptions, and +viewers split by `tier` and `role`; requests and sessions by `tier`; and +accept-loop counters per TCP listener. Payload counters accumulate for the +node's lifetime and leave out the stats feed's own traffic. Host CPU and memory +belong to a node exporter. + +`moq_relay_wire_tx_bytes_total` and `moq_relay_wire_rx_bytes_total` sum the +per-session [transport counters](/concept/stats#sessions) by `tier`, including +internal sessions and the stats feed. Compare them with payload to spot +transport overhead and idle traffic. Transports that do not report them +contribute zero. + +`moq_relay_control_tx_bytes_total` and `moq_relay_control_rx_bytes_total` +count control-message bytes by `tier` and `class` (`session`, `announce`, `track_info`, `subscribe`, `fetch`, +`probe`, `auth`, `goaway`, `other`) to spot control-heavy sessions. They +include message size prefixes but leave out media payload and transport +overhead such as QUIC framing, ACKs, and retransmissions. A few worth alerting on: diff --git a/rs/moq-net/src/client.rs b/rs/moq-net/src/client.rs index 5a2f9fa5231..3e31d37bec7 100644 --- a/rs/moq-net/src/client.rs +++ b/rs/moq-net/src/client.rs @@ -226,6 +226,7 @@ impl Client { let start = lite::start(lite::Config { runtime: runtime.clone(), + stats: self.stats.clone(), client: true, limits: self.limits, session: session.clone(), @@ -314,6 +315,7 @@ impl Client { let auth = crate::auth::Handle::ietf(self.extensions.auth); let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), + stats: self.stats.clone(), limits: self.limits, session: session.clone(), setup: None, @@ -398,7 +400,7 @@ impl Client { Some(p) => return Err(Error::UnknownAlpn(p.to_string())), }; - let mut stream = Stream::open(&mut session, encoding).await?; + let mut stream = Stream::open(&mut session, encoding).await?.with_stats(&self.stats); // The encoding is always an IETF version for SETUP negotiation. let ietf_encoding = ietf::Version::try_from(encoding).map_err(|_| Error::Version)?; @@ -446,6 +448,7 @@ impl Client { let stream = stream.with_version(v); let start = lite::start(lite::Config { runtime: runtime.clone(), + stats: self.stats.clone(), client: true, limits: self.limits, session: session.clone(), @@ -492,6 +495,7 @@ impl Client { // Draft 14-16: the path rode in the bidi SETUP above, not the uni one. let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), + stats: self.stats.clone(), limits: self.limits, session: session.clone(), setup: Some(stream), diff --git a/rs/moq-net/src/coding/decode.rs b/rs/moq-net/src/coding/decode.rs index d1ba55f70f8..2a8443c2851 100644 --- a/rs/moq-net/src/coding/decode.rs +++ b/rs/moq-net/src/coding/decode.rs @@ -141,6 +141,11 @@ pub struct Decoder<'a> { len: usize, /// See [`Self::need`]. need: usize, + /// The control message read, counted once the read is consumed. + pub(crate) control: Option, + /// Leading bytes left out of the `control` count: a type selector the sender writes + /// apart from the message, and so never counts. + pub(crate) uncounted: usize, } impl<'a> Decoder<'a> { @@ -152,6 +157,8 @@ impl<'a> Decoder<'a> { form, len: buf.len(), need: 0, + control: None, + uncounted: 0, } } @@ -176,6 +183,12 @@ impl<'a> Decoder<'a> { self.form } + /// How many bytes have been read. + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn position(&self) -> usize { + self.len - self.buf.len() + } + /// The number of unread bytes. #[cfg_attr(not(feature = "async"), allow(dead_code))] pub fn remaining(&self) -> usize { diff --git a/rs/moq-net/src/coding/encode.rs b/rs/moq-net/src/coding/encode.rs index 34bfa547a5a..90cb6a590a1 100644 --- a/rs/moq-net/src/coding/encode.rs +++ b/rs/moq-net/src/coding/encode.rs @@ -59,12 +59,17 @@ pub trait Encode { pub struct Encoder<'a> { buf: &'a mut Vec, form: Form, + pub(crate) control: Option, } impl<'a> Encoder<'a> { /// Append to `buf`, with varints in the given form. pub fn new(buf: &'a mut Vec, form: Form) -> Self { - Self { buf, form } + Self { + buf, + form, + control: None, + } } /// The varint form this encoder writes. diff --git a/rs/moq-net/src/coding/reader.rs b/rs/moq-net/src/coding/reader.rs index 61a9ceb0d04..f565b482b4b 100644 --- a/rs/moq-net/src/coding/reader.rs +++ b/rs/moq-net/src/coding/reader.rs @@ -26,6 +26,7 @@ pub struct Reader { want: usize, /// A contiguous copy of a message that spans transport chunks, empty between decodes. scratch: Vec, + stats: crate::stats::Session, } /// How much payload may pile up behind a parked consumer before @@ -49,9 +50,20 @@ impl Reader { version, want: 0, scratch: Vec::new(), + stats: Default::default(), } } + pub(crate) fn with_stats(mut self, stats: &crate::stats::Session) -> Self { + self.stats = stats.clone(); + self + } + + /// Where control messages read here are counted. + pub(crate) fn stats(&self) -> &crate::stats::Session { + &self.stats + } + /// Poll the stream for a lease of at least `want` bytes. fn poll_lease(&mut self, cx: &mut Context<'_>, want: usize) -> Poll, Error>> { let version = &self.version; @@ -68,6 +80,19 @@ impl Reader { consume: bool, decode: impl Fn(&mut Decoder<'_>) -> Result, ) -> Poll> + where + V: Into
+ Copy, + { + self.poll_with_sized(cx, consume, decode).map_ok(|(value, _)| value) + } + + /// [`Self::poll_with`], also returning how many bytes the value took. + fn poll_with_sized( + &mut self, + cx: &mut Context<'_>, + consume: bool, + decode: impl Fn(&mut Decoder<'_>) -> Result, + ) -> Poll> where V: Into + Copy, { @@ -101,15 +126,21 @@ impl Reader { let need = match decode(&mut r) { Ok(value) => { let used = buf.len() - r.remaining(); + let control = r.control.map(|class| (class, used - r.uncounted)); self.want = 0; if self.scratch.capacity() > SCRATCH_KEEP { self.scratch = Vec::new(); } if consume { + // Counted only once the bytes are released, so a peek or a retry + // after a short read never counts the same message twice. + if let Some((class, size)) = control { + self.stats.control(class, size, true); + } lease.advance(used); lease.release(); } - return Poll::Ready(Ok(value)); + return Poll::Ready(Ok((value, used))); } // A short read inside a complete body is malformed, but reports what it // needed from the body, so always ask for at least one byte more. @@ -170,6 +201,27 @@ impl Reader { std::future::poll_fn(|cx| self.poll_decode(cx)).await } + /// [`Self::poll_decode`], also returning the bytes it consumed, for a handshake to + /// count once the session's stats are known. + pub(crate) fn poll_decode_sized + Debug>( + &mut self, + cx: &mut Context<'_>, + ) -> Poll> + where + V: Into + Copy, + { + let version = self.version; + self.poll_with_sized(cx, true, |r| T::decode(r, version)) + } + + /// [`Self::decode`], also returning the bytes it consumed. + pub(crate) async fn decode_sized + Debug>(&mut self) -> Result<(T, usize), Error> + where + V: Into + Copy, + { + std::future::poll_fn(|cx| self.poll_decode_sized(cx)).await + } + /// Poll for the next message unless the stream is closed cleanly first. pub fn poll_decode_maybe + Debug>(&mut self, cx: &mut Context<'_>) -> Poll, Error>> where @@ -405,6 +457,7 @@ impl Reader { version, want: self.want, scratch: self.scratch, + stats: self.stats, } } } @@ -441,6 +494,75 @@ mod tests { Reader::new(Buffered::new(StopLog::default()), version) } + /// A control message counts once, when it is consumed: never on a peek, a retry after + /// a short read, or a failed decode, and in full when it spans chunks. + #[test] + fn control_counts_consumed_messages_once() { + use crate::{ + lite, + stats::{Bytes, Control, Registry, Tier}, + }; + let registry = Registry::new(Default::default()); + let session = registry.tier(Tier::default()).session(""); + let version = lite::Version::Lite07; + let msg = lite::Probe { + bitrate: Some(123), + rtt: Some(4), + }; + let bytes: &'static [u8] = msg.encode_bytes(version).unwrap().to_vec().leak(); + let size = bytes.len() as u64; + let (head, tail) = bytes.split_at(2); + let counted = || registry.snapshot().control(); + let probes = |n: u64| vec![(Tier::default(), Control::Probe, Bytes { tx: 0, rx: size * n })]; + + let mut reader = queue(&[bytes, head], version).with_stats(&session); + let mut cx = Context::from_waker(std::task::Waker::noop()); + assert!(matches!( + reader.poll_decode_peek_maybe::(&mut cx), + Poll::Ready(Ok(Some(_))) + )); + assert!(counted().is_empty()); + assert!(matches!(reader.poll_decode::(&mut cx), Poll::Ready(Ok(_)))); + assert_eq!(counted(), probes(1)); + + // The second probe arrives in two chunks, so the reader parks and retries. + assert!(reader.poll_decode::(&mut cx).is_pending()); + assert!(reader.poll_decode::(&mut cx).is_pending()); + assert_eq!(counted(), probes(1)); + reader.stream.get_mut().chunks.push_back(tail); + assert!(matches!(reader.poll_decode::(&mut cx), Poll::Ready(Ok(_)))); + assert_eq!(counted(), probes(2)); + + // A wrapped message counts its type too, as the sender's one encode does. + let reply = lite::AuthReply::Error(lite::AuthError { + code: 1, + reason: "no".into(), + }); + let reply: &'static [u8] = reply.encode_bytes(version).unwrap().to_vec().leak(); + reader.stream.get_mut().chunks.push_back(reply); + assert!(matches!( + reader.poll_decode::(&mut cx), + Poll::Ready(Ok(_)) + )); + let auth = ( + Tier::default(), + Control::Auth, + Bytes { + tx: 0, + rx: reply.len() as u64, + }, + ); + assert_eq!(counted(), [vec![auth.clone()], probes(2)].concat()); + + // A body cut short inside its declared size is malformed, and never counted. + reader.stream.get_mut().chunks.push_back(&[1, 0x80]); + assert!(matches!( + reader.poll_decode::(&mut cx), + Poll::Ready(Err(_)) + )); + assert_eq!(counted(), [vec![auth], probes(2)].concat()); + } + /// STOP_SENDING refuses a stream, so it must carry a stream code. Reusing the session /// table here would have the peer decode it against the wrong registry: `Cancel` would /// arrive as INTERNAL_ERROR, and `Unauthorized` as DELIVERY_TIMEOUT. diff --git a/rs/moq-net/src/coding/stream.rs b/rs/moq-net/src/coding/stream.rs index 23c08969f26..a84ffddbc4b 100644 --- a/rs/moq-net/src/coding/stream.rs +++ b/rs/moq-net/src/coding/stream.rs @@ -77,6 +77,13 @@ impl Stream { Self { writer, reader } } + pub(crate) fn with_stats(self, stats: &crate::stats::Session) -> Self { + Self { + writer: self.writer.with_stats(stats), + reader: self.reader.with_stats(stats), + } + } + /// Cast the stream to a different version, used during version negotiation. pub fn with_version(self, version: V2) -> Stream { Stream { diff --git a/rs/moq-net/src/coding/writer.rs b/rs/moq-net/src/coding/writer.rs index 31a7f5d9304..c852833560e 100644 --- a/rs/moq-net/src/coding/writer.rs +++ b/rs/moq-net/src/coding/writer.rs @@ -18,6 +18,7 @@ pub struct Writer { /// How much of `buffer` has already hit the stream. flushed: usize, version: V, + stats: crate::stats::Session, } impl Writer { @@ -28,9 +29,15 @@ impl Writer { buffer: Default::default(), flushed: 0, version, + stats: Default::default(), } } + pub(crate) fn with_stats(mut self, stats: &crate::stats::Session) -> Self { + self.stats = stats.clone(); + self + } + /// Encode the given message into the write buffer, to be sent by /// [`Self::poll_flush`]. An encode error leaves the buffer untouched. pub fn buffer + Debug>(&mut self, msg: &T) -> Result<(), Error> @@ -61,6 +68,9 @@ impl Writer { self.buffer.truncate(start); return Err(err.into()); } + if let Some(class) = w.control { + self.stats.control(class, self.buffer.len() - start, false); + } Ok(()) } @@ -227,6 +237,7 @@ impl Writer { buffer: std::mem::take(&mut self.buffer), flushed: self.flushed, version, + stats: self.stats.clone(), } } } @@ -348,6 +359,7 @@ mod tests { impl Encode for Poison { fn encode(&self, w: &mut Encoder<'_>, _: crate::lite::Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::Probe); w.slice(b"junk"); Err(EncodeError::BoundsExceeded) } @@ -369,6 +381,71 @@ mod tests { assert_eq!(*log.writes.lock().unwrap(), vec![5, 7], "the partial encode leaked"); } + #[test] + fn control_body_counts_successful_encodes_once_before_transport_delivery() { + use crate::{ + lite, + stats::{Bytes, Control, Registry, Tier}, + }; + let stats = Registry::new(Default::default()); + let session = stats.tier(Tier::new("internal")).session(""); + let gate = kio::Producer::new(false); + let mut writer = + Writer::new(SinkSend::gated(Log::default(), gate.consume()), lite::Version::Lite07).with_stats(&session); + writer.buffer(&lite::ControlType::Probe).unwrap(); + let msg = lite::Probe { + bitrate: Some(1_000_000), + rtt: Some(12), + }; + let size = msg.encode_bytes(lite::Version::Lite07).unwrap().len() as u64; + writer.buffer(&msg).unwrap(); + writer.buffer(&Poison).unwrap_err(); + assert!(writer.poll_flush(&mut Context::from_waker(Waker::noop())).is_pending()); + assert!(writer.poll_flush(&mut Context::from_waker(Waker::noop())).is_pending()); + assert_eq!( + stats.snapshot().control(), + vec![(Tier::new("internal"), Control::Probe, Bytes { tx: size, rx: 0 })] + ); + writer + .buffer(&lite::Track { + broadcast: "test".into(), + epoch: None, + track: "video".into(), + }) + .unwrap(); + let track_info = || { + stats + .snapshot() + .control() + .into_iter() + .find(|(_, class, _)| *class == Control::TrackInfo) + .unwrap() + .2 + .tx + }; + let before = track_info(); + let properties = lite::ietf::TrackProperties::default(); + writer.buffer(&properties).unwrap(); + assert_eq!( + track_info() - before, + properties.encode_bytes(lite::Version::Lite07).unwrap().len() as u64, + "negotiated track-property extensions are control bodies too" + ); + + writer + .buffer(&lite::SubscribeStart { + group: 3, + largest: None, + }) + .unwrap(); + writer.buffer(&lite::AnnounceBroadcast::Synced).unwrap(); + assert_eq!( + stats.snapshot().control().len(), + 4, + "classes do not collapse into the same counter" + ); + } + /// A flush interrupted mid-message resumes where it left off, so an /// abandoned wrapper future never desynchronizes the stream. #[test] diff --git a/rs/moq-net/src/ietf/adapter.rs b/rs/moq-net/src/ietf/adapter.rs index 38d8527c08b..bbab81b152e 100644 --- a/rs/moq-net/src/ietf/adapter.rs +++ b/rs/moq-net/src/ietf/adapter.rs @@ -849,6 +849,7 @@ impl Shared { #[derive(Clone)] pub struct ControlStreamAdapter { + stats: crate::stats::Session, inner: S, shared: Arc, control: Control, @@ -860,6 +861,7 @@ pub struct ControlStreamAdapter { impl ControlStreamAdapter { pub fn new(inner: S, control: Control, version: Version) -> Self { Self { + stats: Default::default(), inner, shared: Arc::new(Shared::default()), control, @@ -868,6 +870,11 @@ impl ControlStreamAdapter { } } + pub(super) fn with_stats(mut self, stats: &crate::stats::Session) -> Self { + self.stats = stats.clone(); + self + } + /// Open a real (non-virtual) bidi stream, bypassing control stream multiplexing. /// Used for v16 SubscribeNamespace which moved to its own bidi stream. pub async fn open_native_bi(&mut self) -> Result<(AdapterSend, AdapterRecv), crate::Error> { @@ -924,39 +931,22 @@ impl ControlStreamAdapter { /// message. Best-effort: an encode failure or a closed control stream is /// logged and dropped, since the session is tearing down anyway. pub(super) fn send_goaway(&self, uri: &str, timeout_ms: u64, version: Version) { - use crate::coding::Encode; - let msg = crate::ietf::GoAway { new_session_uri: std::borrow::Cow::Borrowed(uri), timeout: timeout_ms, request_id: None, }; - // The size prefix is a u16 on a stream shared with every other request, so the - // encode refuses a body that would wrap it and desynchronize the framing for all. - let mut raw = Vec::new(); - let mut w = Encoder::new(&mut raw, version.into()); - if let Err(err) = w - .varint(crate::ietf::GoAway::ID) - .and_then(|()| msg.encode(&mut w, version)) - { + if let Err(err) = self.queue_message(&msg, version) { tracing::warn!(%err, "failed to encode goaway"); - return; - } - - if !self.shared.control.push(raw.into()) { - tracing::debug!("control stream closed; goaway not sent"); } } - /// Refuse an update no request here can apply, naming its own Request ID. Draft 14 - /// answers no update. + /// Refuse an update no request here can apply, naming its own Request ID. fn refuse_update(&self, request_id: RequestId) { if self.version == Version::Draft14 { return; } - let mut raw = Vec::new(); - let mut w = crate::coding::Encoder::new(&mut raw, self.version.into()); let msg = ietf::RequestError { request_id: Some(request_id), error_code: super::error::request::to_code( @@ -967,31 +957,31 @@ impl ControlStreamAdapter { reason_phrase: "update not supported".into(), retry_interval: 0, }; - if let Err(err) = w - .varint(ietf::RequestError::ID) - .and_then(|()| crate::coding::Encode::encode(&msg, &mut w, self.version)) - { + if let Err(err) = self.queue_message(&msg, self.version) { tracing::warn!(%err, "failed to encode REQUEST_ERROR"); - return; } - // A closed control stream means the session is ending; there is no one to answer. - let _ = self.shared.control.push(raw.into()); } /// Queue a MAX_REQUEST_ID granting the peer room up to `max`. fn send_max_request_id(&self, max: RequestId) { - let mut raw = Vec::new(); - let mut w = crate::coding::Encoder::new(&mut raw, self.version.into()); - let msg = ietf::MaxRequestId { request_id: max }; - if let Err(err) = w - .varint(ietf::MaxRequestId::ID) - .and_then(|()| crate::coding::Encode::encode(&msg, &mut w, self.version)) - { + if let Err(err) = self.queue_message(&ietf::MaxRequestId { request_id: max }, self.version) { tracing::warn!(%err, "failed to encode MAX_REQUEST_ID"); - return; } - // A closed control stream means the session is ending; there is no one to grant. - let _ = self.shared.control.push(raw.into()); + } + + fn queue_message(&self, msg: &T, version: Version) -> Result<(), Error> { + use crate::coding::Encode; + let mut raw = Vec::new(); + let mut encoder = Encoder::new(&mut raw, version.into()); + encoder.varint(T::ID)?; + msg.encode(&mut encoder, version)?; + let selector = crate::coding::varint::size(T::ID, version.into()).map_err(crate::coding::EncodeError::from)?; + self.stats + .control(super::message::control(T::ID, version), raw.len() - selector, false); + if !self.shared.control.push(raw.into()) { + tracing::debug!("control stream closed; message not sent"); + } + Ok(()) } /// Writer task: drains the queue and writes to the control stream. The peer's @@ -1041,6 +1031,9 @@ impl ControlStreamAdapter { .map_err(super::session::control_stream_error)? .0; + self.stats + .control(super::message::control(type_id, self.version), body.len() + 2, true); + // Reconstruct raw message bytes: [type_id][size][body] let raw = encode_raw(type_id, &body, self.version); @@ -2179,6 +2172,69 @@ mod tests { (adapter, Reader::new(recv, VERSION), writer, log) } + #[moq_clock::test] + async fn legacy_multiplexing_counts_a_control_body_once() { + use crate::{ + coding::{Encode, Stream}, + lite::test_transport::{Log, ScriptedSession, SinkSend}, + stats::{Bytes, Control as Class, Registry, Tier}, + }; + for version in [Version::Draft14, Version::Draft15, Version::Draft16] { + let registry = Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session(""); + let msg = ietf::FetchCancel { + request_id: RequestId(0), + }; + let body = msg.encode_bytes(version).unwrap(); + let raw_response = encode_raw(ietf::FetchCancel::ID, &body.slice(2..), version); + let mut session = ScriptedSession::new(raw_response.to_vec()); + let (_, recv) = crate::transport::poll::Session::open_bi(&mut session, Default::default()) + .await + .unwrap(); + let adapter = ControlStreamAdapter::new(session, Control::new(None, true), version).with_stats(&stats); + let mut virtual_stream = super::super::request_stream::with_stats( + Stream::open(&mut adapter.clone(), version).await.unwrap(), + &stats, + version, + ); + let msg = ietf::FetchCancel { + request_id: RequestId(0), + }; + let size = msg.encode_bytes(version).unwrap().len() as u64; + virtual_stream.writer.buffer_varint(ietf::FetchCancel::ID).unwrap(); + virtual_stream.writer.encode(&msg).await.unwrap(); + let raw = adapter.shared.control.pop().await.unwrap(); + let log = Log::default(); + let mut physical = Writer::new(SinkSend::new(log.clone()), version); + physical.write_all(&mut raw.clone()).await.unwrap(); + assert_eq!( + registry.snapshot().control(), + vec![(Tier::default(), Class::Fetch, Bytes { tx: size, rx: 0 })], + "{version:?}" + ); + assert_eq!(*log.writes.lock().unwrap(), raw); + let (_, goaway) = crate::goaway::Handle::new(false); + let mut reading = std::pin::pin!(adapter.run_read(Reader::new(recv, version), goaway)); + assert!(futures::poll!(reading.as_mut()).is_pending()); + assert_eq!(virtual_stream.reader.varint().await.unwrap(), ietf::FetchCancel::ID); + virtual_stream.reader.decode::().await.unwrap(); + assert_eq!( + registry.snapshot().control(), + vec![(Tier::default(), Class::Fetch, Bytes { tx: size, rx: size })], + "virtual decode counted a physical frame twice: {version:?}" + ); + adapter.send_max_request_id(RequestId(123)); + assert!( + registry + .snapshot() + .control() + .iter() + .any(|(_, class, bytes)| *class == Class::Other && bytes.tx > 0), + "adapter-generated messages bypassed the typed writer" + ); + } + } + #[moq_clock::test] async fn a_legacy_control_stream_fin_or_reset_is_a_protocol_violation() { use crate::lite::test_transport::{DeadStreamSession, ScriptedSession}; diff --git a/rs/moq-net/src/ietf/auth.rs b/rs/moq-net/src/ietf/auth.rs index 71ecc4097f0..d698d503af3 100644 --- a/rs/moq-net/src/ietf/auth.rs +++ b/rs/moq-net/src/ietf/auth.rs @@ -253,6 +253,7 @@ pub(super) fn peer_grant( /// says it negotiated the extension. Never returns: a session without it just fails /// the tokens as unsupported. #[cfg(feature = "async")] +#[allow(clippy::too_many_arguments)] pub(super) async fn run_present( runtime: crate::time::Deadlines, session: S, @@ -261,6 +262,7 @@ pub(super) async fn run_present( peer_setup: peer::PeerSetup, version: Version, going_away: crate::goaway::GoingAway, + stats: crate::stats::Session, ) -> Result<(), Error> where S: crate::transport::poll::Boxable, @@ -285,6 +287,7 @@ where .await? { let present = Present { + stats: stats.clone(), runtime: runtime.clone(), session: session.clone(), control: control.clone(), @@ -320,6 +323,7 @@ where /// One token's Auth request. #[cfg(feature = "async")] struct Present { + stats: crate::stats::Session, runtime: crate::time::Deadlines, session: S, control: Control, @@ -341,6 +345,7 @@ impl Present { Ok(stream) => stream, Err(err) => return err, }; + stream = super::request_stream::with_stats(stream, &self.stats, self.version); if let Err(err) = stream.writer.encode_message(&Auth { request_id, token }).await { return err; } diff --git a/rs/moq-net/src/ietf/message.rs b/rs/moq-net/src/ietf/message.rs index e40360826f1..0ff229f5ceb 100644 --- a/rs/moq-net/src/ietf/message.rs +++ b/rs/moq-net/src/ietf/message.rs @@ -2,6 +2,43 @@ use crate::coding::{Decode, DecodeError, Decoder, Encode, EncodeError, Encoder}; use super::Version; +pub(super) fn control(id: u64, version: Version) -> crate::stats::Control { + use super::*; + use crate::stats::Control; + // Draft 14 reuses later drafts' NAMESPACE_DONE type for TRACK_STATUS_OK. + if version == Version::Draft14 && (id == TrackStatusOk::ID_14 || id == TRACK_STATUS_ERROR_14) { + return Control::TrackInfo; + } + // Before draft 16, 0x02 updates only a subscription; later it is the shared REQUEST_UPDATE. + if matches!(version, Version::Draft14 | Version::Draft15) && id == SubscribeUpdate::ID { + return Control::Subscribe; + } + match id { + PublishNamespace::ID + | PublishNamespaceError::ID + | PublishNamespaceDone::ID + | PublishNamespaceCancel::ID + | SubscribeNamespace::ID + | SubscribeNamespaceLegacy::ID + | SubscribeNamespaceOk::ID + | SubscribeNamespaceError::ID + | UnsubscribeNamespace::ID + | NamespaceDone::ID => Control::Announce, + TrackStatus::ID | SubscribeTracks::ID | PublishSkipped::ID => Control::TrackInfo, + Subscribe::ID + | SubscribeOk::ID + | Unsubscribe::ID + | Publish::ID + | PublishOk::ID + | PublishError::ID + | PublishDone::ID => Control::Subscribe, + Fetch::ID | FetchOk::ID | FetchError::ID | FetchCancel::ID => Control::Fetch, + auth::Auth::ID | auth::AuthOk::ID | auth::AuthError::ID => Control::Auth, + GoAway::ID => Control::Goaway, + _ => Control::Other, + } +} + /// A trait for IETF messages that are automatically size-prefixed during encoding/decoding. /// /// IETF messages use a u16 size prefix and have a message type ID for control stream dispatch. @@ -20,7 +57,9 @@ impl Encode for T { tracing::trace!(?self, "encoding"); let prefix = w.prefix_u16(); self.encode_msg(w, version)?; - w.fill(prefix) + w.fill(prefix)?; + w.control = Some(control(T::ID, version)); + Ok(()) } } @@ -72,6 +111,35 @@ impl Decode for T { Ok(msg) => tracing::trace!(?msg, "decoded"), Err(err) => tracing::warn!(%err, "decode failed"), } + if result.is_ok() { + r.control = Some(control(T::ID, version)); + } result } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::stats::Control; + + /// Draft 14's track-status replies reuse type IDs later drafts give other classes. + #[test] + fn draft14_track_status_ids_are_track_info() { + for id in [super::super::TrackStatusOk::ID_14, super::super::TRACK_STATUS_ERROR_14] { + assert_eq!(control(id, Version::Draft14), Control::TrackInfo); + } + assert_eq!( + control(super::super::NamespaceDone::ID, Version::Draft16), + Control::Announce + ); + assert_eq!( + control(super::super::SubscribeUpdate::ID, Version::Draft15), + Control::Subscribe + ); + // Shared replies and updates identify no request class. + for id in [super::super::RequestOk::ID, super::super::SubscribeUpdate::ID] { + assert_eq!(control(id, Version::Draft16), Control::Other); + } + } +} diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 61a96af3a0d..92b7226d376 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -3668,6 +3668,10 @@ where } } + pub(super) fn stats(&self) -> &crate::stats::Session { + self.origin.stats() + } + /// Open a stream for one advertisement, or `None` if the peer did not give us one in /// time. /// @@ -3681,7 +3685,13 @@ where kio::wait(|waiter| { if let Poll::Ready(res) = waiter.poll_future(open.as_mut()) { - return Poll::Ready(res.map(Some)); + return Poll::Ready(res.map(|stream| { + Some(super::request_stream::with_stats( + stream, + self.origin.stats(), + self.version, + )) + })); } if timeout.poll(waiter).is_ready() { return Poll::Ready(Ok(None)); diff --git a/rs/moq-net/src/ietf/publisher/tracks.rs b/rs/moq-net/src/ietf/publisher/tracks.rs index e3a6a4144c3..82bf92bbc16 100644 --- a/rs/moq-net/src/ietf/publisher/tracks.rs +++ b/rs/moq-net/src/ietf/publisher/tracks.rs @@ -748,7 +748,7 @@ where let Some(opened) = opened else { return Ok(()); }; - let mut stream = opened?; + let mut stream = super::super::request_stream::with_stats(opened?, self.stats(), self.version); let allocated = kio::wait(|waiter| { if listed.poll_closed(waiter).is_ready() || live.poll_closed(waiter).is_ready() { return Poll::Ready(None); diff --git a/rs/moq-net/src/ietf/request_stream.rs b/rs/moq-net/src/ietf/request_stream.rs index fbf2f3134b4..4acb53ae5b5 100644 --- a/rs/moq-net/src/ietf/request_stream.rs +++ b/rs/moq-net/src/ietf/request_stream.rs @@ -6,6 +6,20 @@ use crate::{ coding::{Decode, DecodeError, Decoder, Stream}, }; +/// Legacy receives are framed once by the physical multiplexer, not again by +/// the virtual request decoder. Encodes still count at their typed codec. +pub(super) fn with_stats( + stream: Stream, + stats: &crate::stats::Session, + version: Version, +) -> Stream { + let mut stream = stream.with_stats(stats); + if matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) { + stream.reader = stream.reader.with_stats(&Default::default()); + } + stream +} + pub(super) fn fin_cancels(version: Version) -> bool { matches!( version, @@ -130,8 +144,12 @@ pub(super) enum FollowUp { impl Decode for FollowUp { fn decode(r: &mut Decoder<'_>, version: Version) -> Result { - let update = r.varint()? == super::SubscribeUpdate::ID; + let id = r.varint()?; + let update = id == super::SubscribeUpdate::ID; + let start = r.position(); let body = super::Body::decode(r, version)?; + r.control = Some(super::message::control(id, version)); + r.uncounted = start; Ok(match update { true => Self::Update(body), false => Self::End, @@ -147,7 +165,11 @@ pub(super) struct Framed(pub(super) u64, pub(super) super::Body); impl Decode for Framed { fn decode(r: &mut Decoder<'_>, version: Version) -> Result { let id = r.varint()?; - Ok(Self(id, super::Body::decode(r, version)?)) + let start = r.position(); + let body = super::Body::decode(r, version)?; + r.control = Some(super::message::control(id, version)); + r.uncounted = start; + Ok(Self(id, body)) } } @@ -171,6 +193,7 @@ impl> Decode for Next { let mut peek = r.clone(); if peek.varint()? == super::GoAway::ID { *r = peek; + r.uncounted = r.position(); let msg = super::GoAway::decode(r, version)?; // Draft-18's Request ID belongs on the control stream only. if msg.request_id.is_some() { @@ -472,10 +495,14 @@ impl Decode for Update { if r.varint()? != super::SubscribeUpdate::ID { return Err(DecodeError::InvalidValue); } + let start = r.position(); let size = r.u16()? as usize; let mut data = r.sub(size)?; // The complete frame is present; a short field inside it is malformed. - Self::decode_body(&mut data, version).map_err(DecodeError::complete) + let update = Self::decode_body(&mut data, version).map_err(DecodeError::complete)?; + r.control = Some(super::message::control(super::SubscribeUpdate::ID, version)); + r.uncounted = start; + Ok(update) } } @@ -506,6 +533,7 @@ impl Decode for NamespaceUpdate { if r.varint()? != super::SubscribeUpdate::ID { return Err(DecodeError::InvalidValue); } + let start = r.position(); let size = r.u16()? as usize; let mut data = r.sub(size)?; let mut decode = || { @@ -525,7 +553,10 @@ impl Decode for NamespaceUpdate { refused: super::token::single(token).err().map(Into::into), }) }; - decode().map_err(DecodeError::complete) + let update = decode().map_err(DecodeError::complete)?; + r.control = Some(super::message::control(super::SubscribeUpdate::ID, version)); + r.uncounted = start; + Ok(update) } } @@ -551,6 +582,27 @@ mod tests { } } + /// A request-stream message counts as its sender counts it: the length and body, not the + /// type that selects it, including a GOAWAY read ahead of the response. + #[moq_clock::test] + async fn request_stream_messages_count_without_their_type() { + use crate::lite::test_transport::ScriptedSession; + use crate::stats::{Control, Registry, Tier}; + let registry = Registry::new(Default::default()); + let stats = registry.tier(Tier::default()).session(""); + let version = Version::Draft22; + let mut session = ScriptedSession::new(vec![0x10, 0, 2, 0, 0, 0x07, 0, 0]); + let mut stream = with_stats(Stream::open(&mut session, version).await.unwrap(), &stats, version); + let mut messages = Messages::new(session.clone(), true); + assert_eq!(messages.message(&mut stream.reader).await.unwrap().0, 0x07); + let rx = |class| { + let rows = registry.snapshot().control(); + rows.into_iter().find(|row| row.1 == class).map(|row| row.2.rx) + }; + assert_eq!(rx(Control::Goaway), Some(4)); + assert_eq!(rx(Control::Other), Some(2)); + } + #[moq_clock::test] async fn a_goaway_does_not_cancel_a_pending_request() { use crate::lite::test_transport::ScriptedSession; diff --git a/rs/moq-net/src/ietf/session.rs b/rs/moq-net/src/ietf/session.rs index f2c5fde93c1..ca2e15e11d3 100644 --- a/rs/moq-net/src/ietf/session.rs +++ b/rs/moq-net/src/ietf/session.rs @@ -19,6 +19,9 @@ pub struct Config { /// The runtime that arms the session's timers. pub runtime: crate::time::Deadlines, + /// Where the session's control traffic is counted, even with neither origin set. + pub stats: crate::stats::Session, + pub session: S, /// The bidi SETUP stream (draft-14 through draft-16 only). Draft-17+ passes `None` @@ -115,6 +118,7 @@ where { let Config { runtime, + stats, mut session, setup, request_id_max, @@ -206,8 +210,12 @@ where // moq-transport threads concrete origins through the publisher/subscriber. // An unset half gets an empty origin: an empty publish origin announces // nothing, and an empty subscribe origin issues no SUBSCRIBE_NAMESPACE. - let publish = publish.unwrap_or_else(|| origin::Producer::empty(Hop::random()).consume()); - let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random())); + let publish = publish.unwrap_or_else(|| { + origin::Producer::empty(Hop::random()) + .consume() + .with_stats(stats.clone()) + }); + let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random()).with_stats(stats.clone())); let subscriptions = crate::session::Slots::new(limits.subscriptions) .with_stats(publish.stats(), crate::stats::Cap::Subscriptions); let announces = @@ -224,7 +232,7 @@ where return Err(err); }; let control = Control::new(request_id_max, client).with_window(limits.requests(), client, version); - let adapter = ControlStreamAdapter::new(session.clone(), control.clone(), version); + let adapter = ControlStreamAdapter::new(session.clone(), control.clone(), version).with_stats(&stats); // No AUTH on these drafts, but a limit still reaches both halves. let mut publisher = Publisher::new( @@ -407,6 +415,7 @@ where }, goaway.clone(), control.clone(), + stats.clone(), )); let mut publisher = Publisher::new( @@ -465,6 +474,7 @@ where peer_setup.clone(), version, goaway.going_away.clone(), + stats.clone(), ); // Only a side that offered MoQ Auth answers the peer's Auth requests: one that // arrives otherwise falls through to the protocol violation in `run_dispatch`. @@ -483,9 +493,13 @@ where // arrives on the wire. let goaway_recv = { let goaway = goaway.clone(); + let stats = stats.clone(); async move { match peer_setup_stream { - Some(reader) => run_goaway(reader.with_version(version), version, client, goaway).await, + Some(reader) => { + run_goaway(reader.with_stats(&stats).with_version(version), version, client, goaway) + .await + } None => std::future::pending().await, } } @@ -652,6 +666,9 @@ pub struct PeerSetup { /// The SETUP stream, which becomes the GOAWAY channel. pub stream: Reader, + /// The SETUP's encoded size, counted once the session's stats are known. + pub size: usize, + /// The request path the peer advertised, for URL-less transports. pub path: Option, @@ -712,7 +729,7 @@ pub async fn accept_setup( } reader.set_priority(crate::priority::RECV_CONTROL); - let setup: setup::Setup = reader.decode().await?; + let (setup, size) = reader.decode_sized::().await?; let (params, _) = ietf::Parameters::decode_slice(&setup.parameters, version)?; let path = match params.get_bytes(ietf::ParameterBytes::Path) { Some(bytes) => Some( @@ -727,6 +744,7 @@ pub async fn accept_setup( return Ok(PeerSetup { stream: reader, + size, path, token, declared, @@ -785,6 +803,7 @@ async fn run_setup( extensions: crate::setup::Extensions, goaway: crate::goaway::Protocol, control: Control, + stats: crate::stats::Session, ) -> Result<(), Error> { let outer_version = crate::Version::Ietf(version); @@ -792,7 +811,7 @@ async fn run_setup( .open_uni(Default::default()) .await .map_err(Error::from_transport)?; - let mut writer: Writer = Writer::new(send, outer_version); + let mut writer: Writer = Writer::new(send, outer_version).with_stats(&stats); let mut parameters = ietf::Parameters::default(); parameters.set_bytes(ietf::ParameterBytes::Implementation, b"moq-lite-rs".to_vec()); @@ -1041,6 +1060,7 @@ struct Uni { impl Uni { /// Read the stream's type and serve it, failing only when the stream breaks the session. async fn serve(self, mut reader: Reader) -> Result<(), Error> { + reader = reader.with_stats(self.subscriber.stats()); let Self { mut subscriber, peer_setup, @@ -1286,6 +1306,7 @@ struct Bidi { impl Bidi { /// Read and serve a request, failing only when it breaks the session. async fn serve(self, mut stream: Stream) -> Result<(), Error> { + stream = super::request_stream::with_stats(stream, self.publisher.stats(), self.version); let Self { publisher, mut subscriber, @@ -1315,6 +1336,14 @@ impl Bidi { } Err(err) => return Err(err), }; + let native = !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) + || (version == Version::Draft16 && id == ietf::SubscribeNamespaceLegacy::ID); + if native { + stream.reader = stream.reader.with_stats(publisher.stats()); + publisher + .stats() + .control(super::message::control(id, version), data.0.len() + 2, true); + } if !matches!(version, Version::Draft14 | Version::Draft15 | Version::Draft16) && matches!( id, @@ -1450,6 +1479,9 @@ async fn read_goaways( } let msg: ietf::GoAway = body.message(version)?; + reader + .stats() + .control(crate::stats::Control::Goaway, body.0.len() + 2, true); tracing::info!(message = ?msg, "received GOAWAY"); // Draft-18 names the first of our own Request IDs it did not process: even for a @@ -1577,6 +1609,7 @@ mod tests { let log = session.log.clone(); let (driver, _) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -1632,6 +1665,7 @@ mod tests { let log = session.log.clone(); let (driver, _) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -1678,6 +1712,7 @@ mod tests { let log = session.log.clone(); let (driver, _) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -1773,6 +1808,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -1836,6 +1872,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -1909,6 +1946,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: Some(setup), request_id_max: None, @@ -1997,6 +2035,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2100,6 +2139,7 @@ mod tests { let handle = crate::auth::Handle::ietf(true); let (driver, goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2184,6 +2224,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2280,6 +2321,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2369,6 +2411,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2412,6 +2455,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2678,6 +2722,7 @@ mod tests { fn bidi_driver(session: crate::lite::test_transport::ScriptedSession, publish: Option) -> Driver { start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2821,6 +2866,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2889,6 +2935,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -2991,6 +3038,7 @@ mod tests { let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: Some(setup), request_id_max: None, @@ -3113,6 +3161,7 @@ mod tests { let origin = crate::origin::Config::new(crate::Hop::new(1).unwrap()).produce(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup, request_id_max: None, @@ -3344,6 +3393,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, @@ -3423,6 +3473,7 @@ mod tests { Default::default(), goaway_protocol, control.clone(), + Default::default(), )); handle.producer().send(crate::goaway::Goaway::new()).unwrap(); assert!(futures::poll!(setup.as_mut()).is_pending()); @@ -3445,6 +3496,7 @@ mod tests { let log = session.log.clone(); let (driver, _goaway) = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup: None, request_id_max: None, diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 0c0ba532556..f19b317eef3 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -1144,6 +1144,10 @@ where Some(track) } + pub(super) fn stats(&self) -> &crate::stats::Session { + self.origin.stats() + } + /// Send SUBSCRIBE_NAMESPACE for one prefix on a bidi stream. /// The caller is responsible for opening the appropriate stream type /// (virtual for v14/v15, real bidi for v16+), one per prefix. @@ -1161,6 +1165,11 @@ where prefix: PathOwned, snapshot: crate::model::Snapshot, ) -> Result<(), Error> { + stream = if self.version == Version::Draft16 { + stream.with_stats(self.origin.stats()) + } else { + super::request_stream::with_stats(stream, self.origin.stats(), self.version) + }; // Hidden namespaces are requested too, as on moq-lite: the session mirrors the // peer into the origin and each local reader opts in on its own. The parameter // fails decoding at a peer that doesn't know it, so it waits on the peer's SETUP @@ -2885,7 +2894,7 @@ where }) .await { - Ok(stream) => stream, + Ok(stream) => super::request_stream::with_stats(stream, self.origin.stats(), self.version), Err(err) => { target.fail(err); return None; @@ -3897,6 +3906,7 @@ where Stream::poll_open(&mut opening, this.version, &mut waiter.context()) }) .await?; + stream = super::request_stream::with_stats(stream, this.origin.stats(), this.version); let mut timeout = crate::time::Deadline::after(&this.runtime, Control::REQUEST_ID_TIMEOUT); let fetch_id = kio::wait(|waiter| { validity.check(waiter)?; @@ -4098,6 +4108,7 @@ where return None; } }; + stream = super::request_stream::with_stats(stream, self.origin.stats(), self.version); let mut timeout = crate::time::Deadline::after(&self.runtime, Control::REQUEST_ID_TIMEOUT); let request_id = match kio::wait(|waiter| { if gate.poll_denied(waiter).is_ready() { @@ -5194,6 +5205,7 @@ where Ok(stream) => stream, Err(err) => return request.reject(err), }; + stream = super::request_stream::with_stats(stream, self.origin.stats(), self.version); let mut timeout = crate::time::Deadline::after(&self.runtime, Control::REQUEST_ID_TIMEOUT); let fetch_id = match kio::wait(|waiter| { validity.check(waiter)?; diff --git a/rs/moq-net/src/ietf/subscriber/tracks.rs b/rs/moq-net/src/ietf/subscriber/tracks.rs index 22a4417acac..63af5130715 100644 --- a/rs/moq-net/src/ietf/subscriber/tracks.rs +++ b/rs/moq-net/src/ietf/subscriber/tracks.rs @@ -443,7 +443,7 @@ where let Some(stream) = opened else { return Ok(()); }; - let mut stream = stream?; + let mut stream = request_stream::with_stats(stream?, self.origin.stats(), self.version); // No draft with SUBSCRIBE_TRACKS has MAX_REQUEST_ID, so this never waits on credit. let request_id = kio::wait(|waiter| self.control.poll_next_request_id(waiter)).await; stream.writer.buffer_varint(ietf::SubscribeTracks::ID)?; @@ -641,7 +641,7 @@ where let Some(stream) = opened else { return Ok(None); }; - let mut stream = stream?; + let mut stream = request_stream::with_stats(stream?, self.origin.stats(), self.version); let request_id = kio::wait(|waiter| self.control.poll_next_request_id(waiter)).await; self.state.lock().subscribes.insert( request_id, diff --git a/rs/moq-net/src/lite/announce.rs b/rs/moq-net/src/lite/announce.rs index ff927379911..8cad993a1c3 100644 --- a/rs/moq-net/src/lite/announce.rs +++ b/rs/moq-net/src/lite/announce.rs @@ -244,6 +244,7 @@ impl Decode for Cost { impl Encode for AnnounceBroadcast<'_> { fn encode(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::Announce); if version.has_announce_id() { // Lite06+: outer type discriminator, then a size-prefixed body (like the // subscribe stream). The body varies by type. @@ -332,6 +333,7 @@ impl Encode for AnnounceBroadcast<'_> { impl Decode for AnnounceBroadcast<'_> { fn decode(buf: &mut Decoder<'_>, version: Version) -> Result { + buf.control = Some(crate::stats::Control::Announce); if version.has_announce_id() { // Lite06+: outer type, then a size-prefixed body decoded within its bounds. let typ = buf.varint()?; @@ -455,6 +457,7 @@ pub struct AnnounceRequest<'a> { } impl Message for AnnounceRequest<'_> { + const CONTROL: Option = Some(crate::stats::Control::Announce); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { let prefix = Path::decode(r, version)?; let exclude_hop = match version.has_exclude_hop() { @@ -499,6 +502,7 @@ enum AnnounceStatus { impl Decode for AnnounceStatus { fn decode(r: &mut Decoder<'_>, _: Version) -> Result { + r.control = Some(crate::stats::Control::Announce); let status = r.u8()?; status.try_into().map_err(|_| DecodeError::InvalidValue) } @@ -506,6 +510,7 @@ impl Decode for AnnounceStatus { impl Encode for AnnounceStatus { fn encode(&self, w: &mut Encoder<'_>, _: Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::Announce); w.u8(*self as u8); Ok(()) } @@ -521,6 +526,7 @@ pub struct AnnounceInit<'a> { } impl Message for AnnounceInit<'_> { + const CONTROL: Option = Some(crate::stats::Control::Announce); // Published lite01/02 carry the whole initial set in one message. const MAX_SIZE: usize = 64 * 1024 * 1024; @@ -575,6 +581,7 @@ pub struct AnnounceOk { } impl Message for AnnounceOk { + const CONTROL: Option = Some(crate::stats::Control::Announce); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_announce_ok() { return Err(DecodeError::Version); diff --git a/rs/moq-net/src/lite/auth.rs b/rs/moq-net/src/lite/auth.rs index 1a072b24ff9..39e302025a6 100644 --- a/rs/moq-net/src/lite/auth.rs +++ b/rs/moq-net/src/lite/auth.rs @@ -17,6 +17,7 @@ pub struct Auth { } impl Message for Auth { + const CONTROL: Option = Some(crate::stats::Control::Auth); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_auth() { return Err(DecodeError::Version); @@ -75,6 +76,7 @@ fn decode_patterns(r: &mut Decoder<'_>) -> Result { } impl Message for AuthOk { + const CONTROL: Option = Some(crate::stats::Control::Auth); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_auth() { return Err(DecodeError::Version); @@ -123,6 +125,7 @@ pub struct AuthError { const MAX_REASON: usize = 8192; impl Message for AuthError { + const CONTROL: Option = Some(crate::stats::Control::Auth); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_auth() { return Err(DecodeError::Version); diff --git a/rs/moq-net/src/lite/fetch.rs b/rs/moq-net/src/lite/fetch.rs index df57064f947..a793517dd8d 100644 --- a/rs/moq-net/src/lite/fetch.rs +++ b/rs/moq-net/src/lite/fetch.rs @@ -29,6 +29,7 @@ pub struct Fetch<'a> { } impl Message for Fetch<'_> { + const CONTROL: Option = Some(crate::stats::Control::Fetch); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 => { diff --git a/rs/moq-net/src/lite/goaway.rs b/rs/moq-net/src/lite/goaway.rs index f142a50ec7e..2f93e59842e 100644 --- a/rs/moq-net/src/lite/goaway.rs +++ b/rs/moq-net/src/lite/goaway.rs @@ -13,6 +13,7 @@ pub struct Goaway<'a> { } impl Message for Goaway<'_> { + const CONTROL: Option = Some(crate::stats::Control::Goaway); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 | Version::Lite03 => { diff --git a/rs/moq-net/src/lite/ietf.rs b/rs/moq-net/src/lite/ietf.rs index 83b88b2843a..8b987ae0f37 100644 --- a/rs/moq-net/src/lite/ietf.rs +++ b/rs/moq-net/src/lite/ietf.rs @@ -317,6 +317,7 @@ impl Records { pub struct TrackProperties(pub Option>); impl Message for TrackProperties { + const CONTROL: Option = Some(crate::stats::Control::TrackInfo); fn encode_msg(&self, w: &mut Encoder<'_>, _: Version) -> Result<(), EncodeError> { // A track published over moq-lite is a lone 0. let Some(properties) = &self.0 else { diff --git a/rs/moq-net/src/lite/info.rs b/rs/moq-net/src/lite/info.rs index 301764fdcaf..687bd675829 100644 --- a/rs/moq-net/src/lite/info.rs +++ b/rs/moq-net/src/lite/info.rs @@ -8,6 +8,7 @@ pub struct SessionInfo { } impl Message for SessionInfo { + const CONTROL: Option = Some(crate::stats::Control::Session); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 => {} diff --git a/rs/moq-net/src/lite/message.rs b/rs/moq-net/src/lite/message.rs index b38b2a910e9..1b4ede3e2ca 100644 --- a/rs/moq-net/src/lite/message.rs +++ b/rs/moq-net/src/lite/message.rs @@ -23,6 +23,8 @@ pub(super) fn decode_size(r: &mut Decoder<'_>, max: usize) -> Result = None; /// The largest body this receiver accepts for this message. const MAX_SIZE: usize = MAX_MESSAGE_SIZE; @@ -43,7 +45,9 @@ impl Encode for T { w.discard(prefix); return Err(EncodeError::TooLarge); } - w.fill(prefix) + w.fill(prefix)?; + w.control = T::CONTROL; + Ok(()) } } @@ -64,6 +68,9 @@ impl Decode for T { Ok(msg) => tracing::trace!(?msg, "decoded"), Err(err) => tracing::warn!(%err, "decode failed"), } + if result.is_ok() { + r.control = T::CONTROL; + } result } } diff --git a/rs/moq-net/src/lite/probe.rs b/rs/moq-net/src/lite/probe.rs index 812d1f8362c..041bd77d188 100644 --- a/rs/moq-net/src/lite/probe.rs +++ b/rs/moq-net/src/lite/probe.rs @@ -13,6 +13,7 @@ pub struct Probe { } impl Message for Probe { + const CONTROL: Option = Some(crate::stats::Control::Probe); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 => { diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 0b4a3aa7909..37c51fa9504 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -237,6 +237,7 @@ where loop { match Stream::poll_accept(&mut self.accept, self.shared.version, &mut cx) { Poll::Ready(Ok(stream)) => { + let stream = stream.with_stats(self.shared.origin.stats()); self.children.push(Control { shared: self.shared.clone(), runtime: self.runtime.clone(), diff --git a/rs/moq-net/src/lite/session.rs b/rs/moq-net/src/lite/session.rs index 703b58f7924..ef8cb2fce8b 100644 --- a/rs/moq-net/src/lite/session.rs +++ b/rs/moq-net/src/lite/session.rs @@ -111,9 +111,9 @@ impl SetupAccept { // before reporting the terminal error from accepting the next stream. let result = setup .as_mut() - .map(|reader| reader.poll_decode::(&mut cx)) + .map(|reader| reader.poll_decode_sized::(&mut cx)) .unwrap_or(Poll::Pending); - let setup = match result { + let (setup, size) = match result { Poll::Ready(result) => result?, Poll::Pending => match accept_error { Some(err) => return Poll::Ready(Err(err)), @@ -128,6 +128,7 @@ impl SetupAccept { early.append(pending); Poll::Ready(Ok(AcceptedSetup { setup, + size, early: std::mem::take(early), })) } @@ -137,6 +138,8 @@ impl SetupAccept { pub struct AcceptedSetup { /// The SETUP decoded before origin binding. pub setup: Setup, + /// The SETUP's encoded size, counted once the session's stats are known. + pub size: usize, /// Streams with their type still buffered for the session classifier. pub early: Vec>, } @@ -146,6 +149,9 @@ pub struct Config { /// The runtime that arms the session's timers. pub runtime: crate::time::Deadlines, + /// Where the session's control traffic is counted, even with neither origin set. + pub stats: crate::stats::Session, + /// Whether we dialed the session. Only the dialing side aborts on a publication /// its grant does not cover: a server's publish origin is everything the peer /// may read, not what it intends to push. @@ -201,6 +207,7 @@ where { let Config { runtime, + stats, client, session, setup_stream, @@ -254,8 +261,12 @@ where // An unset half gets an empty origin: an empty publish origin announces nothing // (and answers the peer's announce-interest with an empty set), and an empty // subscribe origin issues no ANNOUNCE_PLEASE. - let publish = publish.unwrap_or_else(|| origin::Producer::empty(Hop::random()).consume()); - let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random())); + let publish = publish.unwrap_or_else(|| { + origin::Producer::empty(Hop::random()) + .consume() + .with_stats(stats.clone()) + }); + let subscribe = subscribe.unwrap_or_else(|| origin::Producer::empty(Hop::random()).with_stats(stats.clone())); let subscriptions = crate::session::Slots::new(limits.subscriptions).with_stats(publish.stats(), crate::stats::Cap::Subscriptions); let announces = @@ -329,6 +340,7 @@ where let driver = Driver { auth: Present { + stats: stats.clone(), runtime: runtime.clone(), session: session.clone(), version, @@ -342,9 +354,9 @@ where local_close: Default::default(), setup: version .has_setup_stream() - .then(|| SendSetup::new(session.clone(), our_setup, version)), - goaway: Some(SendGoaway::new(session.clone(), goaway, version)), - session_stream: setup_stream, + .then(|| SendSetup::new(session.clone(), our_setup, version, stats.clone())), + goaway: Some(SendGoaway::new(session.clone(), goaway, version, stats.clone())), + session_stream: setup_stream.map(|stream| stream.with_stats(&stats)), ending: None, publisher, subscriber: SubscriberDriver::new(subscriber, early, client), @@ -484,6 +496,7 @@ impl Drop for Driver { /// Opens one AUTH stream per token this side presents. struct Present { + stats: crate::stats::Session, runtime: crate::time::Deadlines, session: S, version: Version, @@ -509,6 +522,7 @@ impl Present { } while let Poll::Ready(Some((id, token))) = self.handle.poll_opening(waiter) { self.tokens.push(PresentToken { + stats: self.stats.clone(), runtime: self.runtime.clone(), session: self.session.clone(), version: self.version, @@ -531,6 +545,7 @@ impl Present { /// One token's AUTH stream: send the token, then track the grant until either /// side ends it. struct PresentToken { + stats: crate::stats::Session, runtime: crate::time::Deadlines, session: S, version: Version, @@ -595,6 +610,7 @@ impl PresentToken { Ok(stream) => stream, Err(err) => return Poll::Ready(err), }; + stream = stream.with_stats(&self.stats); let res = stream .writer .buffer(&super::ControlType::Auth) @@ -686,6 +702,7 @@ fn poll_session_stream( /// error is logged and the machine finishes; the peer falls back to "no /// capabilities" for us. struct SendSetup { + stats: crate::stats::Session, version: Version, state: SendSetupState, } @@ -706,8 +723,9 @@ enum SendSetupState { } impl SendSetup { - fn new(session: S, setup: Setup, version: Version) -> Self { + fn new(session: S, setup: Setup, version: Version, stats: crate::stats::Session) -> Self { Self { + stats, version, state: SendSetupState::Open { session, @@ -731,7 +749,7 @@ impl SendSetup { SendSetupState::Open { session, setup } => { let stream = ready!(session.poll_open_uni(cx, Default::default())) .map_err(crate::lite::error::from_transport)?; - let mut writer = Writer::new(stream, self.version); + let mut writer = Writer::new(stream, self.version).with_stats(&self.stats); writer.buffer(&super::DataType::Setup)?; writer.buffer(&**setup)?; self.state = SendSetupState::Send { @@ -756,6 +774,7 @@ impl SendSetup { /// Send our single GOAWAY when the drain trigger fires. The session supervisor /// enforces the deadline independently, including on versions with no GOAWAY message. struct SendGoaway { + stats: crate::stats::Session, version: Version, goaway: crate::goaway::Protocol, /// Opening a stream and watching the session close each need their own handle. @@ -777,8 +796,9 @@ enum SendGoawayState { } impl SendGoaway { - fn new(session: S, goaway: crate::goaway::Protocol, version: Version) -> Self { + fn new(session: S, goaway: crate::goaway::Protocol, version: Version, stats: crate::stats::Session) -> Self { Self { + stats, version, goaway, closed: session.clone(), @@ -811,6 +831,7 @@ impl SendGoaway { return Poll::Ready(()); } }; + stream = stream.with_stats(&self.stats); let msg = super::Goaway { uri: std::borrow::Cow::Borrowed(payload.uri.as_str()), }; @@ -916,6 +937,7 @@ mod tests { let start = start(Config { runtime: crate::time::sim(), + stats: Default::default(), client: true, limits: Default::default(), session: transport, @@ -948,6 +970,7 @@ mod tests { let log = transport.log.clone(); let start = start(Config { runtime: crate::time::sim(), + stats: Default::default(), client: false, limits: Default::default(), session: transport, @@ -1034,6 +1057,7 @@ mod tests { let log = transport.log.clone(); let start = start(Config { runtime: crate::time::sim(), + stats: Default::default(), client: true, limits: Default::default(), session: transport, @@ -1103,6 +1127,7 @@ mod tests { let log = session.log.clone(); let mut started = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup_stream: None, publish: None, @@ -1157,6 +1182,7 @@ mod tests { let log = session.log.clone(); let mut started = start(Config { runtime: crate::time::sim(), + stats: Default::default(), session, setup_stream: None, publish: None, @@ -1189,6 +1215,7 @@ mod tests { ) -> SessionStart { start(Config { runtime: crate::time::sim(), + stats: Default::default(), client, session: transport, setup_stream: None, @@ -1256,6 +1283,7 @@ mod tests { probe: super::super::ProbeLevel::Report, ..Default::default() }, + size: 0, early: Vec::new(), }; let mut started = raw_session(transport, version, true, Some(peer_setup)); diff --git a/rs/moq-net/src/lite/setup.rs b/rs/moq-net/src/lite/setup.rs index 314807509cb..2cfcb16b144 100644 --- a/rs/moq-net/src/lite/setup.rs +++ b/rs/moq-net/src/lite/setup.rs @@ -190,6 +190,7 @@ pub struct Setup { } impl Message for Setup { + const CONTROL: Option = Some(crate::stats::Control::Session); const MAX_SIZE: usize = crate::setup::MAX_SETUP_SIZE; fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { @@ -240,6 +241,7 @@ pub(super) struct ReceivedSetup { } impl Message for ReceivedSetup { + const CONTROL: Option = Some(crate::stats::Control::Session); const MAX_SIZE: usize = Setup::MAX_SIZE; fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { diff --git a/rs/moq-net/src/lite/subscribe.rs b/rs/moq-net/src/lite/subscribe.rs index 279a07ffdf8..c8b3aca1982 100644 --- a/rs/moq-net/src/lite/subscribe.rs +++ b/rs/moq-net/src/lite/subscribe.rs @@ -53,6 +53,7 @@ impl Version { } impl Message for Subscribe<'_> { + const CONTROL: Option = Some(crate::stats::Control::Subscribe); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { let id = r.varint()?; let broadcast = Path::decode(r, version)?; @@ -268,6 +269,7 @@ pub struct SubscribeOk { } impl Message for SubscribeOk { + const CONTROL: Option = Some(crate::stats::Control::Subscribe); fn encode_msg(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { match version { Version::Lite01 => { @@ -337,6 +339,7 @@ pub struct SubscribeStart { } impl Message for SubscribeStart { + const CONTROL: Option = Some(crate::stats::Control::Subscribe); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_track_stream() { return Err(DecodeError::Version); @@ -381,6 +384,7 @@ pub struct SubscribeEnd { } impl Message for SubscribeEnd { + const CONTROL: Option = Some(crate::stats::Control::Subscribe); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_track_stream() { return Err(DecodeError::Version); @@ -418,6 +422,7 @@ pub struct SubscribeUpdate { } impl Message for SubscribeUpdate { + const CONTROL: Option = Some(crate::stats::Control::Subscribe); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 => { @@ -534,6 +539,7 @@ impl SubscribeDrop { } impl Message for SubscribeDrop { + const CONTROL: Option = Some(crate::stats::Control::Subscribe); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { match version { Version::Lite01 | Version::Lite02 => { diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index 817fb1d9f01..90cd6500cf7 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -740,7 +740,9 @@ impl UniAccept { for reader in early { children.push(UniServe { subscriber: subscriber.clone(), - state: UniState::Start { reader }, + state: UniState::Start { + reader: reader.with_stats(subscriber.origin.stats()), + }, }); } Self { @@ -760,7 +762,8 @@ impl UniAccept { self.children.push(UniServe { subscriber: self.subscriber.clone(), state: UniState::Start { - reader: Reader::new(stream, self.subscriber.version), + reader: Reader::new(stream, self.subscriber.version) + .with_stats(self.subscriber.origin.stats()), }, }); } @@ -1529,6 +1532,7 @@ impl ProbeStream { return Poll::Ready(Ok(())); } let mut stream = ready!(Stream::poll_open(&mut self.session, self.subscriber.version, &mut cx))?; + stream = stream.with_stats(self.subscriber.origin.stats()); stream.writer.buffer(&lite::ControlType::Probe)?; self.state = ProbeState::Send { stream }; } @@ -1636,6 +1640,8 @@ impl AnnouncePrefix { &mut cx ))?; + stream = stream.with_stats(self.subscriber.origin.stats()); + stream.writer.buffer(&lite::ControlType::Announce)?; // Lite04/05: ask the peer to filter out announces that already passed // through us, so the reflected ones never hit the wire. Encoding drops @@ -1890,6 +1896,7 @@ impl TracksRecv { Some(stream) => stream, None => { let mut stream = ready!(Stream::poll_open(&mut self.session, version, &mut cx))?; + stream = stream.with_stats(source.subscriber.origin.stats()); stream.writer.buffer(&lite::ControlType::Tracks)?; stream.writer.buffer(&lite::Tracks { broadcast: source.path.as_path(), @@ -6476,6 +6483,7 @@ impl Establish { end_frame: bounds.end_frame, order: self.subscription.order, }; + stream = stream.with_stats(self.serve.subscriber.origin.stats()); stream.writer.buffer(&lite::ControlType::Subscribe)?; stream.writer.buffer(&msg)?; self.state = EstablishState::Send { stream }; @@ -6791,6 +6799,7 @@ impl TrackInfoFetch { match &mut self.state { TrackInfoState::Open => { let mut stream = ready!(Stream::poll_open(&mut self.session, serve.subscriber.version, &mut cx))?; + stream = stream.with_stats(serve.subscriber.origin.stats()); stream.writer.buffer(&lite::ControlType::Track)?; stream.writer.buffer(&lite::Track { epoch: serve.epoch.clone(), @@ -7602,6 +7611,7 @@ impl kio::Task for FetchServeRun { // downstream cap is applied when serving, not when fetching. end_frame: None, }; + stream = stream.with_stats(self.serve.subscriber.origin.stats()); let buffered = stream .writer .buffer(&lite::ControlType::Fetch) diff --git a/rs/moq-net/src/lite/track.rs b/rs/moq-net/src/lite/track.rs index eeaa1f8a112..1b667a0a664 100644 --- a/rs/moq-net/src/lite/track.rs +++ b/rs/moq-net/src/lite/track.rs @@ -24,6 +24,7 @@ pub struct Track<'a> { } impl Message for Track<'_> { + const CONTROL: Option = Some(crate::stats::Control::TrackInfo); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_track_stream() { return Err(DecodeError::Version); @@ -133,6 +134,7 @@ impl Framing { } impl Message for TrackInfo { + const CONTROL: Option = Some(crate::stats::Control::TrackInfo); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_track_stream() { return Err(DecodeError::Version); diff --git a/rs/moq-net/src/lite/tracks.rs b/rs/moq-net/src/lite/tracks.rs index 5528b29a837..8a249837ec2 100644 --- a/rs/moq-net/src/lite/tracks.rs +++ b/rs/moq-net/src/lite/tracks.rs @@ -23,6 +23,7 @@ pub struct Tracks<'a> { } impl Message for Tracks<'_> { + const CONTROL: Option = Some(crate::stats::Control::TrackInfo); fn decode_msg(r: &mut Decoder<'_>, version: Version) -> Result { if !version.has_tracks() { return Err(DecodeError::Version); @@ -56,6 +57,7 @@ pub enum TracksEvent<'a> { impl Encode for TracksEvent<'_> { fn encode(&self, w: &mut Encoder<'_>, version: Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::TrackInfo); if !version.has_tracks() { return Err(EncodeError::Version); } @@ -83,6 +85,7 @@ impl Encode for TracksEvent<'_> { impl Decode for TracksEvent<'_> { fn decode(r: &mut Decoder<'_>, version: Version) -> Result { + r.control = Some(crate::stats::Control::TrackInfo); if !version.has_tracks() { return Err(DecodeError::Version); } diff --git a/rs/moq-net/src/server.rs b/rs/moq-net/src/server.rs index 1efabc74d38..a76eae490e8 100644 --- a/rs/moq-net/src/server.rs +++ b/rs/moq-net/src/server.rs @@ -176,6 +176,7 @@ impl Server { let start = lite::start(lite::Config { runtime: runtime.clone(), + stats: self.stats.clone(), client: false, limits: self.limits, session: session.clone(), @@ -414,8 +415,9 @@ impl Server { // Legacy bidi SETUP exchange (IETF 14-16, lite 01/02). Read the client's // SETUP to choose the version; `ok()` sends the server SETUP and starts. + // Stats attach in `ok()`: a caller may still replace them on the handshake. let mut stream = Stream::accept(&mut session, encoding).await?; - let client: setup::Client = stream.reader.decode().await?; + let (client, client_setup_size) = stream.reader.decode_sized::().await?; let version = client .versions @@ -476,6 +478,7 @@ impl Server { version, request_id_max, peer_declared, + client_setup_size, })), }), }) @@ -622,12 +625,14 @@ where version, peer_setup, } = *self; + server.stats.control(stats::Control::Session, peer_setup.size, true); let (publish, subscribe) = server.stat_tagged_origins(); // The client's SETUP was read at the pause; hand the stream back // for GOAWAY. A server never advertises a path, hence `None`. let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), + stats: server.stats.clone(), limits: server.limits, session: session.clone(), setup: None, @@ -681,6 +686,8 @@ struct PausedLegacy { request_id_max: Option, /// What the client's SETUP declared, for the options `ok` acts on. peer_declared: ietf::peer::Peer, + /// The client SETUP's encoded size, counted under the final stats. + client_setup_size: usize, } impl Paused for PausedLegacy @@ -700,11 +707,14 @@ where async move { let Self { session, - mut stream, + stream, version, request_id_max, peer_declared, + client_setup_size, } = *self; + let mut stream = stream.with_stats(&server.stats); + server.stats.control(stats::Control::Session, client_setup_size, true); let (publish, subscribe) = server.stat_tagged_origins(); // Encode parameters using the version-appropriate type. @@ -741,6 +751,7 @@ where // Pre-lite-05: no Setup Stream, so nothing to advertise or seed. let start = lite::start(lite::Config { runtime: runtime.clone(), + stats: server.stats.clone(), client: false, limits: server.limits, session: session.clone(), @@ -766,6 +777,7 @@ where // Draft 14-16: path came in the bidi SETUP, no uni SETUP to hand back. let (protocol, goaway) = ietf::start(ietf::Config { runtime: runtime.clone(), + stats: server.stats.clone(), limits: server.limits, session: session.clone(), setup: Some(stream), @@ -932,7 +944,10 @@ where session, version, client_setup, - } => server.start_lite(runtime, session, version, Some(client_setup), peer_hop, auth), + } => { + server.stats.control(stats::Control::Session, client_setup.size, true); + server.start_lite(runtime, session, version, Some(client_setup), peer_hop, auth) + } PausedHandshake::Boxed(paused) => paused.ok(server, runtime, peer_hop, auth).await, } } @@ -1525,6 +1540,7 @@ mod tests { crate::lite::test_transport::PendingRecv, Version::Ietf(version), ), + size: 0, path: None, token: None, declared: ietf::peer::Peer::default(), diff --git a/rs/moq-net/src/setup.rs b/rs/moq-net/src/setup.rs index 1e8eb5dcfd9..97ca1f1317e 100644 --- a/rs/moq-net/src/setup.rs +++ b/rs/moq-net/src/setup.rs @@ -95,6 +95,7 @@ impl Setup { impl Encode for Setup { fn encode(&self, w: &mut Encoder<'_>, v: Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::Session); Self::check_version(v); w.varint(SETUP_V17)?; let prefix = w.prefix_u16(); @@ -105,6 +106,7 @@ impl Encode for Setup { impl Decode for Setup { fn decode(r: &mut Decoder<'_>, v: Version) -> Result { + r.control = Some(crate::stats::Control::Session); Self::check_version(v); let kind = r.varint()?; if kind != SETUP_V17 { @@ -174,6 +176,7 @@ impl Client { impl Decode for Client { /// Decode a client setup message (draft-14 through draft-16 only). fn decode(r: &mut Decoder<'_>, v: Version) -> Result { + r.control = Some(crate::stats::Control::Session); let kind = r.u8()?; if kind != CLIENT_SETUP { return Err(DecodeError::InvalidValue); @@ -202,6 +205,7 @@ impl Decode for Client { impl Encode for Client { /// Encode a client setup message (draft-14 through draft-16 only). fn encode(&self, w: &mut Encoder<'_>, v: Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::Session); w.u8(CLIENT_SETUP); let prefix = prefix_body(w, v)?; self.encode_inner(w, v)?; @@ -275,6 +279,7 @@ impl Server { impl Encode for Server { /// Encode a server setup message (draft-14 through draft-16 only). fn encode(&self, w: &mut Encoder<'_>, v: Version) -> Result<(), EncodeError> { + w.control = Some(crate::stats::Control::Session); w.u8(SERVER_SETUP); let prefix = prefix_body(w, v)?; self.encode_inner(w, v)?; @@ -285,6 +290,7 @@ impl Encode for Server { impl Decode for Server { /// Decode a server setup message (draft-14 through draft-16 only). fn decode(r: &mut Decoder<'_>, v: Version) -> Result { + r.control = Some(crate::stats::Control::Session); let kind = r.u8()?; if kind != SERVER_SETUP { return Err(DecodeError::InvalidValue); diff --git a/rs/moq-net/src/stats.rs b/rs/moq-net/src/stats.rs index 44d1845277c..1f8d048b79e 100644 --- a/rs/moq-net/src/stats.rs +++ b/rs/moq-net/src/stats.rs @@ -307,6 +307,7 @@ struct SessionCounters { bytes: Mutex, wire_tx: AtomicU64, wire_rx: AtomicU64, + control: [[AtomicU64; 2]; Control::ALL.len()], } impl SessionCounters { @@ -345,6 +346,14 @@ impl SessionCounters { (presence, empty) } + fn control(&self, totals: &mut [[u64; 2]; Control::ALL.len()]) { + for (total, counters) in totals.iter_mut().zip(&self.control) { + for (total, counter) in total.iter_mut().zip(counters) { + *total += counter.load(Ordering::Relaxed); + } + } + } + fn bytes(&self) -> MutexGuard<'_, SessionBytes> { self.bytes.lock().unwrap_or_else(PoisonError::into_inner) } @@ -895,6 +904,73 @@ impl Role { } } +/// Cumulative bytes sent and received, returned per [`Control`] class by [`Snapshot::control`]. +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +#[non_exhaustive] +pub struct Bytes { + /// Bytes successfully encoded for sending, counted even if the stream later fails. + pub tx: u64, + /// Bytes of complete messages decoded and consumed. + pub rx: u64, +} + +/// The bounded class of a successfully encoded or decoded control message. +/// +/// Counters include prefixes inside a complete codec value, but exclude separately +/// coded selectors and raw payload writes. A framed IETF body counts after its +/// framing decodes, before request validation. Shared reply IDs use [`Self::Other`]. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[non_exhaustive] +pub enum Control { + /// Session setup and information. + Session, + /// Namespace advertisements and listings. + Announce, + /// Track metadata and listings. + TrackInfo, + /// Live subscriptions and their updates. + Subscribe, + /// Finite fetch requests. + Fetch, + /// Bandwidth probes. + Probe, + /// Authorization messages. + Auth, + /// Session or request migration. + Goaway, + /// Other protocol control messages, including shared request responses. + Other, +} + +impl Control { + const ALL: [Self; 9] = [ + Self::Session, + Self::Announce, + Self::TrackInfo, + Self::Subscribe, + Self::Fetch, + Self::Probe, + Self::Auth, + Self::Goaway, + Self::Other, + ]; + + /// The stable label used by control metrics. + pub fn as_str(self) -> &'static str { + match self { + Self::Session => "session", + Self::Announce => "announce", + Self::TrackInfo => "track_info", + Self::Subscribe => "subscribe", + Self::Fetch => "fetch", + Self::Probe => "probe", + Self::Auth => "auth", + Self::Goaway => "goaway", + Self::Other => "other", + } + } +} + /// A point-in-time, host-level rollup of a registry's counters, returned /// by [`Registry::snapshot`]. /// @@ -913,9 +989,27 @@ pub struct Snapshot { traffic: HashMap, /// Session presence per tier; read via [`Self::sessions`]. sessions: HashMap, + control: HashMap, } impl Snapshot { + /// The `(tier, class, totals)` control-message rows with any traffic, sorted by tier and class. + /// Includes prefixes inside complete codec values, but excludes separately coded type selectors, + /// media payload, and transport framing, acknowledgements, and retransmissions. + pub fn control(&self) -> Vec<(Tier, Control, Bytes)> { + let mut rows = Vec::new(); + for (tier, totals) in &self.control { + for class in Control::ALL { + let [tx, rx] = totals[class as usize]; + if tx != 0 || rx != 0 { + rows.push((tier.clone(), class, Bytes { tx, rx })); + } + } + } + rows.sort_by(|a, b| a.0.as_str().cmp(b.0.as_str()).then(a.1.as_str().cmp(b.1.as_str()))); + rows + } + /// The `(tier, role, totals)` traffic rows, one publisher and one subscriber /// row per tier present. Sorted by tier label then role for stable output. pub fn traffic(&self) -> Vec<(Tier, Role, Traffic)> { @@ -1214,6 +1308,7 @@ impl Registry { let totals = snap.sessions.entry(tier.clone()).or_default(); for counters in labels.values() { totals.add(counters.snapshot().0); + counters.control(snap.control.entry(tier.clone()).or_default()); } } } @@ -1281,6 +1376,7 @@ impl Registry { }); if closed { retired.sessions.entry(tier.clone()).or_default().add(presence); + counters.control(retired.control.entry(tier.clone()).or_default()); } !closed }); @@ -1575,6 +1671,15 @@ struct Viewer { } impl Session { + #[cfg_attr(not(feature = "async"), allow(dead_code))] + pub(crate) fn control(&self, class: Control, bytes: usize, received: bool) { + let Some(inner) = &self.inner else { return }; + let current = inner.current.lock().unwrap_or_else(PoisonError::into_inner); + if let Some(counters) = ¤t.presence { + counters.control[class as usize][usize::from(received)].fetch_add(bytes as u64, Ordering::Relaxed); + } + } + /// Count a connection's wire counters under this context's label, following its retags. pub fn wire(&self, counters: crate::transport::Wire) -> Wire { let Some(inner) = &self.inner else { @@ -2523,6 +2628,26 @@ mod tests { assert!(session_snapshot(&stats, &Tier::default(), "acme").is_none()); } + #[test] + fn control_body_totals_follow_retags_and_survive_retirement() { + let registry = Registry::new(Default::default()); + let first = Tier::new("internal"); + let second = Tier::new("edge"); + let context = registry.tier(first.clone()).session(""); + context.control(Control::Announce, 17, false); + context.retag(second.clone(), ""); + context.control(Control::TrackInfo, 23, true); + let expected = vec![ + (second, Control::TrackInfo, Bytes { tx: 0, rx: 23 }), + (first, Control::Announce, Bytes { tx: 17, rx: 0 }), + ]; + assert_eq!(registry.snapshot().control(), expected); + drop(context); + registry.report(&mut Report::default()); + assert_eq!(registry.snapshot().control(), expected); + assert_eq!(registry.snapshot().control(), expected, "scrapes do not drain totals"); + } + #[test] fn snapshot_preserves_retired_counters() { let stats = test_stats(); diff --git a/rs/moq-net/tests/integration/control_stats.rs b/rs/moq-net/tests/integration/control_stats.rs new file mode 100644 index 00000000000..554859399b6 --- /dev/null +++ b/rs/moq-net/tests/integration/control_stats.rs @@ -0,0 +1,111 @@ +#![cfg(feature = "async")] + +//! Control counters cross real protocol drivers, including legacy multiplexing. + +use crate::support::{ + harness::{MockConnectOptions, connect_mock, now, spawn, subscribed}, + mock::create_mock_session_pair, +}; +use moq_net::{ + Timescale, Timestamp, origin, + stats::{Control, Registry, Tier}, +}; + +fn origin() -> origin::Producer { + let (origin, driver) = origin::Producer::new(Default::default()); + spawn(driver); + origin +} + +#[moq_clock::test] +async fn control_classes_cross_lite_and_ietf_sessions_without_counting_payload() { + for version in [ + "moq-lite-03", + "moq-lite-06", + "moq-lite-07-wip", + "moq-transport-14", + "moq-transport-15", + "moq-transport-16", + "moq-transport-22", + ] { + let publisher = origin(); + let subscriber = origin(); + let registry = Registry::new(Default::default()); + let tier = Tier::new("internal"); + let mut options = MockConnectOptions::new(version.parse().unwrap()); + options.server_stats = registry.tier(tier.clone()).session(""); + options.client_publish = Some(publisher.consume()); + options.server_subscribe = Some(subscriber.clone()); + let _pair = connect_mock(options).await; + let broadcast = publisher.create_broadcast("room").unwrap(); + let track = broadcast + .create_track( + "video", + moq_net::track::Info::default().with_timescale(Timescale::MILLI), + ) + .unwrap(); + broadcast.announce(Default::default()).unwrap(); + subscriber.consume().routed("room").await.unwrap(); + let remote = subscriber.consume().request_broadcast("room", None).await.unwrap(); + let mut sub = remote.track("video").unwrap().subscribe(None).await.unwrap(); + subscribed(&track).await; + let mut group = track.append_group().unwrap(); + group + .write_frame(Timestamp::ZERO, bytes::Bytes::from(vec![7; 64 * 1024])) + .unwrap(); + group.finish().unwrap(); + let mut group = sub.recv_group().await.unwrap().unwrap(); + assert_eq!(group.read_frame().await.unwrap().unwrap().payload.len(), 64 * 1024); + let rows = registry.snapshot().control(); + for class in [Control::Announce, Control::Subscribe] { + assert!( + rows.iter() + .any(|(row_tier, row_class, bytes)| *row_tier == tier && *row_class == class && bytes.tx > 0), + "{version}: {class:?}: {rows:?}" + ); + } + let rx: u64 = rows.iter().map(|(_, _, bytes)| bytes.rx).sum(); + assert!( + rx > 0 && rx < 64 * 1024, + "{version}: media payload entered control counters: {rows:?}" + ); + } +} + +/// The SETUP a server reads before admission counts under the stats the handshake is +/// admitted with, even for a session with neither origin. +#[moq_clock::test] +async fn setup_counts_under_the_admitted_stats() { + for version in ["moq-lite-07-wip", "moq-transport-14", "moq-transport-22"] { + let version: moq_net::Version = version.parse().unwrap(); + let registry = Registry::new(Default::default()); + let tier = Tier::new("internal"); + let mut transport = moq_net::transport::memory::Config::default(); + transport.protocol = Some(version.alpn().to_string()); + let (client_transport, server_transport) = create_mock_session_pair(transport); + let client = moq_net::Client::new().with_versions(version.into()); + let server = moq_net::Server::new().with_versions(version.into()); + let stats = registry.tier(tier.clone()).session(""); + let client = async { + let (session, driver) = client.connect(now(), client_transport).await.unwrap(); + spawn(driver); + session + }; + let server = async { + let request = server.accept_request(now(), server_transport).await.unwrap(); + let (session, driver) = request.with_stats(stats).ok().await.unwrap(); + spawn(driver); + session + }; + let _sessions = futures::join!(client, server); + moq_clock::sleep(std::time::Duration::from_millis(100)).await; + let rows = registry.snapshot().control(); + assert!( + rows.iter() + .any(|(row_tier, class, bytes)| *row_tier == tier + && *class == Control::Session + && bytes.tx > 0 && bytes.rx > 0), + "{version}: {rows:?}" + ); + } +} diff --git a/rs/moq-net/tests/integration/main.rs b/rs/moq-net/tests/integration/main.rs index e9b9d79d8e5..06c1c71e13c 100644 --- a/rs/moq-net/tests/integration/main.rs +++ b/rs/moq-net/tests/integration/main.rs @@ -14,6 +14,7 @@ mod catalog_acquisition; mod catalog_churn; mod catalog_handover; mod compress; +mod control_stats; mod datagram; mod dial_split_horizon; mod drained_claim; diff --git a/rs/moq-relay/src/internal.rs b/rs/moq-relay/src/internal.rs index 873ef2aee5f..d61ec6ffb6e 100644 --- a/rs/moq-relay/src/internal.rs +++ b/rs/moq-relay/src/internal.rs @@ -500,6 +500,51 @@ fn render_metrics( |c| c.broadcasts_ended, ); + for (name, help, field) in [ + ( + "moq_relay_wire_tx_bytes_total", + "Transport bytes sent, including framing and retransmissions, excluding IP, UDP, and TCP headers.", + (|p: &moq_net::stats::Presence| p.wire_tx) as fn(&moq_net::stats::Presence) -> u64, + ), + ( + "moq_relay_wire_rx_bytes_total", + "Transport bytes received, including framing, excluding IP, UDP, and TCP headers.", + (|p: &moq_net::stats::Presence| p.wire_rx) as fn(&moq_net::stats::Presence) -> u64, + ), + ] { + let _ = writeln!(out, "# HELP {name} {help}"); + let _ = writeln!(out, "# TYPE {name} counter"); + for (tier, presence) in snap.sessions() { + let _ = writeln!(out, "{name}{{tier=\"{}\"}} {}", tier.as_str(), field(&presence)); + } + } + + let control = snap.control(); + for (name, help, field) in [ + ( + "moq_relay_control_tx_bytes_total", + "Control-message bytes encoded for sending, including codec prefixes; excludes transport overhead.", + (|b: &moq_net::stats::Bytes| b.tx) as fn(&moq_net::stats::Bytes) -> u64, + ), + ( + "moq_relay_control_rx_bytes_total", + "Control-message bytes decoded, including codec prefixes; excludes transport overhead.", + (|b: &moq_net::stats::Bytes| b.rx) as fn(&moq_net::stats::Bytes) -> u64, + ), + ] { + let _ = writeln!(out, "# HELP {name} {help}"); + let _ = writeln!(out, "# TYPE {name} counter"); + for (tier, class, bytes) in &control { + let _ = writeln!( + out, + "{name}{{tier=\"{}\",class=\"{}\"}} {}", + tier.as_str(), + class.as_str(), + field(bytes) + ); + } + } + let _ = writeln!( out, "# HELP moq_relay_requests_total Admitted request/response exchanges." @@ -1172,6 +1217,32 @@ mod tests { ); } + #[test] + fn wire_metrics_include_internal_sessions_and_keep_final_totals() { + let registry = moq_net::stats::Registry::new(Default::default()); + for tier in ["internal", "edge"] { + let session = registry.tier(moq_net::stats::Tier::new(tier)).session(""); + let source = moq_net::transport::Wire::new(); + let guard = session.wire(source.clone()); + source.set_sent_bytes(123); + source.set_received_bytes(456); + let check = || { + let body = render_metrics(®istry.snapshot(), &[], &[]); + for (direction, value) in [("tx", 123), ("rx", 456)] { + let metric = format!("moq_relay_wire_{direction}_bytes_total"); + assert!(body.contains(&format!("# TYPE {metric} counter")), "{body}"); + assert!(body.contains(&format!("{metric}{{tier=\"{tier}\"}} {value}")), "{body}"); + } + }; + check(); + check(); + drop(guard); + drop(session); + registry.report(&mut Default::default()); + check(); + } + } + #[tokio::test] async fn announce_metrics_keep_retired_totals() { let stats = moq_net::stats::Registry::new(Default::default()); diff --git a/rs/moq-relay/tests/integration/wire_stats.rs b/rs/moq-relay/tests/integration/wire_stats.rs index cf97416540d..37b6dda9407 100644 --- a/rs/moq-relay/tests/integration/wire_stats.rs +++ b/rs/moq-relay/tests/integration/wire_stats.rs @@ -34,6 +34,7 @@ async fn wire_sources(io_uring: bool) { config.listen.tls.key = vec![key_path]; config.auth.public = vec![moq_auth::Pattern::all()]; config.stats.enabled = true; + config.internal.listen = Some("127.0.0.1:0".parse().unwrap()); config.stats.interval = 3600; config.drain_timeout = Duration::ZERO; if io_uring { @@ -43,6 +44,7 @@ async fn wire_sources(io_uring: bool) { } let relay = Relay::load(config).await.unwrap().with_signals(false); let addr = relay.addr().unwrap(); + let metrics = format!("http://{}/metrics", relay.internal().addr().unwrap()); let registry = relay.cluster().stats.clone(); let trigger = relay.shutdown_trigger().clone(); let running = tokio::spawn(relay.run()); @@ -61,7 +63,12 @@ async fn wire_sources(io_uring: bool) { let before = tokio::time::timeout(TIMEOUT, async { loop { let row = presence(®istry); - if row.active() == 1 && row.wire_tx > 0 && row.wire_rx > 0 { + if row.active() == 1 + && row.wire_tx > 0 + && row.wire_rx > 0 + && registry.snapshot().control().iter().any(|(_, class, bytes)| { + *class == moq_net::stats::Control::Announce && bytes.tx > 0 && bytes.rx > 0 + }) { break row; } tokio::task::yield_now().await; @@ -69,6 +76,34 @@ async fn wire_sources(io_uring: bool) { }) .await .expect("admitted session has direct wire counters"); + let body = reqwest::get(&metrics) + .await + .unwrap() + .error_for_status() + .unwrap() + .text() + .await + .unwrap(); + for direction in ["tx", "rx"] { + let prefix = format!("moq_relay_wire_{direction}_bytes_total{{tier=\"\"}} "); + let total: u64 = body + .lines() + .find_map(|line| line.strip_prefix(&prefix)) + .expect("wire metric") + .parse() + .unwrap(); + assert!(total > 0, "live {scheme} transport must reach the ops endpoint: {body}"); + } + for direction in ["tx", "rx"] { + let prefix = format!("moq_relay_control_{direction}_bytes_total{{tier=\"\",class=\"announce\"}} "); + let total: u64 = body + .lines() + .find_map(|line| line.strip_prefix(&prefix)) + .expect("control metric") + .parse() + .unwrap(); + assert!(total > 0, "{scheme}: {body}"); + } drop(connection); let after = tokio::time::timeout(TIMEOUT, async { loop {