diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index 9f14e3659ab..2752bc1f0da 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -34,6 +34,19 @@ Rust and TypeScript speak moq-lite 01 through 06 and moq-transport drafts still in progress: it negotiates as `moq-lite-07-wip`, and only when both sides explicitly enable it. +## Subscription completion + +On moq-lite 07, `SUBSCRIBE_END` counts the group streams opened for the +subscription. Rust and TypeScript stop waiting for missing streams once that +many headers have arrived; skipped group sequences add no wait. Groups already +being received continue until their own stream ends or resets. + +A stream reset before its header arrived cannot be counted, so the subscriber +still allows a grace period for late streams. The grace uses the subscription's +nonzero effective maximum age, or one second when no maximum age is set. +moq-lite 05 and 06 instead account for group sequences using received headers +and `SUBSCRIBE_DROP`. + ## Discovery A session can ask for announcements matching a path prefix. The peer replies diff --git a/js/net/src/lite/subscriber.ts b/js/net/src/lite/subscriber.ts index b031f4044fd..7640b3ce322 100644 --- a/js/net/src/lite/subscriber.ts +++ b/js/net/src/lite/subscriber.ts @@ -40,7 +40,15 @@ import { SubscribeUpdate, } from "./subscribe.ts"; import { TrackInfo, Track as TrackMessage } from "./track.ts"; -import { hasAnnounceId, hasAnnounceOk, hasDatagrams, hasProbeRtt, restartSupported, Version } from "./version.ts"; +import { + hasAnnounceId, + hasAnnounceOk, + hasDatagrams, + hasProbeRtt, + hasStreamCount, + restartSupported, + Version, +} from "./version.ts"; // Bound on how long stream-open plus the first response (SUBSCRIBE_OK on older // drafts, or TRACK_INFO on lite-05+) may take. Browsers cap concurrent QUIC streams @@ -83,6 +91,8 @@ interface SubscribeEntry { // (SUBSCRIBE_END), once it declares them. start?: number; end?: number; + // Group streams opened by the publisher, when SUBSCRIBE_END carries the count. + streams?: number; } /** @@ -833,6 +843,7 @@ export class Subscriber { } else if ("end" in resp) { if (entry.end !== undefined) throw new ProtocolViolation("duplicate SUBSCRIBE_END"); entry.end = resp.end.group; + if (hasStreamCount(this.version)) entry.streams = resp.end.streams; // A local close can win the race with the response; there is nothing left to end. if (entry.track.closed.peek() !== undefined) continue; try { @@ -848,13 +859,10 @@ export class Subscriber { // Wait for the group streams the publisher still owes once it has ended the subscription. // - // Its FIN says every group below the end is accounted for, but QUIC does not order streams, - // so one can still be in flight. Wait until each group from SUBSCRIBE_START to the end has - // a stream (read to its end) or a SUBSCRIBE_DROP. A group reset before its header arrived - // never shows up, so give up on missing groups after the subscription's effective max age, - // then end cleanly with them skipped like any stale group. That is a wall-clock stopgap for - // a presentation-time budget; a publisher sending SUBSCRIBE_DROP for every group it reset - // would account for them with no timer at all. + // lite-07 counts streams, so skipped sequences owe nothing. Older drafts account for + // the range using received headers and SUBSCRIBE_DROP. A counted stream reset before + // its header leaves no trace, so the grace still bounds that wait. Streams whose + // headers arrived keep reading until their own FIN or reset. #settleTail(entry: SubscribeEntry): Promise { const { tail, track } = entry; // Already the smaller of the subscriber's and the track's max age. @@ -862,6 +870,7 @@ export class Subscriber { const grace = maxAge > 0 ? maxAge : TAIL_GRACE_MS; const complete = () => { + if (entry.streams !== undefined) return tail.streams >= entry.streams; // Without SUBSCRIBE_END (older drafts) nothing says which groups are owed. if (entry.end === undefined) return false; // Without SUBSCRIBE_START the publisher served no group at all. diff --git a/js/net/src/lite/tail.test.ts b/js/net/src/lite/tail.test.ts index 4edfe88c5c1..245791287da 100644 --- a/js/net/src/lite/tail.test.ts +++ b/js/net/src/lite/tail.test.ts @@ -1,4 +1,4 @@ -import { expect, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { ProtocolViolation, StreamCode, StreamError } from "../error.ts"; import type { Consumer as GroupConsumer } from "../group.ts"; import { randomHop } from "../hop.ts"; @@ -20,12 +20,10 @@ import { Subscriber } from "./subscriber.ts"; import { TrackInfo, Track as TrackMessage } from "./track.ts"; import { ALPN_05, Version } from "./version.ts"; -const VERSION = Version.DRAFT_05; - // The subscription's max age, which is also how long it waits for a group that never arrives. const GRACE = Milli(100); -/** One lite-05 frame: a zero timestamp delta, then the length-prefixed payload. */ +/** One lite-05+ frame: a zero timestamp delta, then the length-prefixed payload. */ function frame(payload: string): Uint8Array { const bytes = new TextEncoder().encode(payload); // Every field is under 64, so each is a one-byte varint. @@ -49,31 +47,31 @@ function groupStream(subscriber: Subscriber, sequence: number) { } /** - * A lite-05 subscriber with one track subscribed, whose publisher the test plays by hand: + * A lite-05+ subscriber with one track subscribed, whose publisher the test plays by hand: * it answers TRACK_INFO, then writes whatever responses the test asks for on the subscribe * stream and FINs it when told. */ -async function subscribed(maxAge = GRACE) { +async function subscribed(version: Version, maxAge = GRACE) { const pair = createMockTransportPair(ALPN_05); - const subscriber = new Subscriber(pair.client, VERSION, randomHop()); + const subscriber = new Subscriber(pair.client, version, randomHop()); const reader = subscriber.consume(Path.from("room")).track("video").subscribe({ maxAge }); const info = await Stream.accept(pair.server); if (!info) throw new Error("the subscriber never asked for TRACK_INFO"); expect(await info.reader.u53()).toBe(StreamId.Track); - await TrackMessage.decode(info.reader, VERSION); - await new TrackInfo({ maxAge: 60_000 }).encode(info.writer, VERSION); + await TrackMessage.decode(info.reader, version); + await new TrackInfo({ maxAge: 60_000 }).encode(info.writer, version); info.close(); const sub = await Stream.accept(pair.server); if (!sub) throw new Error("the subscriber never subscribed"); expect(await sub.reader.u53()).toBe(StreamId.Subscribe); - await Subscribe.decode(sub.reader, VERSION); + await Subscribe.decode(sub.reader, version); return { subscriber, reader, - respond: (resp: SubscribeResponse) => encodeSubscribeResponse(sub.writer, resp, VERSION), + respond: (resp: SubscribeResponse) => encodeSubscribeResponse(sub.writer, resp, version), fin: () => sub.writer.close(), reset: (error: Error) => sub.writer.reset(error), }; @@ -103,126 +101,160 @@ async function settlesWithin(promise: Promise, ms: number): Promise { - const { subscriber, reader, respond, fin } = await subscribed(); - await respond({ start: new SubscribeStart(0) }); - const first = groupStream(subscriber, 0); - first.write("0.0"); - first.finish(); - await respond({ end: new SubscribeEnd(2) }); - await fin(); +describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (version) => { + test("a group stream that arrives after the subscribe stream's FIN is delivered", async () => { + const { subscriber, reader, respond, fin } = await subscribed(version); + await respond({ start: new SubscribeStart(0) }); + const first = groupStream(subscriber, 0); + first.write("0.0"); + first.finish(); + await respond({ end: new SubscribeEnd(2, 2) }); + await fin(); - // The end is known before the last group arrives. - expect(await reader.finished()).toBe(2); + // The end is known before the last group arrives. + expect(await reader.finished()).toBe(2); - // QUIC does not order streams, so group 1 lands after the FIN. - const late = groupStream(subscriber, 1); - late.write("1.0"); - late.finish(); + // QUIC does not order streams, so group 1 lands after the FIN. + const late = groupStream(subscriber, 1); + late.write("1.0"); + late.finish(); - expect(await readAll(await reader.recvGroup())).toEqual(["0.0"]); - expect(await readAll(await reader.recvGroup())).toEqual(["1.0"]); - expect(await reader.recvGroup()).toBeUndefined(); - expect(await reader.closed).toBeNull(); - expect(reader.final()).toBe(2); -}); + expect(await readAll(await reader.recvGroup())).toEqual(["0.0"]); + expect(await readAll(await reader.recvGroup())).toEqual(["1.0"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(2); + }); -test("a group read across the subscribe stream's FIN is delivered whole", async () => { - const { subscriber, reader, respond, fin } = await subscribed(); - await respond({ start: new SubscribeStart(0) }); - const group = groupStream(subscriber, 0); - group.write("0.0"); + test("a group read across the subscribe stream's FIN is delivered whole", async () => { + const { subscriber, reader, respond, fin } = await subscribed(version); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 0); + group.write("0.0"); - await respond({ end: new SubscribeEnd(1) }); - await fin(); - const received = await reader.recvGroup(); - expect(await received?.readString()).toBe("0.0"); - - // The FIN does not end the group: only its own stream does. - group.write("0.1"); - group.finish(); - expect(await readAll(received)).toEqual(["0.1"]); - expect(await reader.recvGroup()).toBeUndefined(); - expect(await reader.closed).toBeNull(); -}); + await respond({ end: new SubscribeEnd(1, 1) }); + await fin(); + const received = await reader.recvGroup(); + expect(await received?.readString()).toBe("0.0"); -test("a group reset after the subscribe stream's FIN is not presented as complete", async () => { - const { subscriber, reader, respond, fin } = await subscribed(); - await respond({ start: new SubscribeStart(0) }); - const group = groupStream(subscriber, 0); - group.write("0.0"); - await respond({ end: new SubscribeEnd(1) }); - await fin(); + // The FIN does not end the group: only its own stream does. + group.write("0.1"); + group.finish(); + expect(await readAll(received)).toEqual(["0.1"]); + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); + }); - const received = await reader.recvGroup(); - expect(await received?.readString()).toBe("0.0"); - group.reset(); - await expect(received?.readString() ?? Promise.resolve()).rejects.toThrow(); + test("a group reset after the subscribe stream's FIN is not presented as complete", async () => { + const { subscriber, reader, respond, fin } = await subscribed(version); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 0); + group.write("0.0"); + await respond({ end: new SubscribeEnd(1, 1) }); + await fin(); - // The track still ends cleanly: the group was accounted for, as a reset. - expect(await reader.recvGroup()).toBeUndefined(); - expect(await reader.closed).toBeNull(); -}); + const received = await reader.recvGroup(); + expect(await received?.readString()).toBe("0.0"); + group.reset(); + await expect(received?.readString() ?? Promise.resolve()).rejects.toThrow(); -test("a group that never arrives is given up on after the subscription's max age", async () => { - const { subscriber, reader, respond, fin } = await subscribed(); - await respond({ start: new SubscribeStart(0) }); - const group = groupStream(subscriber, 1); - group.write("1.0"); - group.finish(); - await respond({ end: new SubscribeEnd(2) }); - await fin(); + // The track still ends cleanly: the group was accounted for, as a reset. + expect(await reader.recvGroup()).toBeUndefined(); + expect(await reader.closed).toBeNull(); + }); - // Group 0 was reset before its header arrived, so nothing ever accounts for it. - expect((await reader.recvGroup())?.sequence).toBe(1); - const started = performance.now(); - expect(await reader.recvGroup()).toBeUndefined(); - expect(performance.now() - started).toBeGreaterThanOrEqual(GRACE - 5); - expect(await reader.closed).toBeNull(); - expect(reader.final()).toBe(2); + test("a group that never arrives is given up on after the subscription's max age", async () => { + const { subscriber, reader, respond, fin } = await subscribed(version); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 1); + group.write("1.0"); + group.finish(); + await respond({ end: new SubscribeEnd(2, 2) }); + await fin(); + + // Group 0 was reset before its header arrived, so nothing ever accounts for it. + expect((await reader.recvGroup())?.sequence).toBe(1); + const started = performance.now(); + expect(await reader.recvGroup()).toBeUndefined(); + expect(performance.now() - started).toBeGreaterThanOrEqual(GRACE - 5); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(2); + }); + + test.skipIf(version === Version.DRAFT_07)( + "a subscription ends without waiting once every group is accounted for", + async () => { + // A max age far past the test's patience: only the accounting may end it. + const { subscriber, reader, respond, fin } = await subscribed(version, Milli(60_000)); + await respond({ start: new SubscribeStart(0) }); + await respond({ drop: new SubscribeDrop({ start: 0, end: 0, error: 0 }) }); + const group = groupStream(subscriber, 1); + group.write("1.0"); + group.finish(); + await respond({ end: new SubscribeEnd(2, 2) }); + await fin(); + + expect((await reader.recvGroup())?.sequence).toBe(1); + expect(await settlesWithin(reader.recvGroup(), 1000)).toBe(true); + expect(await reader.closed).toBeNull(); + }, + ); + + test("a subscription that served nothing ends at SUBSCRIBE_END", async () => { + const { reader, respond, fin } = await subscribed(version, Milli(60_000)); + await respond({ end: new SubscribeEnd(0) }); + await fin(); + + expect(await settlesWithin(reader.recvGroup(), 1000)).toBe(true); + expect(await reader.closed).toBeNull(); + expect(reader.final()).toBe(0); + }); + + test("a SUBSCRIBE_END below a group already received aborts the track", async () => { + const { subscriber, reader, respond } = await subscribed(version); + await respond({ start: new SubscribeStart(0) }); + const group = groupStream(subscriber, 3); + group.finish(); + await group.handled; + await respond({ end: new SubscribeEnd(2, 2) }); + + const closed = await reader.closed; + expect(closed).toBeInstanceOf(Error); + }); }); -test("a subscription ends without waiting once every group is accounted for", async () => { - // A max age far past the test's patience: only the accounting may end it. - const { subscriber, reader, respond, fin } = await subscribed(Milli(60_000)); +test("lite-07 ends without grace when every counted stream arrived despite a skipped group", async () => { + const { subscriber, reader, respond, fin } = await subscribed(Version.DRAFT_07, Milli(60_000)); await respond({ start: new SubscribeStart(0) }); - await respond({ drop: new SubscribeDrop({ start: 0, end: 0, error: 0 }) }); - const group = groupStream(subscriber, 1); - group.write("1.0"); - group.finish(); - await respond({ end: new SubscribeEnd(2) }); + for (const sequence of [0, 2]) { + const group = groupStream(subscriber, sequence); + group.write(`${sequence}.0`); + group.finish(); + await group.handled; + } + await respond({ end: new SubscribeEnd(3, 2) }); await fin(); - expect((await reader.recvGroup())?.sequence).toBe(1); + expect((await reader.recvGroup())?.sequence).toBe(0); + expect((await reader.recvGroup())?.sequence).toBe(2); expect(await settlesWithin(reader.recvGroup(), 1000)).toBe(true); expect(await reader.closed).toBeNull(); }); -test("a subscription that served nothing ends at SUBSCRIBE_END", async () => { - const { reader, respond, fin } = await subscribed(Milli(60_000)); - await respond({ end: new SubscribeEnd(0) }); +test("lite-07 zero count ends without grace even when the announced range is nonempty", async () => { + const { reader, respond, fin } = await subscribed(Version.DRAFT_07, Milli(60_000)); + await respond({ start: new SubscribeStart(0) }); + await respond({ end: new SubscribeEnd(3, 0) }); await fin(); expect(await settlesWithin(reader.recvGroup(), 1000)).toBe(true); expect(await reader.closed).toBeNull(); - expect(reader.final()).toBe(0); -}); - -test("a SUBSCRIBE_END below a group already received aborts the track", async () => { - const { subscriber, reader, respond } = await subscribed(); - await respond({ start: new SubscribeStart(0) }); - const group = groupStream(subscriber, 3); - group.finish(); - await group.handled; - await respond({ end: new SubscribeEnd(2) }); - - const closed = await reader.closed; - expect(closed).toBeInstanceOf(Error); + expect(reader.final()).toBe(3); }); for (const started of [false, true]) { test(`a bare FIN ${started ? "after SUBSCRIBE_START" : "without responses"} aborts the track`, async () => { - const { reader, respond, fin } = await subscribed(); + const { reader, respond, fin } = await subscribed(Version.DRAFT_05); if (started) await respond({ start: new SubscribeStart(0) }); await fin(); expect(await reader.closed).toBeInstanceOf(ProtocolViolation); @@ -231,7 +263,7 @@ for (const started of [false, true]) { } test("a subscribe stream reset preserves the publisher's failure", async () => { - const { reader, reset } = await subscribed(); + const { reader, reset } = await subscribed(Version.DRAFT_05); reset(new StreamError(StreamCode.NotFound)); const closed = await reader.closed; expect(closed).toBeInstanceOf(StreamError); diff --git a/quest/m1/lite-count-settle.md b/quest/m1/lite-count-settle.md index 0ff1f18a715..2cb41dcf52e 100644 --- a/quest/m1/lite-count-settle.md +++ b/quest/m1/lite-count-settle.md @@ -11,17 +11,16 @@ accounting JS already has and Rust track tail adds. ## Plan -- Both publishers already send the count, and both subscribers decode it - (`lite::SubscribeEnd::streams` in Rust, `SubscribeEnd.streams` in JS) and - ignore it. -- JS track tail has landed, and its `Tail` already counts streams. Rust track - tail has landed (`rs/moq-net/src/tail.rs`), so lite-07's completion check - becomes "headers read >= Stream Count" after SUBSCRIBE_END, in place of - every sequence from start to end being covered. The Rust subscriber needs - the same count. -- Tests in both languages: a late stream after SUBSCRIBE_END, a skipped group - that settles without the grace, a reset stream, and a count of zero. Add - the Rust-JS case to the track tail interop test. +- Rust and JS subscribers now settle on the received header count for lite-07. + lite-05 and -06 keep their range and DROP accounting. The grace still covers + counted streams reset before their headers arrive. +- Local Rust and JS regressions cover late streams, missing/reset streams, + skipped sequences, and zero streams. JS also verifies a reset after its header + and groups still being read across the subscription's FIN. +- Remaining: add the count-specific Rust-JS case to track-tail interop. That + harness currently exposes a relay start-floor defect: when a newer group + arrives first, earlier in-flight groups can be lost. Finish the count proof + once that defect is resolved; keep this quest and its PR open until then. ## Related diff --git a/rs/moq-net/src/lite/subscriber.rs b/rs/moq-net/src/lite/subscriber.rs index fdaa29b2012..8a0259fec9d 100644 --- a/rs/moq-net/src/lite/subscriber.rs +++ b/rs/moq-net/src/lite/subscriber.rs @@ -2715,8 +2715,8 @@ struct SubStream { tail: kio::Producer, /// The first group the publisher serves (SUBSCRIBE_START), once declared. served: Option, - /// The track's exclusive end (SUBSCRIBE_END), once declared. - end: Option, + /// The track's exclusive end and stream count (SUBSCRIBE_END), once declared. + end: Option, } impl SubStream { @@ -2725,7 +2725,7 @@ impl SubStream { /// `None` when nothing says which: drafts before SUBSCRIBE_END only have the FIN. /// Without a SUBSCRIBE_START the publisher served no group at all. fn owed(&self, requested_end: Option) -> Option> { - let end = self.end?; + let end = self.end.as_ref()?.group; let end = requested_end.map_or(end, |requested| requested.min(end)); Some(self.served.unwrap_or(end)..end) } @@ -3501,11 +3501,13 @@ enum ServeMode { /// exactly like the old inline await. Establish(Establish), /// The upstream FIN'd, so the track is over, but QUIC does not order streams: keep - /// the subscription routable until every group it owes is accounted for (a stream's - /// header or a SUBSCRIBE_DROP), or the grace gives up on one reset before its header. + /// the subscription routable until the counted headers arrive (lite-07), or every + /// owed group is accounted for (older drafts). The grace bounds a stream reset + /// before its header arrived. Tail { settle: Settle, owed: Option>, + streams: Option, }, } @@ -3552,10 +3554,13 @@ impl ServeLoop { } } } - ServeMode::Tail { settle, owed } => { + ServeMode::Tail { settle, owed, streams } => { let _ = self.fetches.poll(waiter); if settle - .poll(waiter, |tail| owed.clone().is_some_and(|owed| tail.covers(owed))) + .poll(waiter, |tail| match streams { + Some(streams) => tail.streams() >= *streams, + None => owed.clone().is_some_and(|owed| tail.covers(owed)), + }) .is_ready() { return Poll::Ready(ServeEnd::Finished); @@ -3668,7 +3673,7 @@ impl ServeLoop { if let Err(err) = self.serving.finish_at(end.group) { tracing::warn!(track = %serve.name, group = end.group, %err, "invalid subscribe end"); } - active.end = Some(end.group); + active.end = Some(end.clone()); } // SUBSCRIBE_START names the first group this feed serves: // the publisher skipped everything below it (e.g. it could @@ -3730,6 +3735,11 @@ impl ServeLoop { self.mode = ServeMode::Tail { settle: Settle::new(&serve.subscriber.runtime, active.tail.consume(), grace), owed: active.owed(requested_end), + streams: active + .end + .as_ref() + .filter(|_| serve.subscriber.version.has_stream_count()) + .map(|end| end.streams), }; continue; } diff --git a/rs/moq-net/tests/track_tail.rs b/rs/moq-net/tests/track_tail.rs index f9dd8ca201e..a2e21403bf9 100644 --- a/rs/moq-net/tests/track_tail.rs +++ b/rs/moq-net/tests/track_tail.rs @@ -56,12 +56,12 @@ struct Outcome { elapsed: Duration, } -/// Publish a one-group track, end it while its group stream is held back, then deliver or -/// lose that stream, and return what the subscriber read. -async fn round(version: &str, late: Late) -> Outcome { +/// Publish a group below the declared end (or none for an empty track), hold its stream +/// back until the subscription ends, then deliver or lose it. +async fn round(version: &str, late: Late, final_sequence: u64) -> Outcome { let publisher = produce_origin(1); let broadcast = publisher.create_broadcast("bcast").unwrap(); - let track = broadcast.create_track("video", None).unwrap(); + let mut track = broadcast.create_track("video", None).unwrap(); broadcast.announce(Default::default()).unwrap(); let subscriber = produce_origin(2); @@ -112,19 +112,23 @@ async fn round(version: &str, late: Late) -> Outcome { .unwrap(); pair.server_transport.hold_unis(); - let mut group = track.append_group().unwrap(); - group.write_frame(Timestamp::ZERO, PAYLOAD).unwrap(); - group.finish().unwrap(); - track.finish().unwrap(); + if final_sequence > 0 { + let mut group = track.append_group().unwrap(); + group.write_frame(Timestamp::ZERO, PAYLOAD).unwrap(); + group.finish().unwrap(); + } + track.finish_at(final_sequence).unwrap(); drop(track); - // Paused time only advances once every task is idle, so this runs the publisher to its - // end of the subscription and the subscriber through reading it. - tokio::time::sleep(GRACE / 10).await; - assert!( - !reader.is_finished(), - "{version}: the track ended before its group arrived" - ); + if final_sequence > 0 { + // Paused time advances only when every task is idle, after the publisher and + // subscriber have processed the subscription's end. + tokio::time::sleep(GRACE / 10).await; + assert!( + !reader.is_finished(), + "{version}: the track ended before its group arrived" + ); + } let released = tokio::time::Instant::now(); match late { @@ -150,7 +154,7 @@ async fn round(version: &str, late: Late) -> Outcome { async fn a_group_after_the_end_is_delivered() { tokio::time::pause(); for version in VERSIONS { - let outcome = round(version, Late::Delivered).await; + let outcome = round(version, Late::Delivered, 1).await; assert!( outcome.err.is_none() && outcome.frames == [PAYLOAD], "{version}: got {} frame(s), err={:?}", @@ -176,7 +180,7 @@ async fn a_group_after_the_end_is_delivered() { async fn a_lost_group_ends_the_track_after_the_grace() { tokio::time::pause(); for version in VERSIONS { - let outcome = round(version, Late::Lost).await; + let outcome = round(version, Late::Lost, 1).await; assert!( outcome.err.is_none() && outcome.frames.is_empty(), "{version}: got {} frame(s), err={:?}", @@ -190,3 +194,23 @@ async fn a_lost_group_ends_the_track_after_the_grace() { ); } } + +/// Skipped sequences have no stream and must not hold lite-07's counted tail open. +#[tokio::test] +async fn lite07_skipped_groups_end_without_the_grace() { + tokio::time::pause(); + let outcome = round("moq-lite-07-wip", Late::Delivered, 3).await; + assert!(outcome.err.is_none()); + assert_eq!(outcome.frames, [PAYLOAD]); + assert!(outcome.elapsed < GRACE / 10, "ended after {:?}", outcome.elapsed); +} + +/// A zero stream count leaves no tail to wait for. +#[tokio::test] +async fn lite07_zero_streams_end_without_the_grace() { + tokio::time::pause(); + let outcome = round("moq-lite-07-wip", Late::Delivered, 0).await; + assert!(outcome.err.is_none()); + assert!(outcome.frames.is_empty()); + assert!(outcome.elapsed < GRACE / 10, "ended after {:?}", outcome.elapsed); +}