From 1d4e820ec8857082fb206508db02dbe14140ffa1 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:02:38 -0700 Subject: [PATCH 1/8] quest: claim datagram-unfetchable From 6a0eba7973c7603647b80cea8679f1538b3217f7 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:26:30 -0700 Subject: [PATCH 2/8] feat(net)!: datagrams are live-only A subscription takes only the datagrams pushed after it opened, through whichever copy of the track reaches it, and its group range bounds them the way it bounds groups. A draft-16+ fetch object flagged as a datagram fails only that fetch, as a group that does not exist. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 5 + doc/concept/standard.md | 4 +- drafts/draft-lcurley-moq-lite.md | 5 +- js/net/src/track.test.ts | 28 +++++ js/net/src/track.ts | 12 +- quest/m1/README.md | 1 - quest/m1/datagram-unfetchable.md | 55 --------- quest/m1/js-ietf-datagram.md | 4 - rs/moq-c/src/consume.rs | 6 +- rs/moq-net/src/ietf/publisher.rs | 25 ++++ rs/moq-net/src/ietf/subscriber.rs | 122 +++++++++++++++++++ rs/moq-net/src/model/datagram.rs | 19 +++ rs/moq-net/src/model/resume.rs | 23 ++-- rs/moq-net/src/model/track.rs | 195 +++++++++++++++++++++++++----- 14 files changed, 402 insertions(+), 102 deletions(-) delete mode 100644 quest/m1/datagram-unfetchable.md diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index ff5e6d433c..4ea49eac3a 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -375,6 +375,11 @@ datagram: unreliable, unordered, under about 1200 bytes, and never retransmitted. It suits real-time audio and sensor data. There is no stream fallback, so a datagram that doesn't fit isn't delivered that way. +Datagrams are live-only. A subscription gets the ones sent after it opened and +inside its group range, never a buffered backlog, and nothing caches them, so +`FETCH` treats a datagram group as one that does not exist. Use a group for +anything a late joiner needs. + ## What moq-lite leaves out Compared with moq-transport: no request IDs (a stream per request instead), no diff --git a/doc/concept/standard.md b/doc/concept/standard.md index 9b4730a795..e972127149 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -74,7 +74,9 @@ tracks waiting for a group that never arrives. A moq-lite datagram is a single-frame group, so on moq-transport it travels as an `OBJECT_DATAGRAM` at object 0 whose Group ID is the sequence, and a relay forwards it without renumbering. A datagram carrying any other Object ID, or a -status other than Normal, is dropped. JavaScript does not yet carry datagrams +status other than Normal, is dropped. Datagrams are never fetchable: a fetch +object flagged as a datagram fails that fetch as a group that does not exist, +and leaves the session up. JavaScript does not yet carry datagrams on moq-transport. A client may present one credential in its `SETUP` with the `AUTHORIZATION diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index b5d6ab02fe..360fdea800 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -656,7 +656,9 @@ The subscriber MAY cache the error and potentially retry later. QUIC datagrams provide unreliable, unordered delivery for latency-sensitive content that does not need retransmission. A publisher MAY transmit a Group consisting of exactly one Frame as a single QUIC datagram, in addition to (or instead of) opening a Group Stream, based on application hints, group size, and network conditions; a multi-frame Group is delivered via a Group Stream only. -A datagram-delivered group is not cached or retransmitted; a publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. +Datagram delivery is live-only: a publisher sends a datagram only to the subscriptions active when it sends it, and only when its Group Sequence is inside the subscription's range, where a non-zero `Frame Start` excludes the start group (see [Positions](#positions)). +A datagram is never cached, retransmitted, or replayed to a later subscription, and FETCH never returns one; a publisher answers a FETCH for a group delivered only as a datagram as for a group that does not exist. +A publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. There is no separate subscription for datagram delivery: datagrams are routed to existing subscriptions via the Subscribe ID, and a subscriber receiving the same group via both a stream and a datagram MUST deduplicate by group sequence. Each datagram body has the following encoding (note: there is no message length prefix; the QUIC datagram boundary delimits the payload): @@ -1380,6 +1382,7 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 +- Datagram delivery is live-only: a datagram goes only to the subscriptions active when it is sent and inside their range, and FETCH never returns one. - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset. - A refusal is not retried at another route of the same prefix either. - Made TRACK_INFO Publisher Max Age optional, encoded as milliseconds plus one with zero meaning no limit. diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 9697149830..323b86cfd0 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -156,6 +156,34 @@ test("appendDatagram delivers to a subscriber", async () => { expect(got && dec.decode(got.payload)).toBe("hello"); }); +test("a late subscriber skips the datagram backlog", async () => { + const producer = new TrackProducer("test"); + const early = producer.subscribe(); + producer.appendDatagram(Timestamp.fromMillis(0), enc.encode("old")); + const late = producer.subscribe(); + producer.appendDatagram(Timestamp.fromMillis(0), enc.encode("new")); + + expect(dec.decode((await early.recvDatagram())?.payload)).toBe("old"); + expect(dec.decode((await early.recvDatagram())?.payload)).toBe("new"); + expect(dec.decode((await late.recvDatagram())?.payload)).toBe("new"); +}); + +// The in-range datagrams, sent between the dropped ones, show the reader is live, so the +// range is what dropped the others. +test("the group range bounds datagrams", async () => { + const producer = new TrackProducer("test"); + const track = producer.subscribe({ groups: { start: { included: 5 } } }); + track.setGroups({ end: { excluded: 7 } }); + for (const sequence of [4, 5, 7, 6]) { + producer.insertDatagram(sequence, Timestamp.fromMillis(0), enc.encode("x")); + } + producer.close(); + + expect((await track.recvDatagram())?.sequence).toBe(5); + expect((await track.recvDatagram())?.sequence).toBe(6); + expect(await track.recvDatagram()).toBeUndefined(); +}); + test("datagram buffers drop oldest at capacity without delaying active subscribers", async () => { const producer = new TrackProducer("test"); const slow = producer.subscribe(); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 54c42587b9..40a5bb755e 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -1430,8 +1430,9 @@ export class Subscriber { * Receive the next datagram in arrival order. * * Datagrams are a separate best-effort channel from groups (see - * {@link Producer.appendDatagram}); they share only the sequence namespace. A consumer - * that falls too far behind silently loses the oldest datagrams. Read this alongside + * {@link Producer.appendDatagram}); they share only the sequence namespace. Only datagrams + * sent after the subscription opened arrive, and those outside its group range are skipped. + * A consumer that falls too far behind silently loses the oldest datagrams. Read this alongside * {@link recvGroup} (e.g. in a separate loop) to receive both channels concurrently. * The two cursors are independent: a datagram never moves the group cursor. */ @@ -1446,8 +1447,11 @@ export class Subscriber { for (;;) { const datagrams = this.#state.datagrams.peek(); - if (datagrams.length > 0) { - return datagrams.shift(); + // The group range bounds datagrams too, judged when each is read. One outside it is + // gone for good: nothing holds it for a later raise, unlike a group past the cap. + const { start, end } = this.#cursor.peek(); + for (let datagram = datagrams.shift(); datagram; datagram = datagrams.shift()) { + if (datagram.sequence >= start && (end === undefined || datagram.sequence < end)) return datagram; } const closed = this.#state.closed.peek(); diff --git a/quest/m1/README.md b/quest/m1/README.md index b4ac1f6d83..969f259ca1 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -82,7 +82,6 @@ blocks. The quests that gated m0 lines moved under them. - [Fetch without SUBSCRIBE](/quest/m1/ietf-fetch-only.md) - a relay fetches from an IETF upstream without subscribing, finished tracks included, with End of Track always reported - [IETF object gaps](/quest/m1/ietf-object-gaps.md) - a gapped object ID is refused loudly like a subgroup in Rust and JS, and an overflowing one closes the session in Rust - [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - `@moq/net` sends and receives datagram groups over moq-transport, like Rust -- [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - no FETCH, replay to a new subscriber, or cache fill ever returns a datagram group - [JavaScript FETCH](/quest/m1/js-fetch.md) - browser publishers answer IETF FETCH through the JS ranges request surface - [Bounded HLS playlists](/quest/m1/hls-bounded.md) - `moq-hls` renders a capped sliding window for a fresh viewer, with a stable media sequence, gap slots, and no video before a sync point - [Request update routing](/quest/m1/ietf-subscribe-update-route.md) - drafts 14 and 15 SUBSCRIBE_UPDATE and draft 16 REQUEST_UPDATE reach the request named by their second field diff --git a/quest/m1/datagram-unfetchable.md b/quest/m1/datagram-unfetchable.md deleted file mode 100644 index 76fa052c69..0000000000 --- a/quest/m1/datagram-unfetchable.md +++ /dev/null @@ -1,55 +0,0 @@ -# [S] Datagrams are live-only - -## Goal - -A datagram group is delivered only to subscribers present when it is sent. -FETCH never returns one, a new subscription never replays buffered datagrams, -and a relay fills no cache miss with one. Anything that needs reliable or -late-joiner delivery should use a stream group instead. - -## Plan - -Decided in planning (09-29), after the moxygen fetch-fill agent found that the -relay's group-fetch decoder refuses a draft-16+ fetch object that carries the -datagram flag, and reports it as "unsupported": - -- Keep the existing 64-datagram send buffer per track - (`rs/moq-net/src/model/track.rs`, `MAX_DATAGRAMS`). An age bound was - considered and dropped. -- A new subscription starts at the next datagram sent, not at the buffered - backlog. This covers Rust and `@moq/net`, lite and IETF. -- FETCH never serves a datagram group. A publisher answers a FETCH for one the - way it answers any group it won't deliver (dropped or does not exist), - never with its payload. -- A received fetch object with the datagram flag is refused as not fetchable. - It fails only that fetch or fill, not the session, because draft-16+ allows - the flag (maintainer, 09-29, on Codex's review). That covers the relay's group - fill (`recv_group_fetch_objects`) and the joining-fetch - fill (`run_fill_objects`). -- Update `drafts/draft-lcurley-moq-lite.md` to say datagrams are neither - cached nor fetchable, and `doc/concept/`. Run `just drafts check` and - `just test interop --all`. - -Not planned: reliable or cached datagrams. If they are ever wanted, the way to -get them is back-pressure in the QUIC library and treating each datagram like -a one-shot stream, not a cache. - -The subscription's range bounds datagrams the way it bounds groups, fixed in -the model rather than filtered per session. Decided in the 2026-09-30 audit: -the moxygen line's datagram-range quest merged here, since it duplicated this -late-join rule. Its edge cases: - -- a datagram at the start group when a frame offset skips object 0; -- SUBSCRIBE_UPDATE moving the range while datagrams are in flight; -- a datagram that lands before the subscription's alias or id is known. - -Tests: a subscriber joining after datagrams were sent receives only later -ones. A FETCH covering a datagram group gets no payload. A fetch stream -carrying a datagram-flagged object fails that fetch and leaves the session up. -Each edge case above gets one, and a test tells a datagram filtered by the -range apart from one dropped for any other reason. - -## Related - -- [JS IETF datagrams](/quest/m1/js-ietf-datagram.md) - JS datagram send and receive over moq-transport -- [SUBSCRIBE_DROP](/quest/m1/subscribe-drop.md) - datagram groups stay best effort there too diff --git a/quest/m1/js-ietf-datagram.md b/quest/m1/js-ietf-datagram.md index 7ecfb8faab..0326b949ca 100644 --- a/quest/m1/js-ietf-datagram.md +++ b/quest/m1/js-ietf-datagram.md @@ -15,7 +15,3 @@ same way Rust does, and close the session on a malformed Type. The integration test `ietf does not deliver datagrams` flips to delivery on every supported draft, and `just test interop --all` covers both directions. - -## Related - -- [Datagrams are live-only](/quest/m1/datagram-unfetchable.md) - the subscribe range for datagrams, settled on both protocols diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index fe6473e7a3..2859fee023 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -796,10 +796,12 @@ impl Consume { }; let id = self.datagram_task.insert(Some(entry))?; - // `subscribe` blocks on SUBSCRIBE_OK, so run it inside the task. + // Open the subscription now: a datagram reaches only the subscriptions open when it is + // published. It resolves on SUBSCRIBE_OK, so await that inside the task. + let subscribing = broadcast.track(&name).map(|track| track.subscribe(None)); tokio::spawn(async move { let res = async move { - let track = broadcast.track(&name)?.subscribe(None).await?; + let track = subscribing?.await?; Self::run_datagrams(on_datagram, track, channel.1).await } .await; diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 30a8faa5fe..9df4c8e663 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4723,6 +4723,31 @@ mod serve_tests { } } + /// A datagram group is never fetchable: a FETCH for one is refused like a group that + /// does not exist, newest or not, and opens no fetch stream for its payload. + #[moq_net_sim::test] + async fn a_standalone_fetch_of_a_datagram_group_is_refused() { + for version in FETCH_DRAFTS { + for newest in [true, false] { + let mut h = serve(version); + publish_pairs(&mut h, 2, None); + let sequence = h.track.append_datagram(timestamp(), b"d".as_slice()).unwrap(); + if !newest { + h.track.append_group().unwrap().finish().unwrap(); + } + settle().await; + + let location = Location { group: sequence, object: 0 }; + let buf = standalone_fetch(&h, location, location, GroupOrder::Ascending).await; + assert_eq!( + fetch_refusal(buf, version), + does_not_exist(version), + "{version}: newest={newest}" + ); + } + } + } + /// A joining FETCH reaching back before the subscription's group is refused, like any /// FETCH touching several groups. #[moq_net_sim::test] diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index f144d58419..635f290d3c 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3795,6 +3795,13 @@ async fn decode_fetch_object( priority, properties, }) => { + // Draft-16 on lets a fetch carry an Object published as a datagram, but a + // datagram group is live-only: never cached, so never filled from a fetch. It is + // refused as a group that does not exist, which fails only this fetch. + if subgroup == ietf::FetchSubgroup::Datagram { + tracing::debug!(?group, ?object, "a datagram group is not fetchable"); + return Err(Error::NotFound); + } let inherits = group.is_none() || object.is_none() || priority.is_none() @@ -4623,6 +4630,43 @@ mod tests { assert!(is_protocol_violation(&subscriber.recv_datagram(malformed).unwrap_err())); } + /// A datagram that lands before SUBSCRIBE_OK binds its alias is dropped, not held for the + /// subscription to replay once it is bound. The reader was already subscribed, so only the + /// missing alias can have dropped it. + #[moq_net_sim::test] + async fn a_datagram_before_its_alias_is_dropped() { + use crate::coding::Encode as _; + use futures::FutureExt as _; + + let subscriber = subscriber_with_tracks(&[(RequestId(11), "cam", "audio")]); + let mut consumer = { + let mut state = subscriber.state.lock(); + let track = state.subscribes.get_mut(&RequestId(11)).unwrap(); + track.producer.as_ref().unwrap().subscribe(None) + }; + let datagram = |group_id: u64| { + ietf::ObjectDatagram { + track_alias: 7, + group_id, + object_id: None, + publisher_priority: None, + end_of_group: true, + properties: None, + body: ietf::DatagramBody::Payload(bytes::Bytes::from_static(b"d")), + } + .encode_bytes(Version::Draft19) + .unwrap() + }; + + subscriber.recv_datagram(datagram(4)).unwrap(); + subscriber.register_alias(RequestId(11), 7).unwrap(); + subscriber.recv_datagram(datagram(5)).unwrap(); + + let received = consumer.recv_datagram().now_or_never().unwrap().unwrap().unwrap(); + assert_eq!(received.sequence, 5); + assert!(consumer.recv_datagram().now_or_never().is_none(), "group 4 never arrives"); + } + /// One alias naming two different tracks is the collision section 11.1 makes fatal. #[test] fn an_alias_reused_for_another_track_is_fatal() { @@ -8036,6 +8080,84 @@ mod stitch_tests { } } + /// One Object published as a datagram, as draft-16 on lets a fetch stream carry it. + fn datagram_object(sequence: u64) -> Vec { + let mut buf = Vec::new(); + ietf::FetchObject::Object { + subgroup: ietf::FetchSubgroup::Datagram, + group: Some(sequence), + object: Some(0), + priority: Some(0), + properties: None, + } + .encode(&mut crate::coding::Encoder::new(&mut buf, VERSION.into()), VERSION) + .unwrap(); + crate::coding::Encoder::new(&mut buf, VERSION.into()) + .varint(1u64) + .unwrap(); + buf.extend_from_slice(b"d"); + buf + } + + /// A datagram group is never fetchable, so a group fetch answered with one fails as a + /// group that does not exist, and nothing is cached. + #[moq_net_sim::test] + async fn a_group_fetch_refuses_a_datagram_object() { + let mut run = GroupFetchRun::new(VERSION, datagram_object(SEQUENCE)).await; + let group = run.track.create_group(group::Info { sequence: SEQUENCE }).unwrap(); + let mut consumer = group.consume(); + let slot = kio::Producer::new(GroupFetch::Ready { + producer: group, + timescale: None, + start: 0, + end: None, + }); + + let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; + assert!(matches!(res, Err(Error::NotFound)), "{res:?}"); + assert!( + matches!(consumer.read_frame().await, Err(Error::NotFound)), + "the group fails without the payload" + ); + } + + /// A fill answered with a datagram Object is refused the same way. Only the fill goes: + /// the subscription keeps delivering the groups after it. + #[moq_net_sim::test] + async fn a_datagram_fill_fails_only_the_fill() { + let mut fill = Vec::new(); + crate::coding::Encoder::new(&mut fill, VERSION.into()) + .varint(ietf::FetchHeader::TYPE) + .unwrap(); + ietf::FetchHeader { request_id: REQUEST } + .encode(&mut crate::coding::Encoder::new(&mut fill, VERSION.into()), VERSION) + .unwrap(); + fill.extend(datagram_object(SEQUENCE)); + + let h = Harness::new( + Fill::Serving(Some(Timescale::MICRO)), + vec![fill, tail_stream(SEQUENCE + 1, 0, &[b"next"])], + ); + let mut consumer = h.track.subscribe(None); + let mut fill = h.stream().await; + let mut tail = h.stream().await; + + assert!(matches!( + h.subscriber.clone().recv_fill(&mut fill).await, + Err(Error::NotFound) + )); + h.subscriber + .clone() + .recv_group(&mut tail) + .await + .expect("the subscription carries on"); + + let (sequence, frames) = read_group(&mut consumer).await; + assert_eq!(sequence, SEQUENCE + 1); + assert_eq!(frames.len(), 1); + assert_eq!(frames[0].1, b"next"); + } + /// A group fetch's objects after the FETCH_HEADER: the first one names the group and /// `start`, and every later one is the next object. fn group_fetch_objects(sequence: u64, start: u64, payloads: &[&[u8]]) -> Vec { diff --git a/rs/moq-net/src/model/datagram.rs b/rs/moq-net/src/model/datagram.rs index a8c20b0979..d93226851a 100644 --- a/rs/moq-net/src/model/datagram.rs +++ b/rs/moq-net/src/model/datagram.rs @@ -10,9 +10,14 @@ //! body exceeds the transport's datagram size, and sessions that can't carry datagrams at all //! (moq-lite before 05, or stream-only transports like WebSocket) never deliver them. //! +//! A datagram is live-only: it reaches the subscriptions open when it is pushed, and is never +//! cached, replayed to a later subscription, or served by a fetch. +//! //! Wire counterparts: [`crate::lite::Datagram`], and on moq-transport an OBJECT_DATAGRAM at //! object 0 whose Group ID is the sequence ([`crate::ietf::ObjectDatagram`]). +use std::sync::atomic::{AtomicU64, Ordering}; + use bytes::Bytes; use crate::Timestamp; @@ -38,3 +43,17 @@ pub struct Datagram { /// The datagram payload. pub payload: Bytes, } + +/// A point on the process-wide datagram clock, which orders every datagram push against every +/// subscription opening. A subscription takes the datagrams pushed after it opened, whichever +/// copy of the track (a relay's upstream, a replacement route) they reach it through. +#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] +pub(crate) struct Tick(u64); + +impl Tick { + /// A tick later than every one taken before it, on any thread. + pub(crate) fn next() -> Self { + static CLOCK: AtomicU64 = AtomicU64::new(0); + Self(CLOCK.fetch_add(1, Ordering::Relaxed)) + } +} diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 5a8195b43c..f9ede4e87f 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -20,6 +20,8 @@ use std::task::{Poll, ready}; use crate::{Datagram, Error, Result, StreamError, group, stats, track}; +use super::datagram::Tick; + use super::subscription::{Cap, Position, Subscription, max_some}; /// How many delivered group sequences a reader remembers, so a group two routes both @@ -136,8 +138,9 @@ impl Consumer { self.state.read().copy.clone() } - /// Subscribe to the logical track with the reader's own preferences. - pub(crate) fn subscribe(&self, subscription: kio::Producer) -> Subscriber { + /// Subscribe to the logical track with the reader's own preferences, taking the + /// datagrams pushed to each copy since the reader `opened`. + pub(crate) fn subscribe(&self, subscription: kio::Producer, opened: Tick) -> Subscriber { let mirrored = subscription.read().clone(); let reader = Arc::new_cyclic(|this| { Mutex::new(Reader { @@ -151,6 +154,7 @@ impl Consumer { newest: None, ordered: None, datagram: None, + opened, groups: (0, Cap::from(..)), }) }); @@ -242,8 +246,8 @@ struct Copy { done: Option>, /// The datagram channel ran out. datagrams_done: Option>, - /// The newest datagram handed out before this copy was subscribed: a copy that - /// buffers datagrams replays them, and those were already handed out. + /// The newest datagram handed out before this copy was subscribed: the copy also hands + /// over what it buffered since the reader opened, which the replaced copy may have too. datagrams_seen: Option, /// Held by every group handed out from this copy, so a replaced copy stays /// subscribed while one is still read. @@ -412,6 +416,8 @@ struct Reader { ordered: Option, /// The newest datagram handed out. datagram: Option, + /// When the reader subscribed: each copy hands it the datagrams pushed after this. + opened: Tick, /// The local group filter; see [`track::Subscriber::set_groups`]. groups: (u64, Cap), } @@ -464,7 +470,7 @@ impl Reader { }; self.copies.push(Copy { generation, - sub: Sub::Pending(track.subscribe(wire)), + sub: Sub::Pending(track.subscribe_since(wire, self.opened)), track, floor, until: None, @@ -1311,14 +1317,17 @@ mod test { assert!(matches!(reading.read_frame().now_or_never(), Some(Err(Error::Old)))); } + /// A new copy hands over what it buffered since the reader subscribed, once each, and + /// nothing it buffered before. #[test] fn datagrams_a_new_copy_replays_are_handed_out_once() { let routes = Producer::new(); let logical = logical(&routes); let mut a = copy(); routes.serve(a.consume()); - let mut sub = subscribe(&logical, Duration::from_secs(10)); let mut b = copy(); + b.insert_datagram(9, ts(9), b"d".as_ref()).unwrap(); + let mut sub = subscribe(&logical, Duration::from_secs(10)); for track in [&mut a, &mut b] { for sequence in [1, 2] { track.insert_datagram(sequence, ts(sequence), b"d".as_ref()).unwrap(); @@ -1331,7 +1340,7 @@ mod test { assert_eq!(recv(&mut sub), Some(1)); assert_eq!(recv(&mut sub), Some(2)); routes.serve(b.consume()); - assert_eq!(recv(&mut sub), Some(3)); + assert_eq!(recv(&mut sub), Some(3), "9 predates the reader"); assert_eq!(recv(&mut sub), None); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index a57a2bbe02..77f54bbe4b 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -16,7 +16,7 @@ use crate::{Error, Result, Timescale, Timestamp, coding}; use crate::{broadcast, cache, group, stats}; -use super::{Datagram, Requests}; +use super::{Datagram, Requests, datagram::Tick}; use super::Cap; pub use super::subscription::{Position, Subscription}; @@ -38,7 +38,8 @@ const DEFAULT_PRIORITY: u8 = 127; /// Maximum number of datagrams retained in the per-track send buffer. /// /// Datagrams are a best-effort send buffer, not a replay cache (unlike groups): only the last -/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog. +/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog, and a new +/// subscriber starts at the next datagram rather than at this backlog. /// The payload size limit also bounds the buffer's memory use. const MAX_DATAGRAMS: usize = 64; @@ -207,9 +208,9 @@ pub(crate) struct TrackState { // what each track writes and never touches another track's cache. debt: u64, - // Datagrams in arrival order, bounded by `MAX_DATAGRAMS`. Shares the group - // sequence namespace but is otherwise independent. - datagrams: VecDeque, + // Datagrams in arrival order, bounded by `MAX_DATAGRAMS`, each stamped with when it was + // pushed. Shares the group sequence namespace but is otherwise independent. + datagrams: VecDeque<(Tick, Datagram)>, // Number of datagrams dropped off the front (over capacity), mapping a subscriber's absolute // cursor to an index into `datagrams` (mirrors `offset` for groups). @@ -494,7 +495,7 @@ impl TrackState { fn poll_recv_datagram(&self, index: usize) -> Poll>> { let start = index.saturating_sub(self.datagram_offset); if self.readable() - && let Some(datagram) = self.datagrams.get(start) + && let Some((_, datagram)) = self.datagrams.get(start) { return Poll::Ready(Ok(Some((datagram.clone(), self.datagram_offset + start)))); } @@ -509,13 +510,19 @@ impl TrackState { } } + /// The absolute index of the first buffered datagram pushed after `opened`: where a reader + /// opened then starts, since datagrams go only to the readers present when they are pushed. + fn datagram_start(&self, opened: Tick) -> usize { + self.datagram_offset + self.datagrams.partition_point(|(pushed, _)| *pushed < opened) + } + /// Push a datagram, dropping the oldest when the send buffer is full. fn push_datagram(&mut self, datagram: Datagram) { if self.datagrams.len() == MAX_DATAGRAMS { self.datagrams.pop_front(); self.datagram_offset += 1; } - self.datagrams.push_back(datagram); + self.datagrams.push_back((Tick::next(), datagram)); } /// Find the smallest-sequence cached group satisfying @@ -1522,9 +1529,11 @@ impl Producer { /// [`Self::append_group`] never reuses a number). There is no group fallback: each /// session drops (with a debug log) any datagram whose encoded body exceeds the /// transport's datagram size, and sessions that can't carry datagrams at all (moq-lite - /// before 05, or stream-only transports like WebSocket) never deliver them. Keep payloads well under the 1200-byte minimum path MTU. An origin - /// publisher uses this; a relay preserving upstream numbering uses - /// [`Self::insert_datagram`]. + /// before 05, or stream-only transports like WebSocket) never deliver them. Keep payloads well under the 1200-byte minimum path MTU. + /// A datagram is live-only: it reaches the subscribers present when it is sent, and is + /// never cached, replayed to a later subscriber, or served by a fetch. Use a group for + /// anything a late joiner needs. An origin publisher uses this; a relay preserving + /// upstream numbering uses [`Self::insert_datagram`]. pub fn append_datagram(&mut self, timestamp: Timestamp, payload: B) -> Result { let payload = payload.into_bytes(); if payload.len() > super::datagram::MAX_DATAGRAM_PAYLOAD { @@ -1876,7 +1885,7 @@ impl Producer { name: self.name.clone(), broadcast, info, - inner: Inner::Plain(Cursor::new(self.state.consume(), subscription)), + inner: Inner::Plain(Cursor::new(self.state.consume(), subscription, Tick::next())), // A producer-side (in-process) subscribe is not egress: stay untagged. stats: stats::Scope::default(), _stats_sub: stats::Subscription::default(), @@ -2586,7 +2595,13 @@ impl Consumer { /// the floor that the budget convicts, so the default budget of zero delivers only /// the latest group and a larger one reaches back over what it can still use. pub fn subscribe(&self, subscription: impl Into>) -> kio::Pending { - let subscription = kio::Producer::new(subscription.into().unwrap_or_default()); + self.subscribe_since(subscription.into().unwrap_or_default(), Tick::next()) + } + + /// Open a live subscription taking the datagrams pushed since `opened`: a reader that + /// opened earlier on another copy of this track keeps what reached this one meanwhile. + pub(crate) fn subscribe_since(&self, subscription: Subscription, opened: Tick) -> kio::Pending { + let subscription = kio::Producer::new(subscription); // Register the subscription if the track is live. If it is already closed, the // returned future resolves to the abort error via `Subscribing::poll_ok`. @@ -2597,6 +2612,7 @@ impl Consumer { broadcast: self.broadcast.clone(), state: self.state.clone(), subscription, + opened, stats: self.stats.clone(), }) } @@ -2883,6 +2899,8 @@ pub struct Subscribing { broadcast: Arc, state: kio::Consumer, subscription: kio::Producer, + /// When the subscription opened: its datagrams are the ones pushed after this. + opened: Tick, stats: stats::Scope, } @@ -2897,10 +2915,10 @@ impl Subscribing { let resume = self.state.read().routes.clone(); let inner = match resume { Some(resume) => Inner::Resume( - Box::new(resume.subscribe(self.subscription.clone())), + Box::new(resume.subscribe(self.subscription.clone(), self.opened)), self.state.clone(), ), - None => Inner::Plain(Cursor::new(self.state.clone(), self.subscription.clone())), + None => Inner::Plain(Cursor::new(self.state.clone(), self.subscription.clone(), self.opened)), }; Poll::Ready(Ok(Subscriber { name: self.name.clone(), @@ -3387,7 +3405,8 @@ struct Cursor { subscription: kio::Producer, /// Arrival-order cursor used by `recv_group`. index: usize, - /// Arrival-order cursor used by `recv_datagram`, independent of groups. + /// Arrival-order cursor used by `recv_datagram`, independent of groups. Starts at the + /// first datagram pushed after the subscription opened, never at an older backlog. datagram_index: usize, /// Minimum sequence to return from any `recv` method. Set by `start_at`. min_sequence: u64, @@ -3425,14 +3444,15 @@ struct Cursor { } impl Cursor { - fn new(state: kio::Consumer, subscription: kio::Producer) -> Self { + fn new(state: kio::Consumer, subscription: kio::Producer, opened: Tick) -> Self { let min_sequence = floor_of(&subscription.read()); + let datagram_index = state.read().datagram_start(opened); Self { state, subscription, min_sequence, index: 0, - datagram_index: 0, + datagram_index, next_sequence: 0, end_sequence: None, parked: BTreeMap::new(), @@ -3587,15 +3607,34 @@ 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 a start frame past 0, which + /// excludes the start group's only frame. 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>> { - let Some((datagram, found_index)) = - ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?) - else { - return Poll::Ready(Ok(None)); - }; - - self.datagram_index = found_index + 1; - Poll::Ready(Ok(Some(datagram))) + loop { + let Some((datagram, found_index)) = + ready!(self.poll(waiter, |state| state.poll_recv_datagram(self.datagram_index))?) + else { + return Poll::Ready(Ok(None)); + }; + self.datagram_index = found_index + 1; + + let sequence = datagram.sequence; + let mid_group = self + .subscription + .read() + .start + .is_some_and(|start| start.group == sequence && start.frame > 0); + if sequence >= self.min_sequence + && super::subscription::before_end(sequence, self.end_sequence) + && !mid_group + { + return Poll::Ready(Ok(Some(datagram))); + } + } } /// The lowest servable group past the last one returned, walking off everything the @@ -3816,8 +3855,9 @@ impl Subscriber { /// /// Datagrams are a separate best-effort channel from groups (see /// [`Producer::append_datagram`]); they share only the sequence namespace, and - /// neither cursor moves the other. A consumer that falls too far behind silently - /// loses the oldest datagrams. + /// neither cursor moves the other. The cursor starts at the next datagram sent, never + /// the buffered backlog, and skips any outside this subscriber's group range. A + /// consumer that falls too far behind silently loses the oldest datagrams. /// /// 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 @@ -4759,6 +4799,107 @@ mod test { assert_eq!(&recv_datagram(&mut dg).payload[..], b"go"); } + /// Datagrams are live-only: a subscriber gets those sent after it subscribed, never the + /// buffered backlog. A remote subscription counts from when it opened, not resolved. + #[test] + fn a_late_subscriber_skips_the_datagram_backlog() { + let mut producer = track_producer("test", None); + let consumer = producer.consume(); + let mut early = producer.subscribe(None); + let pending = consumer.subscribe(None); + producer.append_datagram(Timestamp::ZERO, &b"old"[..]).unwrap(); + + let mut opened = pending.now_or_never().unwrap().unwrap(); + let mut late = producer.subscribe(None); + let mut remote = consumer.subscribe(None).now_or_never().unwrap().unwrap(); + producer.append_datagram(Timestamp::ZERO, &b"new"[..]).unwrap(); + + for present in [&mut early, &mut opened] { + assert_eq!(&recv_datagram(present).payload[..], b"old"); + } + for subscriber in [&mut early, &mut opened, &mut late, &mut remote] { + assert_eq!(&recv_datagram(subscriber).payload[..], b"new"); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + } + } + + /// A fetch never serves a datagram group: it is answered like a group that does not + /// exist, whether or not a group followed it. + #[test] + fn a_datagram_group_is_not_fetchable() { + let mut producer = track_producer("test", None); + let consumer = producer.consume(); + let _subscriber = producer.subscribe(None); + let sequence = producer.append_datagram(Timestamp::ZERO, &b"x"[..]).unwrap(); + let fetch = || consumer.fetch_group(sequence, None).now_or_never(); + assert!(matches!(fetch(), Some(Err(Error::NotFound))), "the newest sequence"); + + producer.append_group().unwrap(); + assert!(matches!(fetch(), Some(Err(Error::NotFound))), "below a newer group"); + } + + /// The subscription's floor and cap bound datagrams as they bound groups. The datagrams + /// in range, sent between the dropped ones, show the reader is live, so the range is + /// what dropped the others. + #[test] + fn the_group_range_bounds_datagrams() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(Subscription::default().with_start(Position::group(5))); + subscriber.set_groups(..7); + for sequence in [4, 5, 7, 6] { + producer + .insert_datagram(sequence, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + } + + assert_eq!(recv_datagram(&mut subscriber).sequence, 5); + assert_eq!(recv_datagram(&mut subscriber).sequence, 6); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + } + + /// A start past the start group's first frame excludes that group's datagram, its only + /// frame; a start at frame 0 keeps it. + #[test] + fn a_mid_group_start_drops_the_start_groups_datagram() { + let mut producer = track_producer("test", None); + let mut mid = producer.subscribe(Subscription::default().with_start(Position { group: 5, frame: 1 })); + let mut whole = producer.subscribe(Subscription::default().with_start(Position::group(5))); + for sequence in [5, 6] { + producer + .insert_datagram(sequence, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + } + + assert_eq!(recv_datagram(&mut mid).sequence, 6); + assert_eq!(recv_datagram(&mut whole).sequence, 5); + assert_eq!(recv_datagram(&mut whole).sequence, 6); + } + + /// A range update judges the datagrams still buffered when they are read, and one it + /// dropped stays dropped once the cap rises again: nothing holds it like a parked group. + #[test] + fn a_range_update_applies_to_buffered_datagrams() { + let mut producer = track_producer("test", None); + let mut subscriber = producer.subscribe(None); + for sequence in 1..=4 { + producer + .insert_datagram(sequence, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + } + + subscriber.set_groups(2..4); + assert_eq!(recv_datagram(&mut subscriber).sequence, 2); + assert_eq!(recv_datagram(&mut subscriber).sequence, 3); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + + subscriber.set_groups(..); + assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); + producer + .insert_datagram(5, Timestamp::ZERO, bytes::Bytes::from_static(b"x")) + .unwrap(); + assert_eq!(recv_datagram(&mut subscriber).sequence, 5); + } + /// 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 f761330f6922f7ad73fe1d87f2c812c3c605e33a Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:31:06 -0700 Subject: [PATCH 3/8] style: fmt and doc wording Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 5 ++++- rs/moq-net/src/ietf/subscriber.rs | 5 ++++- rs/moq-net/src/model/track.rs | 10 +++++----- 3 files changed, 13 insertions(+), 7 deletions(-) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index 9df4c8e663..e7a302ea62 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4737,7 +4737,10 @@ mod serve_tests { } settle().await; - let location = Location { group: sequence, object: 0 }; + let location = Location { + group: sequence, + object: 0, + }; let buf = standalone_fetch(&h, location, location, GroupOrder::Ascending).await; assert_eq!( fetch_refusal(buf, version), diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 635f290d3c..5673a55342 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -4664,7 +4664,10 @@ mod tests { let received = consumer.recv_datagram().now_or_never().unwrap().unwrap().unwrap(); assert_eq!(received.sequence, 5); - assert!(consumer.recv_datagram().now_or_never().is_none(), "group 4 never arrives"); + assert!( + consumer.recv_datagram().now_or_never().is_none(), + "group 4 never arrives" + ); } /// One alias naming two different tracks is the collision section 11.1 makes fatal. diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 77f54bbe4b..fb9cc1ca20 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -38,8 +38,8 @@ const DEFAULT_PRIORITY: u8 = 127; /// Maximum number of datagrams retained in the per-track send buffer. /// /// Datagrams are a best-effort send buffer, not a replay cache (unlike groups): only the last -/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog, and a new -/// subscriber starts at the next datagram rather than at this backlog. +/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog, and a +/// subscriber only gets those pushed after it opened. /// The payload size limit also bounds the buffer's memory use. const MAX_DATAGRAMS: usize = 64; @@ -3855,9 +3855,9 @@ impl Subscriber { /// /// Datagrams are a separate best-effort channel from groups (see /// [`Producer::append_datagram`]); they share only the sequence namespace, and - /// neither cursor moves the other. The cursor starts at the next datagram sent, never - /// the buffered backlog, and skips any outside this subscriber's group range. A - /// consumer that falls too far behind silently loses the oldest datagrams. + /// neither cursor moves the other. Only datagrams pushed after the subscription opened + /// arrive, and any outside this subscriber's group range are skipped. A consumer that + /// falls too far behind silently loses the oldest datagrams. /// /// 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 From 434e837a8b0c11acf984424ce622c18dd3fce207 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 00:34:44 -0700 Subject: [PATCH 4/8] docs(moq-c): datagram subscriptions open before returning Co-Authored-By: Claude Opus 5.5 --- rs/moq-c/src/api.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index c65b5079a5..7a4efed61d 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -3649,6 +3649,8 @@ pub extern "C" fn moq_consume_track_cancel(track: u32) -> i32 { /// [moq_consume_datagrams_cancel]. Read each datagram with [moq_consume_datagram] and release /// it with [moq_consume_datagram_free]. Datagrams arrive only over datagram-capable /// transports on moq-transport or lite-05 and newer moq-lite; there is no stream fallback. +/// Datagrams are live-only: the subscription opens before this returns, and gets only the +/// datagrams published after that. /// /// Returns a non-zero handle to the subscription on success, or a negative code on failure. /// From d2055cee5124c86a1a261b39438c223f182615f6 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 10:17:13 -0700 Subject: [PATCH 5/8] feat(net): refuse a fetched datagram as Error::NotFetchable A datagram-flagged fetch object now fails the fetch or fill with a new Error::NotFetchable instead of NotFound. On the wire it goes out as the code a group that does not exist gets: IETF DOES_NOT_EXIST and moq-lite NOT_FOUND, matching what a publisher that never cached the datagram says. moq-ffi folds it into MoqError::NotFound; the relay HTTP fetch and HLS export treat it as a miss (404). Co-Authored-By: Claude Opus 5.5 --- doc/concept/standard.md | 4 ++-- rs/moq-ffi/src/consumer.rs | 2 +- rs/moq-ffi/src/error.rs | 2 +- rs/moq-hls/src/export/rendition.rs | 8 +++++++- rs/moq-net/src/error.rs | 11 ++++++++++- rs/moq-net/src/ietf/error.rs | 15 +++++++++++++-- rs/moq-net/src/ietf/subscriber.rs | 16 ++++++++-------- rs/moq-relay/src/web.rs | 9 +++++++-- 8 files changed, 49 insertions(+), 18 deletions(-) diff --git a/doc/concept/standard.md b/doc/concept/standard.md index e972127149..d7e0ad1ac3 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -75,8 +75,8 @@ A moq-lite datagram is a single-frame group, so on moq-transport it travels as an `OBJECT_DATAGRAM` at object 0 whose Group ID is the sequence, and a relay forwards it without renumbering. A datagram carrying any other Object ID, or a status other than Normal, is dropped. Datagrams are never fetchable: a fetch -object flagged as a datagram fails that fetch as a group that does not exist, -and leaves the session up. JavaScript does not yet carry datagrams +object flagged as a datagram fails only that fetch, as `NotFetchable`, and a +relay answers its own downstream FETCH as for a group that does not exist. JavaScript does not yet carry datagrams on moq-transport. A client may present one credential in its `SETUP` with the `AUTHORIZATION diff --git a/rs/moq-ffi/src/consumer.rs b/rs/moq-ffi/src/consumer.rs index 7d4e13a12f..cd3bda956f 100644 --- a/rs/moq-ffi/src/consumer.rs +++ b/rs/moq-ffi/src/consumer.rs @@ -302,7 +302,7 @@ impl MoqBroadcastConsumer { fn map_fetch_error(err: moq_net::Error) -> MoqError { match err { - moq_net::Error::NotFound => MoqError::NotFound, + moq_net::Error::NotFound | moq_net::Error::NotFetchable => MoqError::NotFound, moq_net::Error::Unsupported | moq_net::Error::Version => MoqError::Unsupported, err => err.into(), } diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index 30e370dbbf..3eb9b7250c 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -280,7 +280,7 @@ impl From for MoqError { fn from(err: moq_net::Error) -> Self { match err { moq_net::Error::Transport(message) => Self::Transport(message), - moq_net::Error::NotFound => Self::NotFound, + moq_net::Error::NotFound | moq_net::Error::NotFetchable => Self::NotFound, moq_net::Error::Closed | moq_net::Error::GoingAway | moq_net::Error::SessionClosed => Self::Closed, moq_net::Error::Cancel => Self::Cancelled, moq_net::Error::Unauthorized => Self::Unauthorized, diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index e90623e7db..7b034b59e4 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -1047,6 +1047,7 @@ fn is_cache_miss(err: &moq_net::Error) -> bool { matches!( err, moq_net::Error::NotFound + | moq_net::Error::NotFetchable | moq_net::Error::Old | moq_net::Error::Evicted | moq_net::Error::Stream( @@ -1100,7 +1101,12 @@ mod tests { #[test] fn a_cache_miss_that_crossed_a_session_is_still_a_cache_miss() { - for local in [moq_net::Error::NotFound, moq_net::Error::Old, moq_net::Error::Evicted] { + for local in [ + moq_net::Error::NotFound, + moq_net::Error::NotFetchable, + moq_net::Error::Old, + moq_net::Error::Evicted, + ] { assert!(is_cache_miss(&local), "{local:?} locally"); let lite = over_lite(&local); assert!(is_cache_miss(&lite), "{local:?} over lite ({lite:?})"); diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 5443205129..20c3f586eb 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -322,6 +322,10 @@ pub enum Error { #[error("not found")] NotFound, + /// A FETCH reached a datagram, which is live-only and never served from a cache. + #[error("not fetchable")] + NotFetchable, + /// A joining FETCH named a request that is not an active subscription. #[error("invalid joining request ID")] InvalidJoiningRequestId, @@ -566,7 +570,9 @@ impl From<&Error> for StreamError { Error::Old => Self::Old, Error::Evicted => Self::Evicted, Error::Lagged => Self::TooFarBehind, - Error::NotFound => Self::NotFound, + // moq-lite has no code of its own for this: a lite publisher never caches a + // datagram, so it answers the same FETCH with NOT_FOUND, and a relay says the same. + Error::NotFound | Error::NotFetchable => Self::NotFound, Error::Unroutable => Self::Unroutable, Error::WrongSize => Self::WrongSize, Error::FrameTooLarge => Self::FrameTooLarge, @@ -726,6 +732,9 @@ mod tests { let relayed = StreamError::from(&Error::from(StreamError::from_code(0x3))); assert_eq!(relayed.to_code(), 0x3); + // A datagram reached by a FETCH goes out as the NOT_FOUND a lite publisher would send. + assert_eq!(StreamError::from(&Error::NotFetchable), StreamError::NotFound); + // Registered stream codes survive the hop unchanged. for code in [ 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 66f1163408..0210b906e4 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -289,7 +289,12 @@ pub(crate) mod request { return TIMEOUT; } Error::Unsupported | Error::Version | Error::Session(SessionError::Version) => return NOT_SUPPORTED, - Error::NotFound | Error::Stream(StreamError::NotFound) => does_not_exist(kind, version), + // A datagram is never cached, so a FETCH reaching one is answered like a group that + // is not there, which is also what our own publisher sends. NOT_SUPPORTED would + // claim the peer does not implement FETCH at all. + Error::NotFound | Error::NotFetchable | Error::Stream(StreamError::NotFound) => { + does_not_exist(kind, version) + } Error::InvalidRange => invalid_range(kind, version), Error::InvalidJoiningRequestId => invalid_joining_request_id(kind, version), // A path with no route is one we will not carry. A subscriber that asked for it @@ -361,12 +366,13 @@ pub(crate) mod request { /// Every error a rejection distinguishes, plus one it does not, so the checks below cover /// the whole registry rather than the variants someone remembered. A new arm in /// [`to_code`] belongs here. - const EVERY_ERROR: [Error; 11] = [ + const EVERY_ERROR: [Error; 12] = [ Error::Duplicate, Error::Unauthorized, Error::Timeout, Error::Unsupported, Error::NotFound, + Error::NotFetchable, Error::InvalidRange, Error::InvalidJoiningRequestId, Error::Unroutable, @@ -395,6 +401,11 @@ pub(crate) mod request { // A broadcast with no route is, to the peer that asked for it, not here either. assert_eq!(to_code(&Error::Unroutable, Kind::Subscribe, Version::Draft20), 0x10); + + // So is a datagram group: never cached, so a FETCH reaching one finds nothing, and + // NOT_SUPPORTED would say the endpoint has no FETCH at all. + assert_eq!(to_code(&Error::NotFetchable, Kind::Fetch, Version::Draft14), 0x4); + assert_eq!(to_code(&Error::NotFetchable, Kind::Fetch, Version::Draft20), 0x10); } /// Draft-14 gives 0x4 to UNINTERESTED on the requests that offer content and to diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 5673a55342..89b053ab42 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3796,11 +3796,11 @@ async fn decode_fetch_object( properties, }) => { // Draft-16 on lets a fetch carry an Object published as a datagram, but a - // datagram group is live-only: never cached, so never filled from a fetch. It is - // refused as a group that does not exist, which fails only this fetch. + // datagram group is live-only: never cached, so never filled from a fetch. Refusing + // it fails only this fetch. if subgroup == ietf::FetchSubgroup::Datagram { tracing::debug!(?group, ?object, "a datagram group is not fetchable"); - return Err(Error::NotFound); + return Err(Error::NotFetchable); } let inherits = group.is_none() || object.is_none() @@ -8102,8 +8102,8 @@ mod stitch_tests { buf } - /// A datagram group is never fetchable, so a group fetch answered with one fails as a - /// group that does not exist, and nothing is cached. + /// A datagram group is never fetchable, so a group fetch answered with one fails as + /// not fetchable, and nothing is cached. #[moq_net_sim::test] async fn a_group_fetch_refuses_a_datagram_object() { let mut run = GroupFetchRun::new(VERSION, datagram_object(SEQUENCE)).await; @@ -8117,9 +8117,9 @@ mod stitch_tests { }); let res = run.subscriber.recv_group_fetch(&mut run.stream, slot).await; - assert!(matches!(res, Err(Error::NotFound)), "{res:?}"); + assert!(matches!(res, Err(Error::NotFetchable)), "{res:?}"); assert!( - matches!(consumer.read_frame().await, Err(Error::NotFound)), + matches!(consumer.read_frame().await, Err(Error::NotFetchable)), "the group fails without the payload" ); } @@ -8147,7 +8147,7 @@ mod stitch_tests { assert!(matches!( h.subscriber.clone().recv_fill(&mut fill).await, - Err(Error::NotFound) + Err(Error::NotFetchable) )); h.subscriber .clone() diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index a443e0cd7b..c67aaa750c 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -959,8 +959,13 @@ async fn serve_fetch( }; let group = match async { crate::fetch_group(&broadcast.track(&track)?, sequence).await }.await { Ok(group) => group, - // A miss upstream arrives as the stream reset that refused the FETCH. - Err(moq_net::Error::NotFound | moq_net::Error::Stream(moq_net::StreamError::NotFound)) => { + // A miss upstream arrives as the stream reset that refused the FETCH. A datagram + // group is never cached, so it is a miss too. + Err( + moq_net::Error::NotFound + | moq_net::Error::NotFetchable + | moq_net::Error::Stream(moq_net::StreamError::NotFound), + ) => { return Err(StatusCode::NOT_FOUND); } Err(_) => return Err(StatusCode::INTERNAL_SERVER_ERROR), From 36fc65f2e5025c8060adb2f4d2a25266024bfebc Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 11:00:17 -0700 Subject: [PATCH 6/8] feat(net): NOT_FETCHABLE stream code on moq-lite-07 A FETCH that reached a datagram resets with 0x3A NOT_FETCHABLE on lite-07; earlier lite versions keep NOT_FOUND, IETF keeps DOES_NOT_EXIST. An origin that still buffers the datagram answers NotFetchable instead of NotFound. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 5 +-- doc/concept/standard.md | 3 +- drafts/draft-lcurley-moq-lite.md | 6 +++- rs/moq-c/src/error.rs | 3 +- rs/moq-ffi/src/error.rs | 3 +- rs/moq-hls/src/export/rendition.rs | 5 ++- rs/moq-net/src/coding/codes.rs | 44 +++++++++++++++++++++---- rs/moq-net/src/error.rs | 20 ++++++++---- rs/moq-net/src/ietf/error.rs | 22 +++++++++---- rs/moq-net/src/lite/version.rs | 11 +++++++ rs/moq-net/src/model/track.rs | 37 ++++++++++++++++----- rs/moq-net/tests/datagram.rs | 52 +++++++++++++++++++++++++++++- rs/moq-relay/src/web.rs | 2 +- 13 files changed, 177 insertions(+), 36 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 4ea49eac3a..817689057a 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -377,8 +377,9 @@ fallback, so a datagram that doesn't fit isn't delivered that way. Datagrams are live-only. A subscription gets the ones sent after it opened and inside its group range, never a buffered backlog, and nothing caches them, so -`FETCH` treats a datagram group as one that does not exist. Use a group for -anything a late joiner needs. +`FETCH` never returns one. A publisher that knows the sequence was a datagram +refuses with `NOT_FETCHABLE` (moq-lite-07), otherwise with `NOT_FOUND`. Use a +group for anything a late joiner needs. ## What moq-lite leaves out diff --git a/doc/concept/standard.md b/doc/concept/standard.md index d7e0ad1ac3..d23498e4e1 100644 --- a/doc/concept/standard.md +++ b/doc/concept/standard.md @@ -76,7 +76,8 @@ as an `OBJECT_DATAGRAM` at object 0 whose Group ID is the sequence, and a relay forwards it without renumbering. A datagram carrying any other Object ID, or a status other than Normal, is dropped. Datagrams are never fetchable: a fetch object flagged as a datagram fails only that fetch, as `NotFetchable`, and a -relay answers its own downstream FETCH as for a group that does not exist. JavaScript does not yet carry datagrams +relay answers its own downstream FETCH with `DOES_NOT_EXIST`, or `NOT_FETCHABLE` +to a moq-lite-07 peer. JavaScript does not yet carry datagrams on moq-transport. A client may present one credential in its `SETUP` with the `AUTHORIZATION diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 360fdea800..152f883bcb 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -328,6 +328,8 @@ Sent when resetting a stream (RESET_STREAM), or when refusing to receive one (ST | ------- | ------------- | ----------- | | 0x39 | TIMESTAMP_MISMATCH | A frame's timestamp does not match its track's timescale. | | ------- | ------------- | ----------- | +| 0x3A | NOT_FETCHABLE | The FETCH named a group delivered only as a datagram, which is never cached (see [Datagrams](#datagrams)). It has no moq-transport value; a bridge refuses the FETCH with DOES_NOT_EXIST. | +| ------- | ------------- | ----------- | Note that CANCELLED is 0x1, not 0x0: a stream reset with 0x0 is an INTERNAL_ERROR, not a routine cancellation. An endpoint terminating a stream because the session is ending SHOULD use SESSION_CLOSED rather than the session's own code, since the two spaces are disjoint. @@ -657,7 +659,8 @@ QUIC datagrams provide unreliable, unordered delivery for latency-sensitive cont A publisher MAY transmit a Group consisting of exactly one Frame as a single QUIC datagram, in addition to (or instead of) opening a Group Stream, based on application hints, group size, and network conditions; a multi-frame Group is delivered via a Group Stream only. Datagram delivery is live-only: a publisher sends a datagram only to the subscriptions active when it sends it, and only when its Group Sequence is inside the subscription's range, where a non-zero `Frame Start` excludes the start group (see [Positions](#positions)). -A datagram is never cached, retransmitted, or replayed to a later subscription, and FETCH never returns one; a publisher answers a FETCH for a group delivered only as a datagram as for a group that does not exist. +A datagram is never cached, retransmitted, or replayed to a later subscription, and FETCH never returns one. +A publisher that knows a FETCH names a group delivered only as a datagram resets the stream with NOT_FETCHABLE; otherwise it answers as for a group that does not exist, with NOT_FOUND. A publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. There is no separate subscription for datagram delivery: datagrams are routed to existing subscriptions via the Subscribe ID, and a subscriber receiving the same group via both a stream and a datagram MUST deduplicate by group sequence. @@ -1383,6 +1386,7 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 - Datagram delivery is live-only: a datagram goes only to the subscriptions active when it is sent and inside their range, and FETCH never returns one. +- Assigned 0x3A NOT_FETCHABLE in the stream error table: a FETCH for a group delivered only as a datagram. - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset. - A refusal is not retried at another route of the same prefix either. - Made TRACK_INFO Publisher Max Age optional, encoded as milliseconds plus one with zero meaning no limit. diff --git a/rs/moq-c/src/error.rs b/rs/moq-c/src/error.rs index ddea6193fa..c0c0f6b10c 100644 --- a/rs/moq-c/src/error.rs +++ b/rs/moq-c/src/error.rs @@ -393,7 +393,8 @@ fn stream_kind(err: &moq_net::StreamError) -> moq_protocol_kind { moq_net::StreamError::GoingAway => MOQ_PROTOCOL_KIND_GOING_AWAY, moq_net::StreamError::TooFarBehind => MOQ_PROTOCOL_KIND_TOO_FAR_BEHIND, moq_net::StreamError::MalformedTrack => MOQ_PROTOCOL_KIND_MALFORMED_TRACK, - moq_net::StreamError::NotFound => MOQ_PROTOCOL_KIND_NOT_FOUND, + // A datagram reached by a FETCH is, to a binding, a miss like any other. + moq_net::StreamError::NotFound | moq_net::StreamError::NotFetchable => MOQ_PROTOCOL_KIND_NOT_FOUND, moq_net::StreamError::Unroutable => MOQ_PROTOCOL_KIND_UNROUTABLE, moq_net::StreamError::Old => MOQ_PROTOCOL_KIND_OLD, moq_net::StreamError::Evicted => MOQ_PROTOCOL_KIND_EVICTED, diff --git a/rs/moq-ffi/src/error.rs b/rs/moq-ffi/src/error.rs index 3eb9b7250c..1f5ef3cdec 100644 --- a/rs/moq-ffi/src/error.rs +++ b/rs/moq-ffi/src/error.rs @@ -136,7 +136,8 @@ fn stream_kind(err: &moq_net::StreamError) -> MoqProtocolKind { moq_net::StreamError::GoingAway => MoqProtocolKind::GoingAway, moq_net::StreamError::TooFarBehind => MoqProtocolKind::TooFarBehind, moq_net::StreamError::MalformedTrack => MoqProtocolKind::MalformedTrack, - moq_net::StreamError::NotFound => MoqProtocolKind::NotFound, + // A datagram reached by a FETCH is, to a binding, a miss like any other. + moq_net::StreamError::NotFound | moq_net::StreamError::NotFetchable => MoqProtocolKind::NotFound, moq_net::StreamError::Unroutable => MoqProtocolKind::Unroutable, moq_net::StreamError::Old => MoqProtocolKind::Old, moq_net::StreamError::Evicted => MoqProtocolKind::Evicted, diff --git a/rs/moq-hls/src/export/rendition.rs b/rs/moq-hls/src/export/rendition.rs index 7b034b59e4..2db0dca4d1 100644 --- a/rs/moq-hls/src/export/rendition.rs +++ b/rs/moq-hls/src/export/rendition.rs @@ -1051,7 +1051,10 @@ fn is_cache_miss(err: &moq_net::Error) -> bool { | moq_net::Error::Old | moq_net::Error::Evicted | moq_net::Error::Stream( - moq_net::StreamError::NotFound | moq_net::StreamError::Old | moq_net::StreamError::Evicted + moq_net::StreamError::NotFound + | moq_net::StreamError::NotFetchable + | moq_net::StreamError::Old + | moq_net::StreamError::Evicted ) ) } diff --git a/rs/moq-net/src/coding/codes.rs b/rs/moq-net/src/coding/codes.rs index 723b4ebe36..409ae657bb 100644 --- a/rs/moq-net/src/coding/codes.rs +++ b/rs/moq-net/src/coding/codes.rs @@ -7,8 +7,8 @@ use crate::{Error, StreamError, ietf, lite}; /// Every [`Reader`](super::Reader) and [`Writer`](super::Writer) carries the negotiated /// version, which is what picks the registry here. The two wires draw on the same names, /// but they do not agree on every value: moq-lite's are fixed by -/// [`StreamError::to_code`], while moq-transport's moved across the drafts we negotiate -/// (see the `ietf::error` module). Sending a code from the wrong table is silent, because the +/// [`StreamError::to_code`] (a later version adds codes but never moves one), while +/// moq-transport's moved across the drafts we negotiate (see the `ietf::error` module). Sending a code from the wrong table is silent, because the /// number is valid in both. /// /// Encoding and decoding live on one trait so a peer cannot be told a code from one table @@ -38,15 +38,21 @@ pub trait StreamCodes { } } -/// moq-lite's registry, specified by draft-lcurley-moq-lite (Error Codes) and identical -/// across the versions we negotiate. +/// moq-lite's registry, specified by draft-lcurley-moq-lite (Error Codes). A code a version +/// lacks is sent there as the closest one it has, and not read back there. impl StreamCodes for lite::Version { fn encode_stream_code(&self, err: &StreamError) -> u32 { - err.to_code() + match err { + StreamError::NotFetchable if !self.has_not_fetchable() => StreamError::NotFound.to_code(), + err => err.to_code(), + } } fn decode_stream_code(&self, code: u32) -> StreamError { - StreamError::from_code(code) + match StreamError::from_code(code) { + StreamError::NotFetchable if !self.has_not_fetchable() => StreamError::Unknown(code), + err => err, + } } } @@ -110,6 +116,32 @@ mod tests { ); } + /// NOT_FETCHABLE is lite-07's: an earlier peer is told NOT_FOUND, the miss it would + /// have been told before, and is not credited with a code its version never assigned. + #[test] + fn not_fetchable_is_lite_07s() { + for version in [lite::Version::Lite05, lite::Version::Lite06] { + assert_eq!( + version.encode_stream_code(&StreamError::NotFetchable), + StreamError::NotFound.to_code(), + "{version}" + ); + assert_eq!( + version.decode_stream_code(0x3a), + StreamError::Unknown(0x3a), + "{version}" + ); + } + + let version = lite::Version::Lite07; + assert_eq!(version.encode_stream_code(&StreamError::NotFetchable), 0x3a); + assert_eq!(version.decode_stream_code(0x3a), StreamError::NotFetchable); + assert_eq!( + version.decode_stream_code(StreamError::NotFound.to_code()), + StreamError::NotFound + ); + } + /// A dropped [`Writer`](super::Writer) resets with a cancellation, and `Drop` cannot /// carry the bound that would reach the negotiated registry. It does not need one only /// while every registry we speak agrees on the value, so pin that. diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index 20c3f586eb..d8155f4812 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -146,6 +146,11 @@ pub enum StreamError { #[error("not found")] NotFound, + /// A FETCH reached a datagram, which is live-only and never cached. Sent on moq-lite-07 + /// and later; an earlier version sends [`NotFound`](Self::NotFound) instead. + #[error("not fetchable")] + NotFetchable, + /// The broadcast is neither announced nor served, so there is no route to it. #[error("unroutable")] Unroutable, @@ -206,6 +211,7 @@ impl StreamError { Self::WrongSize => 0x37, Self::FrameTooLarge => 0x38, Self::TimestampMismatch => 0x39, + Self::NotFetchable => 0x3a, Self::App(app) => *app as u32 + 64, Self::Unknown(code) => *code, } @@ -237,6 +243,7 @@ impl StreamError { 0x37 => Self::WrongSize, 0x38 => Self::FrameTooLarge, 0x39 => Self::TimestampMismatch, + 0x3a => Self::NotFetchable, code @ 64.. => match u16::try_from(code - 64) { Ok(app) => Self::App(app), Err(_) => Self::Unknown(code), @@ -570,9 +577,8 @@ impl From<&Error> for StreamError { Error::Old => Self::Old, Error::Evicted => Self::Evicted, Error::Lagged => Self::TooFarBehind, - // moq-lite has no code of its own for this: a lite publisher never caches a - // datagram, so it answers the same FETCH with NOT_FOUND, and a relay says the same. - Error::NotFound | Error::NotFetchable => Self::NotFound, + Error::NotFound => Self::NotFound, + Error::NotFetchable => Self::NotFetchable, Error::Unroutable => Self::Unroutable, Error::WrongSize => Self::WrongSize, Error::FrameTooLarge => Self::FrameTooLarge, @@ -681,6 +687,7 @@ mod tests { StreamError::WrongSize, StreamError::FrameTooLarge, StreamError::TimestampMismatch, + StreamError::NotFetchable, StreamError::App(7), ]; for err in registered { @@ -698,6 +705,7 @@ mod tests { (StreamError::WrongSize, 0x37), (StreamError::FrameTooLarge, 0x38), (StreamError::TimestampMismatch, 0x39), + (StreamError::NotFetchable, 0x3a), ] { assert_eq!(err.to_code(), code, "{err:?} moved off its assigned code"); } @@ -732,12 +740,12 @@ mod tests { let relayed = StreamError::from(&Error::from(StreamError::from_code(0x3))); assert_eq!(relayed.to_code(), 0x3); - // A datagram reached by a FETCH goes out as the NOT_FOUND a lite publisher would send. - assert_eq!(StreamError::from(&Error::NotFetchable), StreamError::NotFound); + // A datagram reached by a FETCH keeps its own code rather than reading as a plain miss. + assert_eq!(StreamError::from(&Error::NotFetchable), StreamError::NotFetchable); // Registered stream codes survive the hop unchanged. for code in [ - 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, + 0x0, 0x1, 0x2, 0x4, 0x5, 0x12, 0x31, 0x32, 0x33, 0x34, 0x35, 0x36, 0x37, 0x38, 0x39, 0x3a, ] { let relayed = StreamError::from(&Error::from(StreamError::from_code(code))); assert_eq!(relayed.to_code(), code, "stream {code:#x} changed across a relay"); diff --git a/rs/moq-net/src/ietf/error.rs b/rs/moq-net/src/ietf/error.rs index 0210b906e4..44b1010059 100644 --- a/rs/moq-net/src/ietf/error.rs +++ b/rs/moq-net/src/ietf/error.rs @@ -289,12 +289,12 @@ pub(crate) mod request { return TIMEOUT; } Error::Unsupported | Error::Version | Error::Session(SessionError::Version) => return NOT_SUPPORTED, - // A datagram is never cached, so a FETCH reaching one is answered like a group that - // is not there, which is also what our own publisher sends. NOT_SUPPORTED would - // claim the peer does not implement FETCH at all. - Error::NotFound | Error::NotFetchable | Error::Stream(StreamError::NotFound) => { - does_not_exist(kind, version) - } + // moq-transport has no code for a FETCH that reached a datagram, which is never + // cached, so it is a group that is not there. NOT_SUPPORTED would claim the peer + // does not implement FETCH at all. + Error::NotFound + | Error::NotFetchable + | Error::Stream(StreamError::NotFound | StreamError::NotFetchable) => does_not_exist(kind, version), Error::InvalidRange => invalid_range(kind, version), Error::InvalidJoiningRequestId => invalid_joining_request_id(kind, version), // A path with no route is one we will not carry. A subscriber that asked for it @@ -406,6 +406,11 @@ pub(crate) mod request { // NOT_SUPPORTED would say the endpoint has no FETCH at all. assert_eq!(to_code(&Error::NotFetchable, Kind::Fetch, Version::Draft14), 0x4); assert_eq!(to_code(&Error::NotFetchable, Kind::Fetch, Version::Draft20), 0x10); + // Including a lite-07 upstream's NOT_FETCHABLE, bridged by a relay. + assert_eq!( + to_code(&Error::Stream(StreamError::NotFetchable), Kind::Fetch, Version::Draft20), + 0x10 + ); } /// Draft-14 gives 0x4 to UNINTERESTED on the requests that offer content and to @@ -706,6 +711,7 @@ mod tests { fn unregistered_conditions_are_internal() { for err in [ StreamError::NotFound, + StreamError::NotFetchable, StreamError::Unroutable, StreamError::Old, StreamError::Evicted, @@ -739,6 +745,7 @@ mod tests { 0x37, 0x38, 0x39, + 0x3a, 64 + 7, ] { assert_eq!(from_stream_code(code, Version::Draft20), StreamError::Unknown(code)); @@ -751,7 +758,7 @@ mod tests { /// Every stream error this crate can hold, so the conformance check below covers the /// whole space rather than the variants someone remembered. A new variant belongs here. - const EVERY_ERROR: [StreamError; 18] = [ + const EVERY_ERROR: [StreamError; 19] = [ StreamError::Session(SessionError::Cancel), StreamError::Internal, StreamError::Cancel, @@ -761,6 +768,7 @@ mod tests { StreamError::TooFarBehind, StreamError::MalformedTrack, StreamError::NotFound, + StreamError::NotFetchable, StreamError::Unroutable, StreamError::Old, StreamError::Evicted, diff --git a/rs/moq-net/src/lite/version.rs b/rs/moq-net/src/lite/version.rs index 727035b7a1..fe1c15b04a 100644 --- a/rs/moq-net/src/lite/version.rs +++ b/rs/moq-net/src/lite/version.rs @@ -227,6 +227,17 @@ impl Version { } } + /// Whether the stream error table has NOT_FETCHABLE, for a FETCH that reached a + /// datagram. Added in lite-07; earlier versions say NOT_FOUND instead. + #[allow(clippy::match_like_matches_macro)] + pub(crate) fn has_not_fetchable(self) -> bool { + // Match form so future versions default forward (AGENTS.md convention). + match self { + Self::Lite01 | Self::Lite02 | Self::Lite03 | Self::Lite04 | Self::Lite05 | Self::Lite06 => false, + _ => true, + } + } + /// Whether announcements and track requests carry the publisher epoch, the /// identity that says two routes serve the same bytes. Added in lite-07. /// Older versions carry nothing, so a received route has no epoch and is diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index fb9cc1ca20..95397479da 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -516,6 +516,11 @@ impl TrackState { self.datagram_offset + self.datagrams.partition_point(|(pushed, _)| *pushed < opened) } + /// Whether `sequence` was sent as a datagram still in the send buffer. + fn holds_datagram(&self, sequence: u64) -> bool { + self.datagrams.iter().any(|(_, datagram)| datagram.sequence == sequence) + } + /// Push a datagram, dropping the oldest when the send buffer is full. fn push_datagram(&mut self, datagram: Datagram) { if self.datagrams.len() == MAX_DATAGRAMS { @@ -2729,8 +2734,9 @@ impl Consumer { /// or `group::Fetch::default()`. /// /// The returned future resolves to [`Error::NotFound`] when the group can never be served - /// (past the final sequence, or no [`Dynamic`] on the track), the handler's rejection - /// (a relay's upstream miss is [`StreamError::NotFound`](crate::StreamError::NotFound)), + /// (past the final sequence, or no [`Dynamic`] on the track), [`Error::NotFetchable`] + /// instead when this track sent the sequence as a datagram still buffered, the handler's + /// rejection (a relay's upstream miss is [`StreamError::NotFound`](crate::StreamError::NotFound)), /// or the track's abort error if it's already closed. Concurrent fetches for the same sequence coalesce onto one /// handler request. pub fn fetch_group(&self, sequence: u64, options: impl Into>) -> kio::Pending { @@ -3199,11 +3205,15 @@ impl kio::Task for Fetching { let Some(result) = result else { // Never queued: no handler existed when the fetch was made. Fail fast while // that's still true; a handler that appeared since may yet fill the cache. + // A sequence this track sent as a datagram says so, rather than a plain miss. return match fetch.poll(waiter, |fetch| match fetch.has_handlers() { false => Poll::Ready(()), true => Poll::Pending, }) { - Poll::Ready(_guard) => Poll::Ready(Err(Error::NotFound)), + Poll::Ready(_guard) => Poll::Ready(Err(match state.read().holds_datagram(sequence) { + true => Error::NotFetchable, + false => Error::NotFound, + })), Poll::Pending => Poll::Pending, }; }; @@ -4823,19 +4833,30 @@ mod test { } } - /// A fetch never serves a datagram group: it is answered like a group that does not - /// exist, whether or not a group followed it. + /// A fetch never serves a datagram group, and says so rather than reporting a plain + /// miss, whether or not a group followed it. #[test] fn a_datagram_group_is_not_fetchable() { let mut producer = track_producer("test", None); let consumer = producer.consume(); let _subscriber = producer.subscribe(None); let sequence = producer.append_datagram(Timestamp::ZERO, &b"x"[..]).unwrap(); - let fetch = || consumer.fetch_group(sequence, None).now_or_never(); - assert!(matches!(fetch(), Some(Err(Error::NotFound))), "the newest sequence"); + let fetch = |sequence| consumer.fetch_group(sequence, None).now_or_never(); + assert!( + matches!(fetch(sequence), Some(Err(Error::NotFetchable))), + "the newest sequence" + ); producer.append_group().unwrap(); - assert!(matches!(fetch(), Some(Err(Error::NotFound))), "below a newer group"); + assert!( + matches!(fetch(sequence), Some(Err(Error::NotFetchable))), + "below a newer group" + ); + // A sequence that was never a datagram is a plain miss. + assert!( + matches!(fetch(sequence + 2), Some(Err(Error::NotFound))), + "a missing group" + ); } /// The subscription's floor and cap bound datagrams as they bound groups. The datagrams diff --git a/rs/moq-net/tests/datagram.rs b/rs/moq-net/tests/datagram.rs index 9386e4727d..eaa0150e0c 100644 --- a/rs/moq-net/tests/datagram.rs +++ b/rs/moq-net/tests/datagram.rs @@ -10,7 +10,7 @@ mod support; use std::time::Duration; use moq_net::{Hop, Timestamp, Version}; -use support::harness::{MockConnectOptions, MockPair, connect_mock}; +use support::harness::{MockConnectOptions, MockPair, connect_mock, peer}; /// Maximum time any single test may run before being treated as a deadlock. const TEST_TIMEOUT: Duration = Duration::from_secs(10); @@ -219,3 +219,53 @@ async fn inserted_sequences_survive_the_lite_wire() { .await .expect("timed out"); } + +/// A FETCH for a sequence the publisher sent as a datagram is refused with NOT_FETCHABLE on +/// moq-lite-07, and with the NOT_FOUND an earlier version has for it. A relay carries the +/// answer across in the downstream version's terms. The moq-transport side (DOES_NOT_EXIST) +/// is covered by the IETF publisher's own tests. +#[moq_net_sim::test] +async fn a_fetched_datagram_is_refused_per_version() { + const NOT_FETCHABLE: &str = "not fetchable"; + const NOT_FOUND: &str = "not found"; + + /// Fetch a datagram's sequence over one hop per version, publisher first. + async fn fetch_a_datagram(versions: &[&str]) -> &'static str { + let nodes: Vec<_> = (1..=versions.len() as u64 + 1).map(produce_origin).collect(); + let mut _pairs = Vec::new(); + for (pair, version) in nodes.windows(2).zip(versions) { + _pairs.push(peer(version.parse().unwrap(), &pair[0], &pair[1]).await); + } + + let broadcast = nodes[0].create_broadcast("room").unwrap(); + let mut producer = broadcast.create_track("datagrams", None).unwrap(); + broadcast.announce(Default::default()).unwrap(); + let sequence = producer.append_datagram(Timestamp::ZERO, &b"x"[..]).unwrap(); + + let consumer = nodes.last().unwrap().consume(); + consumer.routed("room").await.unwrap(); + let remote = consumer.request_broadcast("room").await.unwrap(); + let err = match remote.track("datagrams").unwrap().fetch_group(sequence, None).await { + Ok(_) => panic!("{versions:?}: a datagram was fetched"), + Err(err) => err, + }; + match err { + moq_net::Error::Stream(moq_net::StreamError::NotFetchable) => NOT_FETCHABLE, + moq_net::Error::NotFound | moq_net::Error::Stream(moq_net::StreamError::NotFound) => NOT_FOUND, + err => panic!("{versions:?}: refused with {err:?}"), + } + } + + for (versions, expected) in [ + (&["moq-lite-05"][..], NOT_FOUND), + (&["moq-lite-06"], NOT_FOUND), + (&["moq-lite-07-wip"], NOT_FETCHABLE), + (&["moq-lite-07-wip", "moq-lite-07-wip"], NOT_FETCHABLE), + (&["moq-lite-07-wip", "moq-lite-06"], NOT_FOUND), + ] { + let refused = moq_net_sim::timeout(TEST_TIMEOUT, fetch_a_datagram(versions)) + .await + .unwrap_or_else(|_| panic!("{versions:?}: timed out")); + assert_eq!(refused, expected, "{versions:?}"); + } +} diff --git a/rs/moq-relay/src/web.rs b/rs/moq-relay/src/web.rs index c67aaa750c..8072df2023 100644 --- a/rs/moq-relay/src/web.rs +++ b/rs/moq-relay/src/web.rs @@ -964,7 +964,7 @@ async fn serve_fetch( Err( moq_net::Error::NotFound | moq_net::Error::NotFetchable - | moq_net::Error::Stream(moq_net::StreamError::NotFound), + | moq_net::Error::Stream(moq_net::StreamError::NotFound | moq_net::StreamError::NotFetchable), ) => { return Err(StatusCode::NOT_FOUND); } From 7b6f4691de777e168bd15fcd760535c48354d84d Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 12:52:58 -0700 Subject: [PATCH 7/8] fix(net): an absolute join skips a datagram group instead of failing A datagram object among an absolute joining FETCH's groups failed the whole fill, aborting the complete group before it and dropping the groups after. The join now finishes that head and skips the datagram; any other fill or a group fetch still refuses it as NotFetchable. Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/subscriber.rs | 110 +++++++++++++++++++++++++----- 1 file changed, 94 insertions(+), 16 deletions(-) diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 89b053ab42..c273980cfa 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3240,7 +3240,7 @@ where let mut prior_group = None; let mut first = true; while let Some(object) = decode_fetch_object(stream, self.version, std::mem::take(&mut first)).await? { - if !object.subgroup_ok { + if !object.datagram && !object.subgroup_ok { tracing::warn!("subgroup ID is not supported, dropping fill"); return Err(Error::Unsupported); } @@ -3251,6 +3251,21 @@ where prior_group = Some(sequence); } + if object.datagram { + // An absolute join spans whole groups, so a datagram among them is a group of + // its own: the head before it is complete, and the groups after it still fill. + // Any other fill is the one group, which a datagram cannot be. + if !matches!(joining, Some(JoiningFetch::Absolute { .. })) || group.is_none() { + tracing::debug!(?group, "a datagram group is not fetchable"); + return Err(Error::NotFetchable); + } + end_fill_group(head, largest)?; + // Draft-16 on, where a datagram can appear, a fetch object has no status field. + let mut remaining = usize::try_from(stream.varint().await?).map_err(|_| Error::FrameTooLarge)?; + std::future::poll_fn(|cx| stream.poll_skip(cx, &mut remaining)).await?; + continue; + } + match head.as_ref().map(|(sequence, next, _)| (*sequence, *next)) { None => { let (Some(sequence), Some(0)) = (group, object.object) else { @@ -3693,6 +3708,10 @@ where tracing::warn!(sequence, "a group fetch continued past its end marker"); return Err(Error::ProtocolViolation); } + if object.datagram { + tracing::debug!(sequence, "a datagram group is not fetchable"); + return Err(Error::NotFetchable); + } if !object.subgroup_ok { tracing::warn!("subgroup ID is not supported, dropping group fetch"); return Err(Error::Unsupported); @@ -3751,6 +3770,9 @@ where struct FetchedObject { group: Option, object: Option, + /// Draft-16 on lets a fetch carry an Object published as a datagram. A datagram group is + /// live-only, never cached, so it is never filled from a fetch. + datagram: bool, subgroup_ok: bool, properties: Option>, } @@ -3777,6 +3799,7 @@ async fn decode_fetch_object( return Ok(Some(FetchedObject { group: Some(group), object: Some(object), + datagram: false, subgroup_ok: subgroup == 0, properties: Some(properties), })); @@ -3795,13 +3818,6 @@ async fn decode_fetch_object( priority, properties, }) => { - // Draft-16 on lets a fetch carry an Object published as a datagram, but a - // datagram group is live-only: never cached, so never filled from a fetch. Refusing - // it fails only this fetch. - if subgroup == ietf::FetchSubgroup::Datagram { - tracing::debug!(?group, ?object, "a datagram group is not fetchable"); - return Err(Error::NotFetchable); - } let inherits = group.is_none() || object.is_none() || priority.is_none() @@ -3813,6 +3829,7 @@ async fn decode_fetch_object( Some(FetchedObject { group, object, + datagram: subgroup == ietf::FetchSubgroup::Datagram, subgroup_ok: matches!( subgroup, ietf::FetchSubgroup::Zero | ietf::FetchSubgroup::Prior | ietf::FetchSubgroup::Explicit(0) @@ -3868,8 +3885,17 @@ fn advance_fill_group( return Err(Error::Unsupported); } + end_fill_group(head, largest)?; + open_fill_group(track, head, sequence) +} + +/// Finish the group an absolute joining FETCH was writing, since a later group started. +fn end_fill_group( + head: &mut Option<(u64, u64, crate::recv::Group)>, + largest: Option, +) -> Result<(), Error> { let Some((prev, _, producer)) = head.take() else { - return open_fill_group(track, head, sequence); + return Ok(()); }; if largest.is_some_and(|largest| prev >= largest.group) { @@ -3879,7 +3905,7 @@ fn advance_fill_group( } producer.finish()?; - open_fill_group(track, head, sequence) + Ok(()) } /// Ready once a finished subscription's data streams are accounted for: Stream Count of @@ -8083,12 +8109,12 @@ mod stitch_tests { } } - /// One Object published as a datagram, as draft-16 on lets a fetch stream carry it. - fn datagram_object(sequence: u64) -> Vec { + /// A group's first fetch Object, with `group` as the wire's Group ID field. + fn first_object(subgroup: ietf::FetchSubgroup, group: u64, payload: &[u8]) -> Vec { let mut buf = Vec::new(); ietf::FetchObject::Object { - subgroup: ietf::FetchSubgroup::Datagram, - group: Some(sequence), + subgroup, + group: Some(group), object: Some(0), priority: Some(0), properties: None, @@ -8096,12 +8122,64 @@ mod stitch_tests { .encode(&mut crate::coding::Encoder::new(&mut buf, VERSION.into()), VERSION) .unwrap(); crate::coding::Encoder::new(&mut buf, VERSION.into()) - .varint(1u64) + .varint(payload.len() as u64) .unwrap(); - buf.extend_from_slice(b"d"); + buf.extend_from_slice(payload); buf } + /// One Object published as a datagram, as draft-16 on lets a fetch stream carry it. + fn datagram_object(sequence: u64) -> Vec { + first_object(ietf::FetchSubgroup::Datagram, sequence, b"d") + } + + /// An absolute join spans whole groups, so a datagram among them is skipped: the group + /// before it is complete, and the groups after it still fill and stitch into the tail. + #[moq_net_sim::test] + async fn an_absolute_join_skips_a_datagram_group() { + const START: u64 = 7; + const LIVE_GROUP: u64 = 9; + let largest = ietf::Location { + group: LIVE_GROUP, + object: 0, + }; + // Group 8 went out as a datagram. From draft-18 on a later Group ID is an ascending + // delta, so 7, 8, 9 is 7, 0, 0. + let mut fill = fill_stream_for(FETCH, &[(START, &[b"g7-0"])], false); + fill.extend(first_object(ietf::FetchSubgroup::Datagram, 0, b"g8")); + fill.extend(first_object(ietf::FetchSubgroup::Zero, 0, b"g9-0")); + + let h = Harness::new(Fill::Serving(None), vec![fill, tail_stream(LIVE_GROUP, 1, &[b"g9-1"])]).with_joining( + JoiningFetch::Absolute { group_id: START }, + FETCH, + largest, + ); + let mut consumer = h + .track + .subscribe(track::Subscription::default().with_max_delay(Duration::from_secs(60))); + + let mut fill = h.stream().await; + let mut tail = h.stream().await; + h.subscriber.clone().recv_fill(&mut fill).await.expect("absolute fetch"); + h.subscriber.clone().recv_group(&mut tail).await.expect("tail"); + + let mut groups = Vec::new(); + for _ in 0..2 { + let (sequence, frames) = read_group(&mut consumer).await; + groups.push(( + sequence, + frames.into_iter().map(|(_, payload)| payload).collect::>(), + )); + } + assert_eq!( + groups, + vec![ + (START, vec![b"g7-0".to_vec()]), + (LIVE_GROUP, vec![b"g9-0".to_vec(), b"g9-1".to_vec()]), + ] + ); + } + /// A datagram group is never fetchable, so a group fetch answered with one fails as /// not fetchable, and nothing is cached. #[moq_net_sim::test] From 85d0158afb446381512e575ccb8077f5f78ee597 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Wed, 7 Oct 2026 13:02:24 -0700 Subject: [PATCH 8/8] refactor(net): drop the datagram tick; a new subscriber may get the send buffer Strict live-only datagram delivery is not a goal: a new subscriber may read the datagrams still in the track's send buffer, as before. Removes model::datagram::Tick, Consumer::subscribe_since and the opened-at plumbing, reverts moq-c's open-before-return change, and rewords the docs and draft. The group range still bounds datagrams and FETCH still never returns one. Co-Authored-By: Claude Opus 5.5 --- doc/concept/moq-lite.md | 10 ++-- drafts/draft-lcurley-moq-lite.md | 7 +-- js/net/src/track.test.ts | 12 ----- js/net/src/track.ts | 6 +-- rs/moq-c/src/api.rs | 2 - rs/moq-c/src/consume.rs | 6 +-- rs/moq-net/src/error.rs | 4 +- rs/moq-net/src/ietf/subscriber.rs | 2 +- rs/moq-net/src/model/datagram.rs | 20 +------- rs/moq-net/src/model/resume.rs | 23 +++------ rs/moq-net/src/model/track.rs | 85 ++++++++----------------------- 11 files changed, 47 insertions(+), 130 deletions(-) diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 817689057a..888a564b67 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -375,11 +375,11 @@ datagram: unreliable, unordered, under about 1200 bytes, and never retransmitted. It suits real-time audio and sensor data. There is no stream fallback, so a datagram that doesn't fit isn't delivered that way. -Datagrams are live-only. A subscription gets the ones sent after it opened and -inside its group range, never a buffered backlog, and nothing caches them, so -`FETCH` never returns one. A publisher that knows the sequence was a datagram -refuses with `NOT_FETCHABLE` (moq-lite-07), otherwise with `NOT_FOUND`. Use a -group for anything a late joiner needs. +Nothing caches a datagram. A subscription gets the ones inside its group +range, and a new one may get the few still in the publisher's short send +buffer, but `FETCH` never returns one. A publisher that knows the sequence was +a datagram refuses with `NOT_FETCHABLE` (moq-lite-07), otherwise with +`NOT_FOUND`. Use a group for anything a late joiner needs. ## What moq-lite leaves out diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 152f883bcb..be2961d83a 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -658,8 +658,9 @@ The subscriber MAY cache the error and potentially retry later. QUIC datagrams provide unreliable, unordered delivery for latency-sensitive content that does not need retransmission. A publisher MAY transmit a Group consisting of exactly one Frame as a single QUIC datagram, in addition to (or instead of) opening a Group Stream, based on application hints, group size, and network conditions; a multi-frame Group is delivered via a Group Stream only. -Datagram delivery is live-only: a publisher sends a datagram only to the subscriptions active when it sends it, and only when its Group Sequence is inside the subscription's range, where a non-zero `Frame Start` excludes the start group (see [Positions](#positions)). -A datagram is never cached, retransmitted, or replayed to a later subscription, and FETCH never returns one. +A publisher sends a datagram to a subscription only when its Group Sequence is inside the subscription's range, where a non-zero `Frame Start` excludes the start group (see [Positions](#positions)). +A datagram is never cached or retransmitted, and FETCH never returns one. +A publisher MAY send a new subscription the datagrams still in a short send buffer, so a subscriber cannot assume a datagram was sent after it subscribed. A publisher that knows a FETCH names a group delivered only as a datagram resets the stream with NOT_FETCHABLE; otherwise it answers as for a group that does not exist, with NOT_FOUND. A publisher SHOULD only send a datagram if the congestion controller can transmit it immediately. There is no separate subscription for datagram delivery: datagrams are routed to existing subscriptions via the Subscribe ID, and a subscriber receiving the same group via both a stream and a datagram MUST deduplicate by group sequence. @@ -1385,7 +1386,7 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 -- Datagram delivery is live-only: a datagram goes only to the subscriptions active when it is sent and inside their range, and FETCH never returns one. +- A subscription's range bounds datagrams like groups, and FETCH never returns a datagram. - Assigned 0x3A NOT_FETCHABLE in the stream error table: a FETCH for a group delivered only as a datagram. - The subscriber FINs its Subscribe Stream after settling its tail; graceful session close waits for that FIN or reset. - A refusal is not retried at another route of the same prefix either. diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 323b86cfd0..9b615d3871 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -156,18 +156,6 @@ test("appendDatagram delivers to a subscriber", async () => { expect(got && dec.decode(got.payload)).toBe("hello"); }); -test("a late subscriber skips the datagram backlog", async () => { - const producer = new TrackProducer("test"); - const early = producer.subscribe(); - producer.appendDatagram(Timestamp.fromMillis(0), enc.encode("old")); - const late = producer.subscribe(); - producer.appendDatagram(Timestamp.fromMillis(0), enc.encode("new")); - - expect(dec.decode((await early.recvDatagram())?.payload)).toBe("old"); - expect(dec.decode((await early.recvDatagram())?.payload)).toBe("new"); - expect(dec.decode((await late.recvDatagram())?.payload)).toBe("new"); -}); - // The in-range datagrams, sent between the dropped ones, show the reader is live, so the // range is what dropped the others. test("the group range bounds datagrams", async () => { diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 40a5bb755e..0fd7b73855 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -1430,9 +1430,9 @@ export class Subscriber { * Receive the next datagram in arrival order. * * Datagrams are a separate best-effort channel from groups (see - * {@link Producer.appendDatagram}); they share only the sequence namespace. Only datagrams - * sent after the subscription opened arrive, and those outside its group range are skipped. - * A consumer that falls too far behind silently loses the oldest datagrams. Read this alongside + * {@link Producer.appendDatagram}); they share only the sequence namespace. Those outside + * the group range are skipped. A consumer that falls too far behind silently loses the + * oldest datagrams. Read this alongside * {@link recvGroup} (e.g. in a separate loop) to receive both channels concurrently. * The two cursors are independent: a datagram never moves the group cursor. */ diff --git a/rs/moq-c/src/api.rs b/rs/moq-c/src/api.rs index 7a4efed61d..c65b5079a5 100644 --- a/rs/moq-c/src/api.rs +++ b/rs/moq-c/src/api.rs @@ -3649,8 +3649,6 @@ pub extern "C" fn moq_consume_track_cancel(track: u32) -> i32 { /// [moq_consume_datagrams_cancel]. Read each datagram with [moq_consume_datagram] and release /// it with [moq_consume_datagram_free]. Datagrams arrive only over datagram-capable /// transports on moq-transport or lite-05 and newer moq-lite; there is no stream fallback. -/// Datagrams are live-only: the subscription opens before this returns, and gets only the -/// datagrams published after that. /// /// Returns a non-zero handle to the subscription on success, or a negative code on failure. /// diff --git a/rs/moq-c/src/consume.rs b/rs/moq-c/src/consume.rs index 2859fee023..fe6473e7a3 100644 --- a/rs/moq-c/src/consume.rs +++ b/rs/moq-c/src/consume.rs @@ -796,12 +796,10 @@ impl Consume { }; let id = self.datagram_task.insert(Some(entry))?; - // Open the subscription now: a datagram reaches only the subscriptions open when it is - // published. It resolves on SUBSCRIBE_OK, so await that inside the task. - let subscribing = broadcast.track(&name).map(|track| track.subscribe(None)); + // `subscribe` blocks on SUBSCRIBE_OK, so run it inside the task. tokio::spawn(async move { let res = async move { - let track = subscribing?.await?; + let track = broadcast.track(&name)?.subscribe(None).await?; Self::run_datagrams(on_datagram, track, channel.1).await } .await; diff --git a/rs/moq-net/src/error.rs b/rs/moq-net/src/error.rs index d8155f4812..ba8af63834 100644 --- a/rs/moq-net/src/error.rs +++ b/rs/moq-net/src/error.rs @@ -146,7 +146,7 @@ pub enum StreamError { #[error("not found")] NotFound, - /// A FETCH reached a datagram, which is live-only and never cached. Sent on moq-lite-07 + /// A FETCH reached a datagram, which is never cached. Sent on moq-lite-07 /// and later; an earlier version sends [`NotFound`](Self::NotFound) instead. #[error("not fetchable")] NotFetchable, @@ -329,7 +329,7 @@ pub enum Error { #[error("not found")] NotFound, - /// A FETCH reached a datagram, which is live-only and never served from a cache. + /// A FETCH reached a datagram, which is never cached. #[error("not fetchable")] NotFetchable, diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index c273980cfa..7d773bbc7c 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -3771,7 +3771,7 @@ struct FetchedObject { group: Option, object: Option, /// Draft-16 on lets a fetch carry an Object published as a datagram. A datagram group is - /// live-only, never cached, so it is never filled from a fetch. + /// never cached, so it is never filled from a fetch. datagram: bool, subgroup_ok: bool, properties: Option>, diff --git a/rs/moq-net/src/model/datagram.rs b/rs/moq-net/src/model/datagram.rs index d93226851a..d6a894988f 100644 --- a/rs/moq-net/src/model/datagram.rs +++ b/rs/moq-net/src/model/datagram.rs @@ -10,14 +10,12 @@ //! body exceeds the transport's datagram size, and sessions that can't carry datagrams at all //! (moq-lite before 05, or stream-only transports like WebSocket) never deliver them. //! -//! A datagram is live-only: it reaches the subscriptions open when it is pushed, and is never -//! cached, replayed to a later subscription, or served by a fetch. +//! A datagram is never cached or served by a fetch. Only the short per-track send buffer holds +//! it, which a new subscription may still read. //! //! Wire counterparts: [`crate::lite::Datagram`], and on moq-transport an OBJECT_DATAGRAM at //! object 0 whose Group ID is the sequence ([`crate::ietf::ObjectDatagram`]). -use std::sync::atomic::{AtomicU64, Ordering}; - use bytes::Bytes; use crate::Timestamp; @@ -43,17 +41,3 @@ pub struct Datagram { /// The datagram payload. pub payload: Bytes, } - -/// A point on the process-wide datagram clock, which orders every datagram push against every -/// subscription opening. A subscription takes the datagrams pushed after it opened, whichever -/// copy of the track (a relay's upstream, a replacement route) they reach it through. -#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord)] -pub(crate) struct Tick(u64); - -impl Tick { - /// A tick later than every one taken before it, on any thread. - pub(crate) fn next() -> Self { - static CLOCK: AtomicU64 = AtomicU64::new(0); - Self(CLOCK.fetch_add(1, Ordering::Relaxed)) - } -} diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index f9ede4e87f..5a8195b43c 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -20,8 +20,6 @@ use std::task::{Poll, ready}; use crate::{Datagram, Error, Result, StreamError, group, stats, track}; -use super::datagram::Tick; - use super::subscription::{Cap, Position, Subscription, max_some}; /// How many delivered group sequences a reader remembers, so a group two routes both @@ -138,9 +136,8 @@ impl Consumer { self.state.read().copy.clone() } - /// Subscribe to the logical track with the reader's own preferences, taking the - /// datagrams pushed to each copy since the reader `opened`. - pub(crate) fn subscribe(&self, subscription: kio::Producer, opened: Tick) -> Subscriber { + /// Subscribe to the logical track with the reader's own preferences. + pub(crate) fn subscribe(&self, subscription: kio::Producer) -> Subscriber { let mirrored = subscription.read().clone(); let reader = Arc::new_cyclic(|this| { Mutex::new(Reader { @@ -154,7 +151,6 @@ impl Consumer { newest: None, ordered: None, datagram: None, - opened, groups: (0, Cap::from(..)), }) }); @@ -246,8 +242,8 @@ struct Copy { done: Option>, /// The datagram channel ran out. datagrams_done: Option>, - /// The newest datagram handed out before this copy was subscribed: the copy also hands - /// over what it buffered since the reader opened, which the replaced copy may have too. + /// The newest datagram handed out before this copy was subscribed: a copy that + /// buffers datagrams replays them, and those were already handed out. datagrams_seen: Option, /// Held by every group handed out from this copy, so a replaced copy stays /// subscribed while one is still read. @@ -416,8 +412,6 @@ struct Reader { ordered: Option, /// The newest datagram handed out. datagram: Option, - /// When the reader subscribed: each copy hands it the datagrams pushed after this. - opened: Tick, /// The local group filter; see [`track::Subscriber::set_groups`]. groups: (u64, Cap), } @@ -470,7 +464,7 @@ impl Reader { }; self.copies.push(Copy { generation, - sub: Sub::Pending(track.subscribe_since(wire, self.opened)), + sub: Sub::Pending(track.subscribe(wire)), track, floor, until: None, @@ -1317,17 +1311,14 @@ mod test { assert!(matches!(reading.read_frame().now_or_never(), Some(Err(Error::Old)))); } - /// A new copy hands over what it buffered since the reader subscribed, once each, and - /// nothing it buffered before. #[test] fn datagrams_a_new_copy_replays_are_handed_out_once() { let routes = Producer::new(); let logical = logical(&routes); let mut a = copy(); routes.serve(a.consume()); - let mut b = copy(); - b.insert_datagram(9, ts(9), b"d".as_ref()).unwrap(); let mut sub = subscribe(&logical, Duration::from_secs(10)); + let mut b = copy(); for track in [&mut a, &mut b] { for sequence in [1, 2] { track.insert_datagram(sequence, ts(sequence), b"d".as_ref()).unwrap(); @@ -1340,7 +1331,7 @@ mod test { assert_eq!(recv(&mut sub), Some(1)); assert_eq!(recv(&mut sub), Some(2)); routes.serve(b.consume()); - assert_eq!(recv(&mut sub), Some(3), "9 predates the reader"); + assert_eq!(recv(&mut sub), Some(3)); assert_eq!(recv(&mut sub), None); } diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index 95397479da..4425797b87 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -16,7 +16,7 @@ use crate::{Error, Result, Timescale, Timestamp, coding}; use crate::{broadcast, cache, group, stats}; -use super::{Datagram, Requests, datagram::Tick}; +use super::{Datagram, Requests}; use super::Cap; pub use super::subscription::{Position, Subscription}; @@ -38,8 +38,7 @@ const DEFAULT_PRIORITY: u8 = 127; /// Maximum number of datagrams retained in the per-track send buffer. /// /// Datagrams are a best-effort send buffer, not a replay cache (unlike groups): only the last -/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog, and a -/// subscriber only gets those pushed after it opened. +/// 64 datagrams are kept, so a stalled consumer cannot retain an unbounded backlog. /// The payload size limit also bounds the buffer's memory use. const MAX_DATAGRAMS: usize = 64; @@ -208,9 +207,9 @@ pub(crate) struct TrackState { // what each track writes and never touches another track's cache. debt: u64, - // Datagrams in arrival order, bounded by `MAX_DATAGRAMS`, each stamped with when it was - // pushed. Shares the group sequence namespace but is otherwise independent. - datagrams: VecDeque<(Tick, Datagram)>, + // Datagrams in arrival order, bounded by `MAX_DATAGRAMS`. Shares the group + // sequence namespace but is otherwise independent. + datagrams: VecDeque, // Number of datagrams dropped off the front (over capacity), mapping a subscriber's absolute // cursor to an index into `datagrams` (mirrors `offset` for groups). @@ -495,7 +494,7 @@ impl TrackState { fn poll_recv_datagram(&self, index: usize) -> Poll>> { let start = index.saturating_sub(self.datagram_offset); if self.readable() - && let Some((_, datagram)) = self.datagrams.get(start) + && let Some(datagram) = self.datagrams.get(start) { return Poll::Ready(Ok(Some((datagram.clone(), self.datagram_offset + start)))); } @@ -510,15 +509,9 @@ impl TrackState { } } - /// The absolute index of the first buffered datagram pushed after `opened`: where a reader - /// opened then starts, since datagrams go only to the readers present when they are pushed. - fn datagram_start(&self, opened: Tick) -> usize { - self.datagram_offset + self.datagrams.partition_point(|(pushed, _)| *pushed < opened) - } - /// Whether `sequence` was sent as a datagram still in the send buffer. fn holds_datagram(&self, sequence: u64) -> bool { - self.datagrams.iter().any(|(_, datagram)| datagram.sequence == sequence) + self.datagrams.iter().any(|datagram| datagram.sequence == sequence) } /// Push a datagram, dropping the oldest when the send buffer is full. @@ -527,7 +520,7 @@ impl TrackState { self.datagrams.pop_front(); self.datagram_offset += 1; } - self.datagrams.push_back((Tick::next(), datagram)); + self.datagrams.push_back(datagram); } /// Find the smallest-sequence cached group satisfying @@ -1535,10 +1528,9 @@ impl Producer { /// session drops (with a debug log) any datagram whose encoded body exceeds the /// transport's datagram size, and sessions that can't carry datagrams at all (moq-lite /// before 05, or stream-only transports like WebSocket) never deliver them. Keep payloads well under the 1200-byte minimum path MTU. - /// A datagram is live-only: it reaches the subscribers present when it is sent, and is - /// never cached, replayed to a later subscriber, or served by a fetch. Use a group for - /// anything a late joiner needs. An origin publisher uses this; a relay preserving - /// upstream numbering uses [`Self::insert_datagram`]. + /// A datagram is never cached or served by a fetch, so use a group for anything a late + /// joiner needs. An origin publisher uses this; a relay preserving upstream numbering + /// uses [`Self::insert_datagram`]. pub fn append_datagram(&mut self, timestamp: Timestamp, payload: B) -> Result { let payload = payload.into_bytes(); if payload.len() > super::datagram::MAX_DATAGRAM_PAYLOAD { @@ -1890,7 +1882,7 @@ impl Producer { name: self.name.clone(), broadcast, info, - inner: Inner::Plain(Cursor::new(self.state.consume(), subscription, Tick::next())), + inner: Inner::Plain(Cursor::new(self.state.consume(), subscription)), // A producer-side (in-process) subscribe is not egress: stay untagged. stats: stats::Scope::default(), _stats_sub: stats::Subscription::default(), @@ -2600,13 +2592,7 @@ impl Consumer { /// the floor that the budget convicts, so the default budget of zero delivers only /// the latest group and a larger one reaches back over what it can still use. pub fn subscribe(&self, subscription: impl Into>) -> kio::Pending { - self.subscribe_since(subscription.into().unwrap_or_default(), Tick::next()) - } - - /// Open a live subscription taking the datagrams pushed since `opened`: a reader that - /// opened earlier on another copy of this track keeps what reached this one meanwhile. - pub(crate) fn subscribe_since(&self, subscription: Subscription, opened: Tick) -> kio::Pending { - let subscription = kio::Producer::new(subscription); + let subscription = kio::Producer::new(subscription.into().unwrap_or_default()); // Register the subscription if the track is live. If it is already closed, the // returned future resolves to the abort error via `Subscribing::poll_ok`. @@ -2617,7 +2603,6 @@ impl Consumer { broadcast: self.broadcast.clone(), state: self.state.clone(), subscription, - opened, stats: self.stats.clone(), }) } @@ -2905,8 +2890,6 @@ pub struct Subscribing { broadcast: Arc, state: kio::Consumer, subscription: kio::Producer, - /// When the subscription opened: its datagrams are the ones pushed after this. - opened: Tick, stats: stats::Scope, } @@ -2921,10 +2904,10 @@ impl Subscribing { let resume = self.state.read().routes.clone(); let inner = match resume { Some(resume) => Inner::Resume( - Box::new(resume.subscribe(self.subscription.clone(), self.opened)), + Box::new(resume.subscribe(self.subscription.clone())), self.state.clone(), ), - None => Inner::Plain(Cursor::new(self.state.clone(), self.subscription.clone(), self.opened)), + None => Inner::Plain(Cursor::new(self.state.clone(), self.subscription.clone())), }; Poll::Ready(Ok(Subscriber { name: self.name.clone(), @@ -3415,8 +3398,7 @@ struct Cursor { subscription: kio::Producer, /// Arrival-order cursor used by `recv_group`. index: usize, - /// Arrival-order cursor used by `recv_datagram`, independent of groups. Starts at the - /// first datagram pushed after the subscription opened, never at an older backlog. + /// Arrival-order cursor used by `recv_datagram`, independent of groups. datagram_index: usize, /// Minimum sequence to return from any `recv` method. Set by `start_at`. min_sequence: u64, @@ -3454,15 +3436,14 @@ struct Cursor { } impl Cursor { - fn new(state: kio::Consumer, subscription: kio::Producer, opened: Tick) -> Self { + fn new(state: kio::Consumer, subscription: kio::Producer) -> Self { let min_sequence = floor_of(&subscription.read()); - let datagram_index = state.read().datagram_start(opened); Self { state, subscription, min_sequence, index: 0, - datagram_index, + datagram_index: 0, next_sequence: 0, end_sequence: None, parked: BTreeMap::new(), @@ -3865,9 +3846,9 @@ impl Subscriber { /// /// Datagrams are a separate best-effort channel from groups (see /// [`Producer::append_datagram`]); they share only the sequence namespace, and - /// neither cursor moves the other. Only datagrams pushed after the subscription opened - /// arrive, and any outside this subscriber's group range are skipped. A consumer that - /// falls too far behind silently loses the oldest datagrams. + /// neither cursor moves the other. A new subscriber may get the few still in the send + /// buffer, and any outside its group range are skipped. A consumer that falls too far + /// behind silently loses the oldest datagrams. /// /// 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 @@ -4809,30 +4790,6 @@ mod test { assert_eq!(&recv_datagram(&mut dg).payload[..], b"go"); } - /// Datagrams are live-only: a subscriber gets those sent after it subscribed, never the - /// buffered backlog. A remote subscription counts from when it opened, not resolved. - #[test] - fn a_late_subscriber_skips_the_datagram_backlog() { - let mut producer = track_producer("test", None); - let consumer = producer.consume(); - let mut early = producer.subscribe(None); - let pending = consumer.subscribe(None); - producer.append_datagram(Timestamp::ZERO, &b"old"[..]).unwrap(); - - let mut opened = pending.now_or_never().unwrap().unwrap(); - let mut late = producer.subscribe(None); - let mut remote = consumer.subscribe(None).now_or_never().unwrap().unwrap(); - producer.append_datagram(Timestamp::ZERO, &b"new"[..]).unwrap(); - - for present in [&mut early, &mut opened] { - assert_eq!(&recv_datagram(present).payload[..], b"old"); - } - for subscriber in [&mut early, &mut opened, &mut late, &mut remote] { - assert_eq!(&recv_datagram(subscriber).payload[..], b"new"); - assert!(subscriber.poll_recv_datagram(&kio::Waiter::noop()).is_pending()); - } - } - /// A fetch never serves a datagram group, and says so rather than reporting a plain /// miss, whether or not a group followed it. #[test]