diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index a34ee6283e..036d0013fb 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -740,6 +740,38 @@ 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(); + } +}); + +// 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 () => { @@ -773,7 +805,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..8651b03952 100644 --- a/js/net/src/lite/publisher.ts +++ b/js/net/src/lite/publisher.ts @@ -1015,8 +1015,17 @@ 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. 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() ?? 0) + : 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..12c9b51572 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,11 @@ impl TrackRun { } self.start = Some(resolved.min(start)); } + } 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. 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)); @@ -4568,6 +4572,71 @@ 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); + } + + /// 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]