From 4c5a9ea4113097be07ba4c8d72971cd14022143a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 00:27:46 -0700 Subject: [PATCH 1/7] chore(quest): claim quest/m1/datagram-object-drop.md Quest-Claim: af6026bb4a9e1401 From e8712d96a5477ada705646e903541aaa81762eac Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 00:37:22 -0700 Subject: [PATCH 2/7] fix(net): drop unidentified datagram objects at plain boundaries Co-Authored-By: GPT-6 --- quest/m1/README.md | 1 - quest/m1/datagram-object-drop.md | 54 -------------------------------- 2 files changed, 55 deletions(-) delete mode 100644 quest/m1/datagram-object-drop.md diff --git a/quest/m1/README.md b/quest/m1/README.md index fe80cc5833..d41a93b0b8 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -118,7 +118,6 @@ surface and zero-copy texture import, land before the cut. - [Release audit findings](/quest/m1/release-audit/README.md) - resolve newly identified resource, parser, playback, and interoperability risks before the cut - [Pre-cut audit findings](/quest/m1/cut-audit/README.md) - fix the crashes, hostile-input costs, and correctness bugs the 2026-10-24 audit of main found - [Cancelled moq-transport requests stop](/quest/m1/ietf-request-cancel.md) - a cancelled request releases its state even when the peer withholds credit -- [Datagrams past object 0 never pose as object 0](/quest/m1/datagram-object-drop.md) - a hop or reader that cannot carry an Object ID delivers only object 0 of a moq-transport datagram group - [Empty segments refused](/quest/m1/empty-segments.md) - `room//cam` fails loud instead of naming `room/cam` - [Cluster routing](/quest/m1/cluster-routing/README.md) - prove shared routing contracts, implement per-origin routes, and connect outbound local clusters - [Routing guide](/quest/m1/routing-guide.md) - concepts and practical outbound federation setup diff --git a/quest/m1/datagram-object-drop.md b/quest/m1/datagram-object-drop.md deleted file mode 100644 index f0aca47dd9..0000000000 --- a/quest/m1/datagram-object-drop.md +++ /dev/null @@ -1,54 +0,0 @@ -# [S] Datagrams past object 0 never pose as object 0 - -## Goal - -A moq-transport datagram group with several objects reaches every surface -that keeps Object IDs intact, and every surface that cannot carry an Object ID -delivers only object 0. Today a moq-lite hop without the `ietf` extension -forwards every object as a plain datagram at the same sequence, and the -plain Rust datagram reader hands out every object's payload at one sequence -with no way to tell them apart. If object 0 is lost or arrives late, object 1 -is delivered as the group's datagram, which is silent substitution, not loss. - -Out of scope: exposing an Object ID on the public `Datagram` in Rust, JS, or -the bindings; refusing the track; a dropped-datagram stat; multi-object -datagram support in JS. - -## Plan - -Decided 2026-10-26 (maintainer, importing -[moq-dev/moq#6885](https://github.com/moq-dev/moq/issues/6885)): - -- Keep the passthrough from - [moq-dev/moq#5494](https://github.com/moq-dev/moq/pull/5494). Rust relays - already carry Object IDs between moq-transport peers and across a lite-07 - hop with the `ietf` extension. The report was against release 0.17, whose - IETF ingress still dropped every object past 0. Rejected: refusing at - ingress everywhere, which deletes that passthrough. -- Drop, don't refuse. A hop that cannot carry the Object ID sends only - object 0, matching how a hop drops an oversized datagram or a status object - it cannot carry. The rest of the track keeps flowing. Rejected: aborting the - track with `Unsupported`, which would also cost the lite viewer object 0. -- Surfaces: - - `Subscription::serve_datagram` (`rs/moq-net/src/lite/publisher.rs`), - when `ietf` is off: skip a datagram whose `wire.object_id` is past 0, as - it already skips a status. - - `track::Subscriber::poll_recv_datagram` (`rs/moq-net/src/model/track.rs`): - skip a stored datagram past object 0, as it already skips a status. - - JS's IETF subscriber already drops past object 0 and needs no change. -- Debug log like the other per-hop drops; no new stat. -- Fix the docs inline: `doc/concept/standard.md` (the objects and datagrams - bullet says plain readers and lite peers see "normal datagram payloads") - and the `model/datagram.rs` module doc. The cursor's private - `poll_recv_datagram` comment says ranges match "by Object ID", which the - code does not do; make it say what the code does. -- Test both surfaces with a group carrying objects 0 to 2, including object - 0 arriving last and object 0 never arriving. - -## Closes - -- [#6885](https://github.com/moq-dev/moq/issues/6885) - moq-relay silently drops datagram objects after object 0 in a group - -## Related - -- [moq-transport 18 and 22 only](/quest/m1/ietf-drop/README.md) - the reporter's draft 16 is among the drafts that quest deletes From ac4e8a67e34679620e33c8e5d6425bf0bd498c2d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 00:37:39 -0700 Subject: [PATCH 3/7] fix(net): preserve object identity when projecting datagrams Co-Authored-By: GPT-6 --- doc/concept/standard.md | 2 +- rs/moq-net/src/lite/ietf.rs | 76 ++++++++++++++++++++++++++++++++ rs/moq-net/src/lite/publisher.rs | 19 ++++++-- rs/moq-net/src/model/datagram.rs | 7 ++- rs/moq-net/src/model/track.rs | 75 +++++++++++++++++++++++++++---- 5 files changed, 164 insertions(+), 15 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index f1a056b983..cfcd1b4ca8 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -35,7 +35,7 @@ subset drops. What a peer actually observes against this implementation: - **One `SETUP` credential**, forwarded to the [auth server](/bin/relay/auth#the-contract) unverified. A second `SETUP` token closes the session. We advertise no token cache, so an alias reference in `SETUP` closes it with `PROTOCOL_VIOLATION`, one on a request with `UNKNOWN_AUTH_TOKEN_ALIAS`, and a registration on a request with `AUTH_TOKEN_CACHE_OVERFLOW`. Where the Rust side authorizes the peer or verifies tokens, a token on a `SUBSCRIBE` or `PUBLISH_NAMESPACE` is verified like an in-band session token and authorizes that one request alone, even where that authorization already covers it; it never widens the session. It is refused with `NOT_SUPPORTED` when nothing verifies it, when it is empty, or when its Token Type is anything but 0 (out of band). A side that does neither, JavaScript included, ignores the token. Rust refuses a request carrying more than one distinct token with `NOT_SUPPORTED`, since it verifies one; JavaScript, which verifies none, ignores them all. A token-authorized `SUBSCRIBE` still stays within the grant the peer gave us. A token-authorized `PUBLISH_NAMESPACE` needs a grant covering everything beneath the namespace, and the tracks we subscribe under it rely on that announce instead of our limit, ending with it. An update renews the token, except on a drafts 14 and 15 `PUBLISH_NAMESPACE`, which has no update, and a refused renewal ends only that request. Rust and JavaScript send the token a route names (`Route::auth`) on its `PUBLISH_NAMESPACE`, including the one drafts 14 and 15 answer a solicited `SUBSCRIBE_NAMESPACE` with, and swapping it renews the request with `REQUEST_UPDATE` from draft 16, or ends and re-sends it before; the inline `NAMESPACE` that answers a solicited `SUBSCRIBE_NAMESPACE` from draft 16 carries no token. A token added to a reconnecting connection's auth handle rides every session it makes, so the announcement keeps it across a reconnect; one added to a single session's handle ends with that session. A token on any other request is decoded and grants nothing, and a request repeating one is refused with `MALFORMED_AUTH_TOKEN`. On draft 17 and later, the negotiated [MoQ Auth extension](/concept/moq-lite#authorization) accepts additional session tokens in band. - **Pause delivery.** `FORWARD=0` on a `SUBSCRIBE` or its update drops the subscription, closing its data streams and holding nothing upstream, while the request and its slot stay. `FORWARD=1` subscribes again as a new `SUBSCRIBE` with the same filter would, so nothing carries across the pause: its `REQUEST_OK` names the live edge it joined at. A paused update may change only priority and max age. - **Refused, not fatal.** A legal request this stack does not serve is rejected on its own and the session stays up: the fetch forms above get `NOT_SUPPORTED`. Range Filters get `INVALID_FILTER` on draft 22 because the advertised filter limit is zero. On draft 22, namespaces whose first tuple field is exactly `.` or `.session` get `DOES_NOT_EXIST` before application lookup; other dot prefixes still reach the application. Subscription updates accept priority, negotiated subscriber max age, and delivery timeouts. Rust also carries range changes from draft 15 onward, keeps bounded subscriptions updatable, renews request tokens, and judges `NEW_GROUP_REQUEST` as a join (see dynamic groups below). On draft 22, widening fills cached history on a separate fetch stream, completing earlier groups and handing the newest head to its live subgroup tail; fills follow the requested group order. Other supported-but-unimplemented preferences end that subscription; a `TRACK_PROPERTY_FILTER` there closes the session with `PROTOCOL_VIOLATION`, since it belongs only to `SUBSCRIBE_TRACKS`. On draft 18 and later a namespace update may change its prefix; suffixes after the acknowledgement are relative to that new prefix. -- **Objects and datagrams.** Only subgroup 0 is accepted: a stream for any other subgroup, whether its header gives the ID or takes it from the first Object ID, is stopped and the session stays up. Rust relays preserve Object IDs, gaps, object properties, FIRST\_OBJECT, and END\_OF\_GROUP status objects between moq-transport peers for subgroup 0, including across a moq-lite-07 hop between two relays that negotiate the [ietf](/draft/moq-ietf) extension. Datagrams retain their IDs, properties, priority, and status. Plain Rust readers and moq-lite peers see consecutive frames and normal datagram payloads; JavaScript supports only normal datagrams at Object ID 0. A malformed datagram closes the session; a relayed one the downstream draft cannot carry is dropped. A relayed Timestamp leaves in the units the downstream response declares, and is dropped where it declares none (drafts 14 through 16), as for a plain frame; an immutable one that would change or vanish is refused. On a timed track the frame holds each object's time, so a mutable Timestamp that only repeats the immutable one is not restated; a status object has no frame, so it keeps its Timestamp among its properties. Only property 0x10 is read as a Timestamp. Property 0x06 stays opaque and keeps its value regardless of timestamp units; it is dropped between drafts 18 and 22, or refused when immutable, because draft 22 assigns it to SUBGROUP\_DELIVERY\_TIMEOUT. From draft 15, a status object carrying properties is refused, and one relayed from draft 14 leaves without them. Draft 14 allows them, but an END\_OF\_TRACK carrying properties ends the subscription as unsupported, since a track end has nowhere to keep them. A `FETCH` carries the END\_OF\_GROUP object on drafts 14 and 15; later drafts have no FETCH status, so their response ends at the last frame. A group cannot resume across route changes by frame ordinal once Object IDs skip ahead of the ordinals on either route; recovery refuses it instead of splicing different objects. +- **Objects and datagrams.** Only subgroup 0 is accepted: a stream for any other subgroup, whether its header gives the ID or takes it from the first Object ID, is stopped and the session stays up. Rust relays preserve Object IDs, gaps, object properties, FIRST\_OBJECT, and END\_OF\_GROUP status objects between moq-transport peers for subgroup 0, including across a moq-lite-07 hop between two relays that negotiate the [ietf](/draft/moq-ietf) extension. Datagrams retain their IDs, properties, priority, and status. Plain Rust readers and moq-lite peers without the extension see consecutive frames and only normal datagram payloads at Object ID 0; other datagram objects and statuses are dropped on those surfaces. JavaScript supports only normal datagrams at Object ID 0. A malformed datagram closes the session; a relayed one the downstream draft cannot carry is dropped. A relayed Timestamp leaves in the units the downstream response declares, and is dropped where it declares none (drafts 14 through 16), as for a plain frame; an immutable one that would change or vanish is refused. On a timed track the frame holds each object's time, so a mutable Timestamp that only repeats the immutable one is not restated; a status object has no frame, so it keeps its Timestamp among its properties. Only property 0x10 is read as a Timestamp. Property 0x06 stays opaque and keeps its value regardless of timestamp units; it is dropped between drafts 18 and 22, or refused when immutable, because draft 22 assigns it to SUBGROUP\_DELIVERY\_TIMEOUT. From draft 15, a status object carrying properties is refused, and one relayed from draft 14 leaves without them. Draft 14 allows them, but an END\_OF\_TRACK carrying properties ends the subscription as unsupported, since a track end has nowhere to keep them. A `FETCH` carries the END\_OF\_GROUP object on drafts 14 and 15; later drafts have no FETCH status, so their response ends at the last frame. A group cannot resume across route changes by frame ordinal once Object IDs skip ahead of the ordinals on either route; recovery refuses it instead of splicing different objects. - **Dynamic groups.** A track advertises `DYNAMIC_GROUPS` when its publisher sets a refresh ratio, and a track received with `DYNAMIC_GROUPS` gets a ratio of 0. A `NEW_GROUP_REQUEST` (draft 15 and later) counts as one more subscriber joining, which a `SUBSCRIBE` already is: it asks for a new group only when the newest group's bytes after its first frame exceed the ratio times the first frame. Joins coalesce until the asked-for group arrives. A Rust relay sends `NEW_GROUP_REQUEST` upstream, on the `SUBSCRIBE` or in a `REQUEST_UPDATE`, once a join wants a group past the newest it holds; a relay holding nothing yet asks for none. On a track without a ratio, an update carrying one ends the subscription with `UPDATE_FAILED`. On drafts 15 and 16, whose `TRACK_STATUS` answer has no room for it, a relay copy first learned from `TRACK_STATUS` (a fetch-only reader) does not advertise it to later subscribers. A moq-lite-07 hop (`moq-lite-07-wip`, opt-in) carries the same demand natively: the ratio in `TRACK_INFO` and the asked-for group in `SUBSCRIBE_UPDATE`. Earlier moq-lite versions carry neither, so a track received over one never refreshes. - **Priority.** Higher is served first. This stack's priorities run from -128 to 127 with 0 unset, and go out on the IETF wire, which ranks lower first, as `127 - priority`: an unset priority is IETF 127. An omitted subscriber or publisher priority reads as the drafts' default of IETF 128, one step below unset, so a relay forwards it unchanged. A relay's upstream subscription carries the highest priority its subscribers ask for and updates it as they come and go; its group order stays as the `SUBSCRIBE` asked, since no draft lets an update change it. - **Size.** An object extension block larger than 64 KiB ends that subgroup stream. The session stays up. This cap is ours, not the draft's. diff --git a/rs/moq-net/src/lite/ietf.rs b/rs/moq-net/src/lite/ietf.rs index 7da53fd149..4857b98b37 100644 --- a/rs/moq-net/src/lite/ietf.rs +++ b/rs/moq-net/src/lite/ietf.rs @@ -1426,6 +1426,82 @@ mod hop_tests { drop(sessions); } + /// Only a negotiated hop can preserve several datagram Object IDs. A late or lost + /// object 0 never lets another object's payload pose as the whole group. + #[moq_clock::test] + async fn datagram_objects_cross_only_an_identifying_hop() { + for offer in [[false, false], [false, true], [true, false], [true, true]] { + for objects in [&[0, 1, 2][..], &[1, 2, 0], &[1, 2]] { + let negotiated = offer == [true, true]; + let upstream = crate::origin::Config::new(crate::Hop::new(2).unwrap()).produce(); + let broadcast = upstream.publish("room", crate::origin::Route::default()).unwrap(); + let info = track::Info { + properties: draft22(Vec::new()), + ..track::Info::timed().with_timescale(Timescale::MILLI) + }; + let mut track = broadcast.create_track("video", info).unwrap(); + let (downstream, sessions) = test_hop(&upstream, "room", offer).await; + let (_copy, mut subscriber) = subscribe(&downstream).await; + // Receiving a group proves the subscription and its alias are bound before + // the datagrams, independently of how the session tasks are scheduled. + track.write_frame(ts(0), b"before".as_slice()).unwrap(); + let mut before = recv(&mut subscriber).await; + assert_eq!(&next(&mut before).await.unwrap().payload.contiguous(0)[..], b"before"); + for &object in objects { + track + .insert_wire_datagram(crate::model::datagram::Stored { + datagram: crate::Datagram { + sequence: 1, + timestamp: Some(ts(1)), + payload: bytes::Bytes::from(vec![object as u8]), + }, + wire: Some(Arc::new(crate::model::datagram::Wire { + object_id: Some(object), + priority: Some(200), + end_of_group: object == 2, + status: None, + properties: Default::default(), + })), + deflated: None, + }) + .unwrap(); + } + track.insert_datagram(2, ts(2), b"next".as_slice()).unwrap(); + let mut seen = Vec::new(); + loop { + let datagram = kio::wait(|waiter| subscriber.poll_recv_wire_datagram(waiter)) + .await + .unwrap() + .unwrap(); + if datagram.sequence == 2 { + assert_eq!(&datagram.payload[..], b"next"); + assert!(datagram.wire.is_none()); + break; + } + assert_eq!(datagram.sequence, 1); + assert_eq!(datagram.payload.len(), 1); + let object = datagram.payload[0] as u64; + if negotiated { + let wire = datagram.wire.as_ref().unwrap(); + assert_eq!(wire.object_id, Some(object)); + assert_eq!(wire.priority, Some(200)); + assert_eq!(wire.end_of_group, object == 2); + } else { + assert!(datagram.wire.is_none()); + } + seen.push(object); + } + let expected: Vec<_> = objects.iter().copied().filter(|id| negotiated || *id == 0).collect(); + assert_eq!(seen, expected, "offer={offer:?} arrivals={objects:?}"); + track.write_frame(ts(3), b"after".as_slice()).unwrap(); + let mut after = recv(&mut subscriber).await; + assert_eq!(after.sequence, 3); + assert_eq!(&next(&mut after).await.unwrap().payload.contiguous(0)[..], b"after"); + drop(sessions); + } + } + } + /// Provenance keeps IETF times unchecked after a hop, while a peer declining the /// extension gets no timeline. SUBSCRIBE, FETCH, and datagrams must agree with TRACK_INFO, /// while a listing stays untimed. diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 783ca727a9..192f02d2f0 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3691,12 +3691,22 @@ impl Subscription { /// transport's datagram limit or the send fails (congestion / no capacity right now). /// Returns whether it was handed to the transport. fn serve_datagram(&mut self, datagram: crate::model::datagram::Stored) -> bool { - // Only the `ietf` extension can say a datagram is a status rather than a payload. + // Only the `ietf` extension carries distinct Object IDs and non-payload statuses. let status = datagram .wire .as_ref() .is_some_and(|wire| wire.status.is_some_and(|status| status != 0)); - if status && !self.ietf { + if !self.ietf + && let Some(wire) = &datagram.wire + && (status || wire.object_id.is_some_and(|id| id > 0)) + { + tracing::debug!( + subscribe = self.id, + sequence = datagram.sequence, + object_id = wire.object_id, + status = wire.status, + "dropping datagram whose object state needs the ietf extension" + ); return false; } let Ok(timestamp) = self @@ -5800,7 +5810,7 @@ mod serve_group_test { assert!(order > 1); } - /// The peer's frame bounds cut datagrams by Object ID: a public datagram is frame 0, a + /// With `ietf` negotiated, frame bounds cut datagrams by Object ID: a public datagram is frame 0, a /// relayed moq-transport one keeps its own, and both ends are inclusive. #[moq_clock::test] async fn frame_bounds_cut_datagrams_by_object_id() { @@ -5820,7 +5830,8 @@ mod serve_group_test { let subscriber = track.subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(3600))); let session = SinkSession::new(Log::default()).with_groups().with_datagrams(1200); let log = session.log.clone(); - let (mut run, mut writer, _) = lite_run(Version::Lite06, session.clone(), subscriber); + let (mut run, mut writer, _) = lite_run(Version::Lite07, session.clone(), subscriber); + run.ctx.ietf = true; run.ctx.group = crate::transport::poll::Session::send_group(&mut run.ctx.session, 3); run.update(lite::SubscribeUpdate { wanted_group: None, diff --git a/rs/moq-net/src/model/datagram.rs b/rs/moq-net/src/model/datagram.rs index 8d12472ac1..a0350b763f 100644 --- a/rs/moq-net/src/model/datagram.rs +++ b/rs/moq-net/src/model/datagram.rs @@ -14,8 +14,11 @@ //! it, which a new subscription may still read. That buffer bills what it retains to the //! track's [`Pool`](crate::cache::Pool), like cached groups, so pressure drops the oldest. //! -//! Wire counterparts: [`crate::lite::Datagram`], and on moq-transport an OBJECT_DATAGRAM at -//! object 0 whose Group ID is the sequence ([`crate::ietf::ObjectDatagram`]). +//! Wire counterparts: [`crate::lite::Datagram`] and moq-transport's +//! [`crate::ietf::ObjectDatagram`], whose Group ID is the sequence. Relays preserve its Object +//! IDs and statuses between moq-transport peers and across a moq-lite hop negotiating `ietf`. +//! The plain reader and a lite hop without that extension deliver only normal object 0 +//! payloads, dropping other objects and statuses without refusing the track. use bytes::Bytes; diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index ce39ba4661..ecd2c6e8a2 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -6535,9 +6535,8 @@ impl Cursor { /// The next datagram inside this subscriber's range, dropping the ones outside it. /// - /// The range is the one groups get: the floor, the cap, and the frames its group's - /// ranges cover, matched by Object ID. - /// Plain datagrams have Object ID 0. A datagram is judged when the cursor reaches + /// The floor, cap and time/group ranges select whole datagram groups here; frame + /// bounds are applied by the sending session. A datagram is judged when the cursor reaches /// it, against the range in force then, and one outside it is gone for good: unlike a /// group past the cap, nothing holds it for a later raise. fn poll_recv_datagram(&mut self, waiter: &kio::Waiter) -> Poll>> { @@ -7095,6 +7094,9 @@ impl Subscriber { /// buffer. Later arrival-order delivery is unchanged, and any outside its group range /// are skipped. A consumer that falls too far behind silently loses the oldest datagrams. /// + /// A moq-transport group delivers only its normal Object ID 0 payload here. Other + /// Object IDs and statuses are skipped; relay paths preserve them with their IDs intact. + /// /// Returns `Poll::Ready(Ok(Some(datagram)))` when one is available, /// `Poll::Ready(Ok(None))` when the track is finished, `Poll::Ready(Err(e))` when the track /// is aborted, or `Poll::Pending` when none is buffered yet. @@ -7103,12 +7105,16 @@ impl Subscriber { let Some(datagram) = ready!(self.poll_recv_stored_datagram(waiter))? else { return Poll::Ready(Ok(None)); }; - // A moq-transport status is no datagram an application can read. - if datagram - .wire - .as_ref() - .is_some_and(|wire| wire.status.is_some_and(|status| status != 0)) + // A plain payload cannot identify another object in the same group or a status. + if let Some(wire) = &datagram.wire + && (wire.object_id.is_some_and(|id| id > 0) || wire.status.is_some_and(|status| status != 0)) { + tracing::debug!( + sequence = datagram.sequence, + object_id = wire.object_id, + status = wire.status, + "dropping datagram not representable as a plain payload" + ); continue; } self.meter_datagram(datagram.payload.len() as u64); @@ -9323,6 +9329,59 @@ mod test { assert_eq!(recv_datagram(&mut subscriber).sequence, 5); } + /// Plain readers cannot identify another object in the same datagram group. Skipping it + /// must neither consume a late object 0 nor alter the typed reader's arrival stream. + #[test] + fn plain_datagrams_never_substitute_another_object() { + for objects in [&[0, 1, 2][..], &[1, 2, 0], &[1, 2]] { + let mut producer = track_producer("test", Info::default()); + let mut plain = producer.subscribe(None); + let mut typed = producer.subscribe(None); + let waiter = kio::Waiter::noop(); + for &object in objects { + let stored = super::super::datagram::Stored { + datagram: Datagram { + sequence: 5, + timestamp: None, + payload: bytes::Bytes::from(vec![object as u8]), + }, + wire: Some(Arc::new(super::super::datagram::Wire { + object_id: Some(object), + priority: Some(200), + end_of_group: object == 2, + status: None, + properties: Default::default(), + })), + deflated: None, + }; + producer.insert_wire_datagram(stored.clone()).unwrap(); + let Poll::Ready(Ok(Some(kept))) = typed.poll_recv_stored_datagram(&waiter) else { + panic!("typed reader lost object {object}"); + }; + assert_eq!(kept, stored); + match object { + 0 => { + let Poll::Ready(Ok(Some(datagram))) = plain.poll_recv_datagram(&waiter) else { + panic!("object 0 was lost in {objects:?}"); + }; + assert_eq!((datagram.sequence, &datagram.payload[..]), (5, &[0][..])); + } + _ => assert!( + plain.poll_recv_datagram(&waiter).is_pending(), + "object {object} substituted" + ), + } + } + producer.insert_datagram(6, None, b"next".as_slice()).unwrap(); + producer.finish().unwrap(); + let Poll::Ready(Ok(Some(datagram))) = plain.poll_recv_datagram(&waiter) else { + panic!("the next datagram was lost in {objects:?}"); + }; + assert_eq!((datagram.sequence, &datagram.payload[..]), (6, &b"next"[..])); + assert!(matches!(plain.poll_recv_datagram(&waiter), Poll::Ready(Ok(None)))); + } + } + /// Exercises the full producer -> publisher-encode -> subscriber-decode -> producer seam /// (everything but the QUIC datagram send/recv), catching any field-order mismatch between /// the wire codec and the model. From afc237ffb247f50212d465756986f8f7e1528971 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 01:01:16 -0700 Subject: [PATCH 4/7] fix(net): skip unidentified payloads before egress metering Co-Authored-By: GPT-6 --- rs/moq-net/src/lite/publisher.rs | 128 ++++++++++++++++++++++++++++--- 1 file changed, 116 insertions(+), 12 deletions(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 192f02d2f0..af84e5ce97 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3691,22 +3691,12 @@ impl Subscription { /// transport's datagram limit or the send fails (congestion / no capacity right now). /// Returns whether it was handed to the transport. fn serve_datagram(&mut self, datagram: crate::model::datagram::Stored) -> bool { - // Only the `ietf` extension carries distinct Object IDs and non-payload statuses. + // Only the `ietf` extension can say a datagram is a status rather than a payload. let status = datagram .wire .as_ref() .is_some_and(|wire| wire.status.is_some_and(|status| status != 0)); - if !self.ietf - && let Some(wire) = &datagram.wire - && (status || wire.object_id.is_some_and(|id| id > 0)) - { - tracing::debug!( - subscribe = self.id, - sequence = datagram.sequence, - object_id = wire.object_id, - status = wire.status, - "dropping datagram whose object state needs the ietf extension" - ); + if status && !self.ietf { return false; } let Ok(timestamp) = self @@ -4211,6 +4201,21 @@ impl TrackRun { /// Send a datagram, recording its sequence once it went out. fn serve_datagram(&mut self, datagram: crate::model::datagram::Stored) { + // Untyped lite can name only object 0. Drop other payload objects before + // metering them; status objects keep their existing separate drop path. + if !self.ctx.ietf + && let Some(wire) = &datagram.wire + && wire.status.is_none_or(|status| status == 0) + && wire.object_id.is_some_and(|id| id > 0) + { + tracing::debug!( + subscribe = self.ctx.id, + sequence = datagram.sequence, + object_id = wire.object_id, + "dropping nonzero datagram object on a lite hop without ietf" + ); + return; + } if let Some(wire) = &datagram.wire && crate::ietf::flate::egress_object( &wire.properties, @@ -5872,6 +5877,105 @@ mod serve_group_test { } } + /// An object this hop cannot identify is never egress payload. An admitted object's + /// existing best-effort accounting still counts a transport-size drop as an attempt. + #[moq_clock::test] + async fn unidentified_datagram_objects_are_not_metered() { + use crate::model::ProduceTest; + + for ietf in [false, true] { + for objects in [&[1, 2][..], &[0, 1, 2]] { + for max in [1, 1200] { + let origin = crate::origin::Config::new(crate::Hop::new(2).unwrap()).produce(); + let broadcast = origin.publish("room", crate::origin::Route::default()).unwrap(); + let info = track::Info { + properties: Some(Arc::new(crate::model::object::TrackProperties { + version: crate::ietf::Version::Draft22, + values: Vec::new(), + })), + ..track::Info::timed() + }; + let _track = broadcast.create_track("test", info).unwrap(); + let registry = crate::stats::Registry::new(Default::default()); + let tagged = origin + .consume() + .with_stats(registry.tier(Default::default()).session("egress")); + let served = tagged.request_broadcast("room", None).await.unwrap(); + let copy = served.track("test").unwrap(); + let subscriber = copy.subscribe(None).await.unwrap(); + let session = SinkSession::new(Log::default()).with_datagrams(max); + let log = session.log.clone(); + let (mut run, _, _) = lite_run(Version::Lite07, session, subscriber); + run.ctx.ietf = ietf; + for &object in objects { + run.serve_datagram(crate::model::datagram::Stored { + datagram: crate::Datagram { + sequence: 5, + timestamp: Some(Timestamp::ZERO), + payload: bytes::Bytes::from_static(b"payload"), + }, + wire: Some(Arc::new(crate::model::datagram::Wire { + object_id: Some(object), + priority: None, + end_of_group: false, + status: None, + properties: Default::default(), + })), + deflated: None, + }); + } + let sent = registry + .snapshot() + .traffic() + .into_iter() + .find(|(_, role, _)| *role == crate::stats::Role::Publisher) + .map(|(_, _, traffic)| traffic) + .unwrap_or_default(); + let admitted = objects.iter().filter(|&&id| ietf || id == 0).count() as u64; + assert_eq!( + (sent.datagrams, sent.groups, sent.frames, sent.bytes), + (admitted, admitted, admitted, 7 * admitted), + "ietf={ietf} ids={objects:?} max={max}" + ); + assert_eq!( + log.datagrams.lock().unwrap().len() as u64, + if max == 1 { 0 } else { admitted } + ); + // Status attempts have no payload and keep their separate accounting; + // only a typed hop can transmit them. + run.serve_datagram(crate::model::datagram::Stored { + datagram: crate::Datagram { + sequence: 6, + timestamp: None, + payload: bytes::Bytes::new(), + }, + wire: Some(Arc::new(crate::model::datagram::Wire { + object_id: Some(2), + priority: None, + end_of_group: false, + status: Some(crate::ietf::object::END_OF_GROUP), + properties: Default::default(), + })), + deflated: None, + }); + let after_status = registry + .snapshot() + .traffic() + .into_iter() + .find(|(_, role, _)| *role == crate::stats::Role::Publisher) + .map(|(_, _, traffic)| traffic) + .unwrap(); + assert_eq!(after_status.datagrams, admitted + 1); + assert_eq!(after_status.bytes, 7 * admitted); + assert_eq!( + log.datagrams.lock().unwrap().len() as u64, + if max == 1 { 0 } else { admitted + u64::from(ietf) } + ); + } + } + } + } + /// A SUBSCRIBE_UPDATE that flips the group order reranks the next datagram, even with no /// group served in between. #[moq_clock::test] From 06d0a9da8247d056e35ba7cedf0f6f89398a2546 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 01:35:12 -0700 Subject: [PATCH 5/7] test(net): distinguish typed datagrams in existing fixtures Co-Authored-By: GPT-6 --- rs/moq-net/src/ietf/subscriber.rs | 47 +++++++++++++++++++++---------- rs/moq-net/src/model/resume.rs | 5 +++- 2 files changed, 36 insertions(+), 16 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index c52c55bfd2..dfb9053fce 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -7532,8 +7532,8 @@ mod tests { ); } - /// An OBJECT_DATAGRAM at object 0 is a datagram group at its Group ID; anything the model - /// cannot carry as one is dropped, and a malformed one is the peer's violation. + /// Relays keep datagram Object IDs, while plain readers see only normal object 0. + /// An unbound alias is dropped, and a malformed datagram is the peer's violation. #[moq_clock::test] async fn an_object_datagram_is_a_datagram_group() { use crate::coding::Encode as _; @@ -7541,11 +7541,12 @@ mod tests { let subscriber = subscriber_with_tracks(&[(RequestId(11), "cam", "audio")]); subscriber.register_alias(RequestId(11), 7).unwrap(); - let mut consumer = { + let (mut consumer, mut relay) = { let mut state = subscriber.state.lock(); let track = state.subscribes.get_mut(&RequestId(11)).unwrap(); track.timescale = Some(Timescale::MILLI); - track.producer.as_ref().unwrap().subscribe(None) + let producer = track.producer.as_ref().unwrap(); + (producer.subscribe(None), producer.subscribe(None)) }; let timestamp = crate::Timestamp::new(96_000, Timescale::MILLI).unwrap(); @@ -7573,7 +7574,7 @@ mod tests { }; let payload = |bytes: &'static [u8]| ietf::DatagramBody::Payload(bytes::Bytes::from_static(bytes)); - // Nonzero objects are delivered; an unbound alias and non-payload statuses stay invisible to plain readers. + // Nonzero objects survive for relays; plain readers skip them, unbound aliases, and statuses. subscriber .recv_datagram(datagram(7, 4, Some(1), payload(b"no"))) .unwrap(); @@ -7585,8 +7586,11 @@ mod tests { subscriber .recv_datagram(datagram(7, 9, Some(0), payload(b"yes"))) .unwrap(); - let first = consumer.recv_datagram().now_or_never().unwrap().unwrap().unwrap(); + let Poll::Ready(Ok(Some(first))) = relay.poll_recv_stored_datagram(&kio::Waiter::noop()) else { + panic!("the relay did not receive the object"); + }; assert_eq!(first.sequence, 4); + assert_eq!(first.wire.as_ref().unwrap().object_id, Some(1)); assert_eq!(&first.payload[..], b"no"); let received = consumer.recv_datagram().now_or_never().unwrap().unwrap().unwrap(); assert_eq!(received.sequence, 9, "the Group ID is the sequence"); @@ -7594,7 +7598,7 @@ mod tests { assert_eq!(&received.payload[..], b"yes"); assert!( consumer.recv_datagram().now_or_never().is_none(), - "only the payload datagrams got through" + "only the normal object 0 payload got through" ); // A status datagram cannot end the group. @@ -14912,14 +14916,15 @@ mod stitch_tests { } } - /// A datagram's object number does not change its public payload view. + /// Relays retain a datagram's Object ID; a plain reader skips it until object 0 arrives. #[moq_clock::test] async fn passthrough_accepts_datagrams_past_object_zero() { for version in [Version::Draft15, Version::Draft22] { let mut h = Harness::with_timescale(Fill::Done, vec![], None); h.subscriber.version = version; let mut consumer = h.track.subscribe(None); - let bytes = ietf::ObjectDatagram { + let mut plain = h.track.subscribe(None); + let mut datagram = ietf::ObjectDatagram { track_alias: ALIAS, group_id: SEQUENCE, object_id: Some(6), @@ -14927,13 +14932,25 @@ mod stitch_tests { end_of_group: false, properties: None, body: ietf::DatagramBody::Payload(bytes::Bytes::from_static(b"object-six")), - } - .encode_bytes(version) - .unwrap(); - h.subscriber.recv_datagram(bytes).unwrap(); - let received = consumer.poll_recv_datagram(&kio::Waiter::noop()); + }; + h.subscriber + .recv_datagram(datagram.encode_bytes(version).unwrap()) + .unwrap(); + let received = consumer.poll_recv_stored_datagram(&kio::Waiter::noop()); + assert!( + matches!(&received, Poll::Ready(Ok(Some(data))) + if data.payload == b"object-six"[..] && data.wire.as_ref().unwrap().object_id == Some(6)), + "{version}: {received:?}" + ); + assert!(plain.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + datagram.object_id = Some(0); + datagram.body = ietf::DatagramBody::Payload(bytes::Bytes::from_static(b"object-zero")); + h.subscriber + .recv_datagram(datagram.encode_bytes(version).unwrap()) + .unwrap(); + let received = plain.poll_recv_datagram(&kio::Waiter::noop()); assert!( - matches!(&received, Poll::Ready(Ok(Some(data))) if data.payload == b"object-six"[..]), + matches!(&received, Poll::Ready(Ok(Some(data))) if data.payload == b"object-zero"[..]), "{version}: {received:?}" ); } diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index f727a0508d..a4ab4fd254 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -2424,7 +2424,10 @@ mod test { deflated: None, }) .unwrap(); - assert!(sub.poll_recv_datagram(&kio::Waiter::noop()).is_ready()); + let Poll::Ready(Ok(Some(datagram))) = sub.poll_recv_stored_datagram(&kio::Waiter::noop()) else { + panic!("the typed reader did not receive the object"); + }; + assert_eq!(datagram.wire.as_ref().unwrap().object_id, Some(4)); None } }; From 1ff9b76104f8a738112ae250cc74a5f325c44b93 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 02:05:25 -0700 Subject: [PATCH 6/7] test(net): end the datagram fixture after its payloads Co-Authored-By: GPT-6 --- rs/moq-net/src/ietf/subscriber.rs | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index dfb9053fce..d93ba04b49 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -7580,11 +7580,10 @@ mod tests { .unwrap(); subscriber.recv_datagram(datagram(8, 4, None, payload(b"no"))).unwrap(); subscriber - .recv_datagram(datagram(7, 4, None, ietf::DatagramBody::Status(END_OF_TRACK))) + .recv_datagram(datagram(7, 9, Some(0), payload(b"yes"))) .unwrap(); - subscriber - .recv_datagram(datagram(7, 9, Some(0), payload(b"yes"))) + .recv_datagram(datagram(7, 10, None, ietf::DatagramBody::Status(END_OF_TRACK))) .unwrap(); let Poll::Ready(Ok(Some(first))) = relay.poll_recv_stored_datagram(&kio::Waiter::noop()) else { panic!("the relay did not receive the object"); From c464cfc0d24c628e2361171dd5fd24ec6fab6448 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Tue, 27 Oct 2026 12:25:54 -0700 Subject: [PATCH 7/7] test(net): build the egress stats label with Path::from_static moq-dev/moq#6648 removed the &str conversion into Path. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/lite/publisher.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index f2d269254a..37541e0140 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -5899,9 +5899,11 @@ mod serve_group_test { }; let _track = broadcast.create_track("test", info).unwrap(); let registry = crate::stats::Registry::new(Default::default()); - let tagged = origin - .consume() - .with_stats(registry.tier(Default::default()).session("egress")); + let tagged = origin.consume().with_stats( + registry + .tier(Default::default()) + .session(crate::Path::from_static("egress")), + ); let served = tagged.request_broadcast("room", None).await.unwrap(); let copy = served.track("test").unwrap(); let subscriber = copy.subscribe(None).await.unwrap();