From d60481d478b728e24fa34136811f338291566cda Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 19:21:42 -0700 Subject: [PATCH 1/2] fix(net): re-pin a pre-06 cursor when an update drops its floor Pre-06 drafts define an absent Group Start as the latest group, and position_cursor applies that on SUBSCRIBE. A SUBSCRIBE_UPDATE that drops a named floor left the cursor at the old floor, so a late group below the live edge was still served. Re-pin to the latest group in Rust and JS. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/publisher.test.ts | 21 ++++++++++++++- js/net/src/lite/publisher.ts | 10 ++++++- rs/moq-net/src/lite/publisher.rs | 45 +++++++++++++++++++++++++++++-- 3 files changed, 72 insertions(+), 4 deletions(-) diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index a34ee6283e..1a68a3c378 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -740,6 +740,24 @@ test("lite draft-05: an explicit floor still serves a group below the first one" } }); +// Pre-06 an absent Group Start is the latest group. An update that drops a named floor +// re-pins there, so a group created below it afterwards is not served. +test("lite draft-05: an update dropping the floor re-pins to the latest group", async () => { + const sub = await servedSubscription({ startGroup: 0 }); + try { + sub.serve(2); + expect(await sub.servedSequence()).toBe(2); + + await replayUpdate({ priority: 0 }).encode(sub.client.writer, Version.DRAFT_05); + await flush(); + sub.serve(0); + sub.serve(3); + expect(await sub.servedSequence()).toBe(3); + } finally { + await sub.close(); + } +}); + // Lite-06 encodes a floor of group 0 as 0, including a subscribe that omitted Group Start. // That is a floor, so a later group 0 is still served. test("lite draft-06: an omitted group start still serves a group below the first one", async () => { @@ -773,7 +791,8 @@ test("lite draft-05: a group popped before a cap update is still served", async sub.release(); expect(await sub.servedSequence()).toBe(1); while (ranges.mock.calls.length === 0) await flush(); - expect(lastGroups(ranges)).toEqual({ start: undefined, end: { included: 0 } }); + // Dropping the named floor re-pins to the latest group, as an unfloored SUBSCRIBE does. + expect(lastGroups(ranges)).toEqual({ start: { included: 1 }, end: { included: 0 } }); } finally { ranges.mockRestore(); sub.release(); diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index 55014421be..deff89c7ac 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -1015,8 +1015,16 @@ export class Publisher { case "update": { const update = control.update; console.debug(`subscribe update: broadcast=${broadcast} track=${track.name}`); + // `positionCursor` pins an unfloored pre-06 SUBSCRIBE to the latest + // group, so an update that drops a named floor re-pins there too. + const start = + update.startGroup === undefined && + bounds.startGroup !== undefined && + !resolvesStart(this.version) + ? track.latest() + : update.startGroup; hooks.replaceGroups(track, { - start: update.startGroup === undefined ? undefined : { included: update.startGroup }, + start: start === undefined ? undefined : { included: start }, end: update.endGroup === undefined ? undefined : { included: update.endGroup }, }); bounds.startGroup = update.startGroup; diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 0bc24118f7..124e70c55f 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3244,8 +3244,7 @@ impl TrackRun { // cursor that was already at 0 stays where the first served group put it. // No fresh SUBSCRIBE_START follows: the subscriber clears its permanent-miss // floor on the same update, so both sides have to change together. - // Pre-06 an absent start means the latest group, which `position_cursor` - // already applied. + // Pre-06 an absent start means the latest group: see the re-pin below. let lowered = match upd.start_group { Some(start_group) => Some(start_group), None if floored && self.ctx.version.resolves_start() => Some(0), @@ -3266,6 +3265,13 @@ impl TrackRun { } self.start = Some(resolved.min(start)); } + } else if requested.is_some() + && !self.ctx.version.resolves_start() + && let Some(latest) = self.track.latest() + { + // `position_cursor` pins an unfloored SUBSCRIBE to the latest group, so an + // update that drops a named floor re-pins there too. + self.track.start_at(latest); } self.track .end_at(upd.end_group.map_or(Bound::Unbounded, Bound::Included)); @@ -4568,6 +4574,41 @@ mod serve_group_test { assert_eq!(drops(writer, &log, version).await, [(1, 4)]); } + /// Pre-06 an absent Group Start is the latest group, so an update that drops a named + /// floor re-pins the cursor there: a group created below it afterwards is not served. + #[moq_net_sim::test] + async fn a_pre06_update_dropping_the_floor_repins_to_the_latest_group() { + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe( + track::Subscription::default() + .with_start(track::Position::group(0)) + .with_max_delay(Duration::from_secs(60)), + ); + let (mut run, mut writer, _log) = lite_run(Version::Lite05, SinkSession::new(Log::default()), subscriber); + let mut settle = |run: &mut TrackRun| { + let res = kio::wait(|waiter| run.poll(&mut writer, waiter)).now_or_never(); + assert!(res.is_none(), "the run ended"); + run.ctx.opens.opened.load(Ordering::Relaxed) + }; + + write_group(&mut track, 7, 7); + assert_eq!(settle(&mut run), 1); + + run.update(lite::SubscribeUpdate { + priority: 0, + max_delay: Duration::from_secs(60), + start_group: None, + end_group: None, + start_frame: 0, + end_frame: None, + }); + write_group(&mut track, 5, 5); + assert_eq!(settle(&mut run), 1, "served a group below the latest"); + + write_group(&mut track, 8, 8); + assert_eq!(settle(&mut run), 2); + } + /// A gap older than the subscriber's grace is one it no longer waits for, so the run loop /// forgets it rather than holding every gap of a long track until the end. #[moq_net_sim::test] From fcc4830fb1008ec2790e020b8e5b50ccefd08934 Mon Sep 17 00:00:00 2001 From: Luke Curley Date: Sat, 10 Oct 2026 20:04:58 -0700 Subject: [PATCH 2/2] fix(net): re-pin an empty pre-06 track to the next group With no latest group, an update that drops a named floor kept the old floor, so a first group below it was skipped. Pin to 0 instead: on an empty track the next group is the latest. Found by CodeRabbit. Co-Authored-By: Claude Opus 5.5 --- js/net/src/lite/publisher.test.ts | 14 +++++++++++ js/net/src/lite/publisher.ts | 5 ++-- rs/moq-net/src/lite/publisher.rs | 40 ++++++++++++++++++++++++++----- 3 files changed, 51 insertions(+), 8 deletions(-) diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index 1a68a3c378..036d0013fb 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -758,6 +758,20 @@ test("lite draft-05: an update dropping the floor re-pins to the latest group", } }); +// On an empty track the re-pin has no latest group to name, so the first group the +// track produces is served, even below the dropped floor. +test("lite draft-05: an update dropping the floor on an empty track serves the next group", async () => { + const sub = await servedSubscription({ startGroup: 5 }); + try { + await replayUpdate({ priority: 0 }).encode(sub.client.writer, Version.DRAFT_05); + await flush(); + sub.serve(0); + expect(await sub.servedSequence()).toBe(0); + } finally { + await sub.close(); + } +}); + // Lite-06 encodes a floor of group 0 as 0, including a subscribe that omitted Group Start. // That is a floor, so a later group 0 is still served. test("lite draft-06: an omitted group start still serves a group below the first one", async () => { diff --git a/js/net/src/lite/publisher.ts b/js/net/src/lite/publisher.ts index deff89c7ac..8651b03952 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -1016,12 +1016,13 @@ export class Publisher { const update = control.update; console.debug(`subscribe update: broadcast=${broadcast} track=${track.name}`); // `positionCursor` pins an unfloored pre-06 SUBSCRIBE to the latest - // group, so an update that drops a named floor re-pins there too. + // group, so an update that drops a named floor re-pins there too. On + // an empty track the next group is the latest, whatever its sequence. const start = update.startGroup === undefined && bounds.startGroup !== undefined && !resolvesStart(this.version) - ? track.latest() + ? (track.latest() ?? 0) : update.startGroup; hooks.replaceGroups(track, { start: start === undefined ? undefined : { included: start }, diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 124e70c55f..12c9b51572 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -3265,13 +3265,11 @@ impl TrackRun { } self.start = Some(resolved.min(start)); } - } else if requested.is_some() - && !self.ctx.version.resolves_start() - && let Some(latest) = self.track.latest() - { + } else if requested.is_some() && !self.ctx.version.resolves_start() { // `position_cursor` pins an unfloored SUBSCRIBE to the latest group, so an - // update that drops a named floor re-pins there too. - self.track.start_at(latest); + // update that drops a named floor re-pins there too. On an empty track the + // next group is the latest, whatever its sequence. + self.track.start_at(self.track.latest().unwrap_or(0)); } self.track .end_at(upd.end_group.map_or(Bound::Unbounded, Bound::Included)); @@ -4609,6 +4607,36 @@ mod serve_group_test { assert_eq!(settle(&mut run), 2); } + /// On an empty track the re-pin has no latest group to name, so the first group the + /// track produces is served, even below the dropped floor. + #[moq_net_sim::test] + async fn a_pre06_update_dropping_the_floor_on_an_empty_track_serves_the_next_group() { + let mut track = track::Producer::new(Arc::new(broadcast::Info::default()), "test", None); + let subscriber = track.subscribe( + track::Subscription::default() + .with_start(track::Position::group(5)) + .with_max_delay(Duration::from_secs(60)), + ); + let (mut run, mut writer, _log) = lite_run(Version::Lite05, SinkSession::new(Log::default()), subscriber); + run.update(lite::SubscribeUpdate { + priority: 0, + max_delay: Duration::from_secs(60), + start_group: None, + end_group: None, + start_frame: 0, + end_frame: None, + }); + + write_group(&mut track, 0, 0); + let res = kio::wait(|waiter| run.poll(&mut writer, waiter)).now_or_never(); + assert!(res.is_none(), "the run ended"); + assert_eq!( + run.ctx.opens.opened.load(Ordering::Relaxed), + 1, + "group 0 was not served" + ); + } + /// A gap older than the subscriber's grace is one it no longer waits for, so the run loop /// forgets it rather than holding every gap of a long track until the end. #[moq_net_sim::test]