From 7a378e2e435e05b8195c9f583780115ac8aa54ea Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 12:53:30 -0700 Subject: [PATCH 1/2] fix(net): a capped IETF stream never claims END_OF_GROUP (#5077) Backport to release. The Rust regression test decodes the header with release's GroupHeader::decode form and runs under tokio::test, since release lacks coding::decode_buf and moq_net_sim. Quest edits dropped. (cherry picked from commit 9680ca9e412d9ecefef7aec4ce8a7102bfca7bf9) Co-Authored-By: Claude Opus 5.5 --- js/net/src/ietf/publisher.test.ts | 18 ++++++++++++--- js/net/src/ietf/publisher.ts | 4 +++- rs/moq-net/src/ietf/publisher.rs | 37 +++++++++++++++++++++++++++++++ 3 files changed, 55 insertions(+), 4 deletions(-) diff --git a/js/net/src/ietf/publisher.test.ts b/js/net/src/ietf/publisher.test.ts index b7a4a0889e..9f94be6bdc 100644 --- a/js/net/src/ietf/publisher.test.ts +++ b/js/net/src/ietf/publisher.test.ts @@ -1492,6 +1492,8 @@ interface ServedGroup { sequence: number; /** Whether the header claimed the stream starts at the group's first object. */ firstObject: boolean; + /** Whether the header claimed the stream's FIN ends the group (END_OF_GROUP). */ + endOfGroup: boolean; /** Each object's absolute id (reconstructed from its delta) and payload. */ objects: { id: number; payload: string }[]; } @@ -1623,7 +1625,12 @@ async function readGroup(stream: ReadableStream): Promise): Promise /** * An absolute filter names the objects it wants, so the boundary groups are trimmed to it * and the groups outside it are never opened. The first object written carries its absolute - * id, or the subscriber would read a silently renumbered group. + * id, or the subscriber would read a silently renumbered group, and a capped tail does not + * claim END_OF_GROUP, or the subscriber would think the group ended at the cap. */ test("draft-20: an absolute filter trims the range it serves", async () => { const fx = fixture(); @@ -1715,6 +1723,7 @@ test("draft-20: an absolute filter trims the range it serves", async () => { sequence: 1, // The head was trimmed, so the stream does not start at the group's first object. firstObject: false, + endOfGroup: true, objects: [ { id: 1, payload: "1.1" }, { id: 2, payload: "1.2" }, @@ -1726,6 +1735,8 @@ test("draft-20: an absolute filter trims the range it serves", async () => { expect(await readGroup(second)).toEqual({ sequence: 2, firstObject: true, + // The filter ends at object 0 of 3, so the stream stops before the group does. + endOfGroup: false, objects: [{ id: 0, payload: "2.0" }], }); @@ -1787,6 +1798,7 @@ test("draft-20: a fill serves the current group's head on a fetch stream", async expect(await readGroup(live)).toEqual({ sequence: 0, firstObject: false, + endOfGroup: true, objects: [{ id: 2, payload: "0.2" }], }); } finally { @@ -1863,7 +1875,7 @@ test("draft-20: a backwards range within one group serves nothing and ends the s try { const served = await nextUni(fx.uni); if (!served) throw new Error("the group stream never opened"); - expect(await readGroup(served)).toEqual({ sequence: 0, firstObject: false, objects: [] }); + expect(await readGroup(served)).toEqual({ sequence: 0, firstObject: false, endOfGroup: false, objects: [] }); } finally { fx.close(); client.close(); diff --git a/js/net/src/ietf/publisher.ts b/js/net/src/ietf/publisher.ts index ae3ff38850..5d39644a22 100644 --- a/js/net/src/ietf/publisher.ts +++ b/js/net/src/ietf/publisher.ts @@ -591,7 +591,9 @@ export class Publisher { hasExtensions: stamped, hasSubgroup: false, hasSubgroupObject: false, - hasEnd: true, + // A stream capped by the range stops before the group may end, so its FIN + // cannot claim END_OF_GROUP. + hasEnd: slice.until === undefined, hasPriority: true, // Only honest when the stream really starts at the group's first object; // a trimmed head starts partway through. diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index d42749c8ea..cd218539e1 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -2315,6 +2315,9 @@ impl TrackServe { flags: ietf::GroupFlags { has_extensions: self.timescale.is_some(), first_object: slice.skip == 0, + // A stream capped by the range stops before the group may end, so its + // FIN cannot claim END_OF_GROUP. + has_end: slice.until.is_none(), ..Default::default() }, }; @@ -4394,6 +4397,40 @@ mod serve_tests { "the cap excludes cc" ); } + + /// A stream cut short by the range's end Location does not claim END_OF_GROUP, since + /// the group goes on past it. A range ending on a whole group still does. A cap at or past + /// the group's real end clears the bit too, since the header is written before we know. + #[tokio::test] + async fn capped_group_does_not_claim_its_end() { + async fn serve(object: Option) -> ietf::GroupHeader { + let log = crate::lite::test_transport::Log::default(); + let session = SinkSession::new(log.clone()); + + let track = track::Producer::new(std::sync::Arc::new(crate::broadcast::Info::default()), "test", None); + let subscriber = track.subscribe(None); + + let mut group = track.append_group().unwrap(); + for payload in [b"aa", b"bb", b"cc"] { + group.write_frame(crate::Timestamp::ZERO, payload.as_slice()).unwrap(); + } + group.finish().unwrap(); + track.finish().unwrap(); + + let range = ServeRange { + start: Some(Location { group: 0, object: 0 }), + end: Some(EndLocation { group: 0, object }), + }; + let mut serve = TrackServe::new(session, subscriber, RequestId(0), Version::Draft20, range, None); + kio::wait(|waiter| serve.poll(waiter)).await.unwrap(); + + let mut buf = bytes::Bytes::from(log.writes.lock().unwrap().clone()); + ietf::GroupHeader::decode(&mut buf, Version::Draft20).expect("a group header") + } + + assert!(!serve(Some(0)).await.flags.has_end, "capped at object 0 of 3"); + assert!(serve(None).await.flags.has_end, "the whole group"); + } } #[cfg(test)] From 1aec3e6fe510d5935b14573f2e35248c1a8a4487 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Fri, 9 Oct 2026 17:17:13 -0700 Subject: [PATCH 2/2] test(net): pin a cap at or past a group's end clearing END_OF_GROUP Co-Authored-By: Claude Opus 5.5 --- rs/moq-net/src/ietf/publisher.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rs/moq-net/src/ietf/publisher.rs b/rs/moq-net/src/ietf/publisher.rs index cd218539e1..a3bf47e7b0 100644 --- a/rs/moq-net/src/ietf/publisher.rs +++ b/rs/moq-net/src/ietf/publisher.rs @@ -4429,6 +4429,8 @@ mod serve_tests { } assert!(!serve(Some(0)).await.flags.has_end, "capped at object 0 of 3"); + assert!(!serve(Some(2)).await.flags.has_end, "capped at the last object"); + assert!(!serve(Some(3)).await.flags.has_end, "capped past the last object"); assert!(serve(None).await.flags.has_end, "the whole group"); } }