diff --git a/doc/concept/moq-lite.md b/doc/concept/moq-lite.md index ee3f10472ee..de0b9e42508 100644 --- a/doc/concept/moq-lite.md +++ b/doc/concept/moq-lite.md @@ -288,8 +288,9 @@ moq-lite 07 receiver. Since moq-lite 05 and 06 send times look like media time, a receiver drops every timestamp it gets over them and the track arrives untimed. Timestamps received over moq-transport, whose peers may stamp arrival times, -are not checked and are only passed on over moq-transport: a relay serves that -track to moq-lite untimed. +are not checked. Rust relays preserve them across moq-lite 07 when both peers +negotiate the [ietf extension](/draft/moq-ietf), which carries their origin and +keeps them unchecked. A moq-lite peer without it receives that track untimed. A publisher may compress a track, which pays off for repetitive data such as JSON and not for media. Each group's frames become one DEFLATE stream diff --git a/drafts/draft-lcurley-moq-ietf.md b/drafts/draft-lcurley-moq-ietf.md index ac7b0680e59..9ef9dd9068a 100644 --- a/drafts/draft-lcurley-moq-ietf.md +++ b/drafts/draft-lcurley-moq-ietf.md @@ -143,6 +143,11 @@ The Version also gives each property its meaning: drafts 18 and 19 serialize a b Other properties are opaque to this extension. ## Timestamps {#timestamps} +A relay MUST preserve a relayed track's Timescale and timestamps across a negotiated hop, except in a listing ({{track}}). +A nonzero Version identifies timestamps from MoQ Transport, which may be arrival times and need not rise with Group Sequence. +The receiver MUST NOT apply moq-lite's group-start rewind check to such a track; a Version of zero retains that check. +Without this extension, a relay MUST serve a MoQ Transport track untimed as {{moql}} requires. + On a timed track, the FRAME's Timestamp, or the datagram's for a datagram, is the object's time, and nothing else carries it. The Property Block MUST NOT contain a mutable Timestamp (0x10) there; a receiver MUST treat one as malformed. A relay forwarding the object to {{moqt}} restates the time as the mutable Timestamp, unless a Timestamp inside the Immutable Properties already says it. @@ -194,6 +199,7 @@ The receiver parses the keys inside the Immutable Properties from its value, whi It SHOULD read them from the value when needed rather than hold them beside it, as a list of IDs can cost several times the value. A TRACKS_ADD ({{moql}} Tracks Stream) embeds TRACK_INFO without TRACK_PROPERTIES, so when the extension is negotiated a subscriber MUST NOT skip the Track Stream for a track it learned from a listing. +Lacking the Version, a TRACKS_ADD cannot mark a {{moqt}} track's timestamps as unchecked ({{timestamps}}), so it lists that track untimed even when the extension is negotiated. # Datagrams {#datagrams} @@ -258,6 +264,7 @@ A high, distinctive value is used to avoid collisions with the low values {{moql ## Since draft-lcurley-moq-ietf-00 (in progress) +- Preserve relayed track timestamps and their unchecked MoQ Transport provenance across negotiated hops. - Initial version. # Acknowledgments diff --git a/drafts/draft-lcurley-moq-lite.md b/drafts/draft-lcurley-moq-lite.md index 08890fd1b7e..1529456c7a6 100644 --- a/drafts/draft-lcurley-moq-lite.md +++ b/drafts/draft-lcurley-moq-lite.md @@ -184,11 +184,11 @@ Each frame of a timed Track carries a presentation timestamp expressed in the Tr A Track without a `Timescale` is untimed: its frames and datagrams carry no timestamp. A timestamp MUST be less than 2^61: lite-05 and lite-06 zigzag a Group's first [Timestamp Delta](#frame) into a QUIC varint, which carries no more, and the bound lets a relay forward any timestamp to any version. A receiver treats a larger timestamp as a malformed Track, even one it discards. -A publisher MUST NOT start a timed Group's frame 0 earlier than that of any lower-sequence Group it publishes. +Unless a negotiated extension specifies otherwise, a publisher MUST NOT start a timed Group's frame 0 earlier than that of any lower-sequence Group it publishes. lite-05 and lite-06 cannot encode an untimed Track, so a publisher on those versions declares a `Timescale` of 1000 for it and stamps every frame and datagram with its send time in milliseconds. A receiver cannot tell such send times from presentation timestamps, and they need not rise with sequence, so it discards every timestamp received on those versions and treats the Track as untimed. On later versions a receiver MAY check a received Group against this rule, refusing a violating Group's stream (STOP_SENDING) with TIMESTAMP_MISMATCH. -A relay that cannot vouch for a Track's timestamps, such as ones received over another protocol, MUST serve the Track untimed. +A relay that cannot vouch for a Track's timestamps, such as ones received over another protocol, MUST serve the Track untimed unless a negotiated extension identifies their unchecked provenance. ## Positions {#positions} A Position is a (Group Sequence, Frame Index) pair identifying one frame within a Track. @@ -1768,6 +1768,8 @@ The `Message Length` describes the payload size on the wire. ## moq-lite-07 +- Allow a negotiated extension to preserve timestamps with explicitly unchecked provenance. + - Added Padding Stream (0x2), read and discarded incrementally by either endpoint. - Named 128 the midpoint priority for an unset Subscriber or Publisher Priority. @@ -1778,7 +1780,7 @@ The `Message Length` describes the payload size on the wire. - Replaced ANNOUNCE_OK Active Count with an ordered, empty SYNCED event (0x4), marking the initial set without retaining routes that end before delivery. -- A publisher must not start a timed Group's frame 0 earlier than a lower-sequence Group's, and a receiver may refuse a Group that does. A receiver discards the timestamps it receives over lite-05/06, which may be send times, and a relay serves a Track whose timestamps it cannot vouch for untimed. +- A publisher must not start a timed Group's frame 0 earlier than a lower-sequence Group's, and a receiver may refuse a Group that does. A receiver discards the timestamps it receives over lite-05/06, which may be send times, and a relay serves a Track whose timestamps it cannot vouch for untimed unless a negotiated extension identifies their unchecked provenance. - Bounded every timestamp below 2^61, since a lite-05/06 Group's first Timestamp Delta carries no more, so a relay never holds one it cannot forward. A larger one is a malformed Track. diff --git a/js/net/src/ietf/subscriber.ts b/js/net/src/ietf/subscriber.ts index b8fc38f621d..8a4ca40b462 100644 --- a/js/net/src/ietf/subscriber.ts +++ b/js/net/src/ietf/subscriber.ts @@ -1227,7 +1227,7 @@ export class Subscriber { priority: fromWire(ok.properties.priority ?? ABSENT_WIRE), maxAge: maxCacheDuration === undefined ? undefined : Milli(Number(maxCacheDuration)), // A moq-transport peer may stamp arrival times, and nothing on the wire says it didn't. - unchecked: ok.properties.timescale != null, + ietf: version, }); const producer = staged.producer; const subscription: Subscription = { track: producer, tail: new Tail() }; diff --git a/js/net/src/lite/publisher.test.ts b/js/net/src/lite/publisher.test.ts index f1d8b554a66..396eb9e461b 100644 --- a/js/net/src/lite/publisher.test.ts +++ b/js/net/src/lite/publisher.test.ts @@ -6,6 +6,7 @@ import { Producer as BroadcastProducer } from "../broadcast.ts"; import { fromTransport, SessionCode, StreamCode, StreamError, toStreamCode } from "../error.ts"; import { Producer as GroupProducer } from "../group.ts"; import { randomHop } from "../hop.ts"; +import { Version as IetfVersion } from "../ietf/version.ts"; import { hooks } from "../internal.ts"; import { createMockTransportPair, textFrame } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; @@ -555,7 +556,7 @@ test("lite draft-06: an unchecked track's fetched frames carry their send time", const publisher = new Publisher(pair.server, Version.DRAFT_06, randomHop(), origin.consume()); const broadcast = publish(origin, Path.from("test")); - const track = broadcast.createTrack("video", { timescale: Timescale.MICRO, unchecked: true }); + const track = broadcast.createTrack("video", { timescale: Timescale.MICRO, ietf: IetfVersion.DRAFT_22 }); const group = new GroupProducer(0); group.writeFrame(textFrame("hello", Timestamp.fromMillis(36_000_000))); diff --git a/js/net/src/lite/track_stream.test.ts b/js/net/src/lite/track_stream.test.ts index 77579216486..903d7f7fc13 100644 --- a/js/net/src/lite/track_stream.test.ts +++ b/js/net/src/lite/track_stream.test.ts @@ -1,8 +1,10 @@ // An open Track Stream is interest in the track: the subscriber keeps it open until its // SUBSCRIBE is answered, and the publisher keeps the track wanted until the subscriber closes it. + import { describe, expect, test } from "bun:test"; import { Milli } from "@moq/clock"; import { randomHop } from "../hop.ts"; +import { Version as IetfVersion } from "../ietf/version.ts"; import { createMockTransportPair } from "../mock.ts"; import { Producer as OriginProducer } from "../origin.ts"; import * as Path from "../path.ts"; @@ -396,7 +398,7 @@ describe.each([Version.DRAFT_05, Version.DRAFT_06, Version.DRAFT_07])("%s", (ver for (;;) { const request = await wireOf(broadcast).requested(); if (!request) return; - request.accept({ timescale: Timescale.MICRO, unchecked: true }); + request.accept({ timescale: Timescale.MICRO, ietf: IetfVersion.DRAFT_22 }); } })(); diff --git a/js/net/src/track.test.ts b/js/net/src/track.test.ts index 2f2d226625e..97eeef81e40 100644 --- a/js/net/src/track.test.ts +++ b/js/net/src/track.test.ts @@ -3,6 +3,7 @@ import { Manual, Milli, Nano } from "@moq/clock"; import { Producer as BroadcastProducer } from "./broadcast.ts"; import { ProtocolViolation, StreamCode, StreamError, TimestampMismatch, TooFarBehind } from "./error.ts"; import { Producer as GroupProducer, MAX_GROUP_FRAMES } from "./group.ts"; +import { Version as IetfVersion } from "./ietf/version.ts"; import { hooks } from "./internal.ts"; import { textFrame } from "./mock.ts"; import { Timescale, Timestamp } from "./time.ts"; @@ -2977,6 +2978,22 @@ test("an endless track with gaps remembers only those within the budget", () => } }); +test("IETF provenance survives normalization on timed and untimed tracks", () => { + for (const timescale of [undefined, Timescale.MILLI]) { + const info = infoDefaults({ ietf: IetfVersion.DRAFT_22, timescale }); + expect(info.ietf).toBe(IetfVersion.DRAFT_22); + expect(info.timescale).toBe(timescale); + } + const producer = new TrackProducer("ietf").accept({ timescale: Timescale.MILLI, ietf: IetfVersion.DRAFT_22 }); + try { + stampGroup(producer, 0, 1000); + stampGroup(producer, 1, 500); + producer.writeFrame({ payload: enc.encode("arrival"), timestamp: Timestamp.fromMillis(250) }); + } finally { + producer.close(); + } +}); + test("an appended rewind is refused and an equal start is not", () => { const producer = new TrackProducer("test").accept({ timescale: Timescale.MILLI }); producer.writeFrame({ payload: enc.encode("a"), timestamp: Timestamp.fromMillis(1000) }); diff --git a/js/net/src/track.ts b/js/net/src/track.ts index 4c6e06c9869..d5428b4ac15 100644 --- a/js/net/src/track.ts +++ b/js/net/src/track.ts @@ -19,6 +19,7 @@ import { } from "./error.ts"; import { Compression } from "./flate/group.ts"; import { type Frame, type Consumer as GroupConsumer, Producer as GroupProducer } from "./group.ts"; +import type { IetfVersion } from "./ietf/version.ts"; import { type DropRun, dropAt, @@ -130,15 +131,9 @@ export interface Info { * snapshot or keyframe. `0` asks on any join after a delta. An integer in `0..2^32`. */ refresh?: number; - /** - * Whether this timed track's timestamps arrived over moq-transport, whose peers may stamp - * arrival times, so no group is checked for a rewind. A moq-lite hop serves the track - * untimed, since its receivers check every timestamp. Mirrors the Rust - * `track::Info::unchecked`. - * - * @internal - */ - unchecked?: boolean; + /** The original moq-transport publisher's draft, whose timestamps may rewind. @internal */ + // private: passthrough + ietf?: IetfVersion; } // Normalize a duration for the wire, which carries it as an unsigned varint. @@ -166,9 +161,9 @@ function priorityValue(value: number): number { } // private: passthrough -/** The timescale a moq-lite hop serves a track at: none for an {@link Info.unchecked} one. @internal */ +/** The timescale a moq-lite hop serves a track at: none for an {@link Info.ietf} one. @internal */ export function liteTimescale(info: Info): Timescale | undefined { - return info.unchecked ? undefined : (info.timescale ?? undefined); + return info.ietf === undefined ? (info.timescale ?? undefined) : undefined; } /** Fill in any unset {@link Info} fields with their defaults. */ @@ -176,15 +171,13 @@ export function infoDefaults(info: Partial = {}): Info { if (info.compress !== undefined && typeof info.compress !== "boolean") throw new TypeError("compress must be a boolean"); const timescale = info.timescale == null ? undefined : Timescale(info.timescale); - // Only a timed track has timestamps to doubt. - if (info.unchecked && timescale === undefined) throw new RangeError("an untimed track can't be unchecked"); return { compress: info.compress ?? false, timescale, maxAge: info.maxAge === undefined ? undefined : wireMillis("maxAge", info.maxAge), priority: priorityValue(info.priority ?? DEFAULT_PRIORITY), refresh: info.refresh == null ? undefined : refreshRatio(info.refresh), - ...(info.unchecked ? { unchecked: true } : {}), + ...(info.ietf === undefined ? {} : { ietf: info.ietf }), }; } @@ -1490,9 +1483,9 @@ export class Producer { this.#publish(group, start, late); } - // Whether this track's timestamps may rewind (see {@link Info.unchecked}). + // Whether this track's timestamps may rewind (see {@link Info.ietf}). #unchecked(): boolean { - return this.#state.info.peek()?.unchecked === true; + return this.#state.info.peek()?.ietf !== undefined; } // Fan a datagram out to every live subscriber, dropping the oldest once the ring is full. diff --git a/quest/m1/ietf-extension/README.md b/quest/m1/ietf-extension/README.md index fa0d603ae5f..c52db0029f8 100644 --- a/quest/m1/ietf-extension/README.md +++ b/quest/m1/ietf-extension/README.md @@ -1,4 +1,4 @@ -# moq-transport passthrough +# [M] moq-transport passthrough ## Goal @@ -82,10 +82,6 @@ Work owned by this README, after the children land: - FETCH cases are skipped upstream. Section 8 stays out while PUBLISH is declined. -## Required - -- [Timed tracks cross lite-07](/quest/m1/ietf-extension/serve-timed.md) - moq-transport timestamps reach `ietf` extension peers over lite-07, with `unchecked` derived from `Info::ietf()` - ## Related - [#4926](https://github.com/moq-dev/moq/issues/4926) - the declined request for an object-level API; this carries the fields for relays only diff --git a/quest/m1/ietf-extension/serve-timed.md b/quest/m1/ietf-extension/serve-timed.md deleted file mode 100644 index 28e6ea043c8..00000000000 --- a/quest/m1/ietf-extension/serve-timed.md +++ /dev/null @@ -1,26 +0,0 @@ -# [S] Timed moq-transport tracks cross a lite-07 mesh - -## Goal - -A timed track ingested over moq-transport keeps its timestamps across a -lite-07 hop to a peer that negotiated the `ietf` extension, so a relay mesh -carries them end to end. A peer without the extension still gets the track -untimed. - -## Plan - -Decided 2026-10-21 with the maintainer: m1, under moq-transport passthrough. - -[moq-dev/moq#6150](https://github.com/moq-dev/moq/pull/6150) marks a timed -track received over moq-transport with the crate-private -`track::Info::unchecked` and serves it untimed to moq-lite, so its -timestamps stop at the first lite hop. -[moq-dev/moq#6107](https://github.com/moq-dev/moq/pull/6107) gives every -moq-transport track `track::Info::ietf()`, its ingress version. Derive -`unchecked` from `ietf()` instead of keeping a second flag, and let the lite -publisher serve such a track timed to an `ietf` extension peer, which knows -the timestamps came from moq-transport and are unchecked. - -Mirror it in js/net, where `Info.unchecked` is `@internal`. If the timed -encoding on that hop needs a wire change, update -`drafts/draft-lcurley-moq-lite.md` (or the extension's draft) in the same PR. diff --git a/rs/moq-net/src/ietf/subscriber.rs b/rs/moq-net/src/ietf/subscriber.rs index 8eb9a27b13f..b6a26468327 100644 --- a/rs/moq-net/src/ietf/subscriber.rs +++ b/rs/moq-net/src/ietf/subscriber.rs @@ -520,8 +520,6 @@ fn accepted_info(properties: &ietf::Properties, version: Version) -> track::Info version, values: properties.unknown.clone(), })); - // A moq-transport peer may stamp arrival times, and nothing on the wire says it didn't. - info.unchecked = info.timescale.is_some(); info } @@ -12857,7 +12855,7 @@ mod stitch_tests { } /// Relay `script` through a track the ingress times in `timescale`, which the cache - /// normalizes to microseconds like [`accepted_info`], via one relay or two joined by a + /// retains on the publisher's grid like [`accepted_info`], via one relay or two joined by a /// moq-lite `hop`. async fn relay_passthrough_via( script: Vec, @@ -13045,15 +13043,15 @@ mod stitch_tests { /// A relayed Timestamp leaves in the units the publisher declared: the cache holds the /// track on the publisher's grid, so a copy of the same instance over another route - /// matches it. A moq-lite hop serves the unchecked track untimed, so the Timestamp - /// stops there and the other property survives. + /// matches it, including across a moq-lite hop that negotiated the `ietf` extension. #[moq_clock::test] async fn passthrough_keeps_the_publishers_timestamp_units() { let version = Version::Draft22; // Timestamp 42 (ms), then an unknown property that must survive beside it. let properties = [0x10, 42, 0x70, 7]; let script = object_stream(version, true, &[(0, &properties, None)]); - for (hop, timestamp) in [(false, Timestamp::new(42, Timescale::MILLI).ok()), (true, None)] { + for hop in [false, true] { + let timestamp = Timestamp::new(42, Timescale::MILLI).ok(); let (bytes, log) = relay_passthrough_via(script.clone(), version, version, None, Some(Timescale::MILLI), hop).await; assert!(log.resets().is_empty(), "hop: {hop}"); diff --git a/rs/moq-net/src/lite/ietf.rs b/rs/moq-net/src/lite/ietf.rs index 83b88b2843a..750afa06119 100644 --- a/rs/moq-net/src/lite/ietf.rs +++ b/rs/moq-net/src/lite/ietf.rs @@ -1201,7 +1201,7 @@ mod hop_tests { // The upstream group is still open. let frame = next(&mut received).await.unwrap(); assert_eq!(&frame.payload[..], b"a"); - assert_eq!(frame.timestamp, Some(ts(0))); + assert_eq!(frame.timestamp, negotiated.then(|| ts(0))); match negotiated { true => { assert_eq!(fields(&frame), Some((first.id, first.properties.clone())), "{offer:?}"); @@ -1240,7 +1240,12 @@ mod hop_tests { let info = moq_clock::timeout(TIMEOUT, copy.query()).await.unwrap().unwrap(); assert_eq!( (info.max_age, info.timescale, info.priority, info.refresh), - (Some(Duration::from_secs(7)), Some(Timescale::MILLI), 9, Some(0)), + ( + Some(Duration::from_secs(7)), + negotiated.then_some(Timescale::MILLI), + 9, + Some(0) + ), "{offer:?}" ); match negotiated { @@ -1469,6 +1474,82 @@ mod hop_tests { drop(sessions); } + /// Provenance keeps IETF times unchecked after a hop, while a peer declining the + /// extension gets no timeline. SUBSCRIBE, FETCH, and datagrams must agree with TRACK_INFO, + /// while a listing stays untimed. + #[moq_clock::test] + async fn ietf_timestamps_keep_their_provenance_across_a_hop() { + for offer in [[true, true], [true, false], [false, true]] { + let negotiated = offer == [true, true]; + let upstream = crate::origin::Config::new(crate::Hop::new(2).unwrap()).produce(); + let broadcast = upstream.publish("room", crate::origin::Route::default()).unwrap(); + let info = track::Info { + properties: draft22(Vec::new()), + ..track::Info::timed() + }; + let mut track = broadcast.create_track("video", info).unwrap(); + let mut head = track.create_group(0u64.into()).unwrap(); + head.write_frame(ts(100), b"frame".to_vec()).unwrap(); + head.finish().unwrap(); + let (downstream, sessions) = test_hop(&upstream, "room", offer).await; + let broadcast = downstream.consume().request_broadcast("room", None).await.unwrap(); + // A listing cannot carry the provenance, so it lists the track untimed even when + // negotiated: a relay re-listing it would otherwise advertise it as native lite. + let mut tracks = broadcast.tracks().unwrap(); + let listed = loop { + match moq_clock::timeout(TIMEOUT, tracks.next()).await.unwrap().unwrap() { + Some(crate::broadcast::TracksEvent::Added { name, info }) if &*name == "video" => break info, + Some(_) => continue, + None => panic!("listing ended before the track"), + } + }; + assert_eq!((listed.timescale, listed.ietf()), (None, None), "{offer:?}"); + let copy = broadcast.track("video").unwrap(); + let info = moq_clock::timeout(TIMEOUT, copy.query()).await.unwrap().unwrap(); + assert_eq!(info.timescale, negotiated.then_some(Timescale::MILLI), "{offer:?}"); + assert_eq!( + info.ietf(), + negotiated.then_some(crate::ietf::Version::Draft22), + "{offer:?}" + ); + // Before any subscription, this must fetch the head over the hop, rather + // than reading an already subscribed copy's cache. + let mut fetched = moq_clock::timeout(TIMEOUT, copy.fetch_group(0, None)) + .await + .unwrap() + .unwrap(); + assert_eq!(next(&mut fetched).await.unwrap().timestamp, negotiated.then(|| ts(100))); + assert!(next(&mut fetched).await.is_none()); + let mut subscriber = moq_clock::timeout( + TIMEOUT, + copy.subscribe(track::Subscription::default().with_start(track::Position::group(1))), + ) + .await + .unwrap() + .unwrap(); + // A peer may have stamped arrival times. The second group's rewind must not + // acquire native moq-lite checks merely because it crossed a relay hop. + let mut group = track.create_group(1u64.into()).unwrap(); + group.write_frame(ts(50), b"frame".to_vec()).unwrap(); + group.finish().unwrap(); + let mut received = recv(&mut subscriber).await; + assert_eq!(received.sequence, 1); + let frame = next(&mut received).await.unwrap(); + assert_eq!(frame.timestamp, negotiated.then(|| ts(50)), "{offer:?}"); + assert_eq!(&frame.payload[..], b"frame"); + assert!(next(&mut received).await.is_none()); + track.insert_datagram(2, ts(25), b"datagram".to_vec()).unwrap(); + let datagram = moq_clock::timeout(TIMEOUT, kio::wait(|waiter| subscriber.poll_recv_wire_datagram(waiter))) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(datagram.timestamp, negotiated.then(|| ts(25)), "{offer:?}"); + assert_eq!(&datagram.payload[..], b"datagram"); + drop(sessions); + } + } + /// A compressed track crosses a negotiated hop as stored: each frame's Inflated Size /// follows its Object Record, and a datagram's follows its Datagram State. #[moq_clock::test] diff --git a/rs/moq-net/src/lite/publisher.rs b/rs/moq-net/src/lite/publisher.rs index 0b4a3aa7909..9c42c5f40ab 100644 --- a/rs/moq-net/src/lite/publisher.rs +++ b/rs/moq-net/src/lite/publisher.rs @@ -1959,7 +1959,7 @@ impl Request for TrackInfoServe { crate::ietf::flate::egress(&info, shared.version.has_compression())?; // TRACK_INFO only flows on Lite05+ (the encode errors otherwise). - writer.buffer(&lite::TrackInfo::serve(&info, shared.version))?; + writer.buffer(&lite::TrackInfo::serve(&info, shared.version, ietf))?; if ietf { writer.buffer(&lite::ietf::TrackProperties(info.properties.clone()))?; } @@ -2016,7 +2016,9 @@ impl Request for TracksServe { let event = match ready!(self.tracks.poll_next(waiter))? { Some(crate::broadcast::TracksEvent::Added { name, info }) => lite::TracksEvent::Added { name: name.to_string().into(), - info: lite::TrackInfo::serve(&info, shared.version), + // Untimed even to an extension peer: TRACKS_ADD has no TRACK_PROPERTIES to + // carry the unchecked provenance, so a relay would re-list it as native lite. + info: lite::TrackInfo::serve(&info, shared.version, false), }, Some(crate::broadcast::TracksEvent::Removed { name }) => lite::TracksEvent::Removed { name: name.to_string().into(), @@ -2130,7 +2132,7 @@ impl Request for SubscribeServe { // `None` omits the per-frame timestamp prefix: an untimed track on // Lite07+, or any track before Lite05 (whose frames still carry // timestamps in the model, just not on this wire). - let stamp = Stamp::new(shared.version, track.info().lite_timescale()); + let stamp = Stamp::new(shared.version, track.info().lite_timescale(ietf)); // Lite05+ accepts implicitly: no SUBSCRIBE_OK, the immutable properties // live in TRACK_INFO, and the resolved range arrives as @@ -2306,7 +2308,7 @@ impl Request for FetchServe { group.end_at(msg.end_frame.map_or(Bound::Unbounded, Bound::Included)); // Matches the timescale TRACK_INFO declared for the track. - let stamp = Stamp::new(shared.version, group.lite_timescale()); + let stamp = Stamp::new(shared.version, group.lite_timescale(ietf)); // And its compression, which a lite-07 peer receives as stored. if shared.version.has_compression() { group = group.compressed(); @@ -4833,15 +4835,20 @@ mod serve_group_test { let sent = Timestamp::from(runtime.now()).value(); let checked = track::Info::timed(); let unchecked = track::Info { - unchecked: true, + properties: Some(std::sync::Arc::new(crate::model::object::TrackProperties { + version: crate::ietf::Version::Draft22, + values: Vec::new(), + })), ..track::Info::timed() }; - let stamp = Stamp::new(Version::Lite06, checked.lite_timescale()).unwrap(); + let stamp = Stamp::new(Version::Lite06, checked.lite_timescale(false)).unwrap(); assert_eq!(stamp.wire(Some(stamped), &runtime).unwrap(), stamped.value()); - let stamp = Stamp::new(Version::Lite06, unchecked.lite_timescale()).unwrap(); + let stamp = Stamp::new(Version::Lite06, unchecked.lite_timescale(false)).unwrap(); assert_eq!(stamp.wire(Some(stamped), &runtime).unwrap(), sent); - assert!(Stamp::new(Version::Lite07, unchecked.lite_timescale()).is_none()); + assert!(Stamp::new(Version::Lite07, unchecked.lite_timescale(false)).is_none()); + let stamp = Stamp::new(Version::Lite07, unchecked.lite_timescale(true)).unwrap(); + assert_eq!(stamp.wire(Some(stamped), &runtime).unwrap(), stamped.value()); } #[moq_clock::test] diff --git a/rs/moq-net/src/lite/track.rs b/rs/moq-net/src/lite/track.rs index eeaa1f8a112..dfb33c5af96 100644 --- a/rs/moq-net/src/lite/track.rs +++ b/rs/moq-net/src/lite/track.rs @@ -78,13 +78,13 @@ pub struct TrackInfo { impl TrackInfo { /// The TRACK_INFO a publisher serves for `info` on `version`. - pub(crate) fn serve(info: &crate::track::Info, version: Version) -> Self { + pub(crate) fn serve(info: &crate::track::Info, version: Version, ietf: bool) -> Self { Self { refresh: info.refresh, priority: info.priority, max_age: info.max_age, - // Unchecked timestamps never enter moq-lite, whose receivers check them. - timescale: info.lite_timescale(), + // Only an extension peer knows the original publisher may stamp arrival times. + timescale: info.lite_timescale(ietf), // An older peer is served the frames inflated. compress: info.compress && version.has_compression(), } @@ -317,7 +317,7 @@ mod test { #[test] fn track_info_defaults_match_cross_language_wire_bytes() { // The default track is untimed, which lite-05 declares as milliseconds. - let info = TrackInfo::serve(&crate::track::Info::default(), Version::Lite05); + let info = TrackInfo::serve(&crate::track::Info::default(), Version::Lite05, false); let mut buf = Vec::new(); info.encode(&mut Encoder::new(&mut buf, Version::Lite05.into()), Version::Lite05) .unwrap(); diff --git a/rs/moq-net/src/model/group.rs b/rs/moq-net/src/model/group.rs index 73d20ffc7b5..79c42c33e22 100644 --- a/rs/moq-net/src/model/group.rs +++ b/rs/moq-net/src/model/group.rs @@ -2794,8 +2794,8 @@ impl Consumer { /// [`track::Info::lite_timescale`]). #[cfg_attr(not(feature = "async"), allow(dead_code))] // private: passthrough - pub(crate) fn lite_timescale(&self) -> Option { - self.track.lite_timescale() + pub(crate) fn lite_timescale(&self, ietf: bool) -> Option { + self.track.lite_timescale(ietf) } /// The draft the parent track's moq-transport publisher spoke, which serialized every diff --git a/rs/moq-net/src/model/resume.rs b/rs/moq-net/src/model/resume.rs index 841b6099031..7752b12be9a 100644 --- a/rs/moq-net/src/model/resume.rs +++ b/rs/moq-net/src/model/resume.rs @@ -4313,7 +4313,10 @@ mod test { Arc::new(broadcast::Info::default()), "video", crate::track::Info { - unchecked: true, + properties: Some(Arc::new(crate::model::object::TrackProperties { + version: crate::ietf::Version::Draft22, + values: Vec::new(), + })), ..crate::track::Info::timed() }, ); diff --git a/rs/moq-net/src/model/track.rs b/rs/moq-net/src/model/track.rs index e2c528ee326..ba517feeb57 100644 --- a/rs/moq-net/src/model/track.rs +++ b/rs/moq-net/src/model/track.rs @@ -169,14 +169,6 @@ pub struct Info { #[cfg_attr(not(feature = "async"), allow(dead_code))] // private: passthrough pub(crate) properties: Option>, - /// Whether this timed track's timestamps arrived over moq-transport, whose peers may - /// stamp arrival times, so no group is checked for a rewind. - /// - /// They stay on moq-transport: a moq-lite hop serves the track untimed, since its - /// receivers check every timestamp. Never set on an untimed track. - #[cfg_attr(not(feature = "async"), allow(dead_code))] - // private: passthrough - pub(crate) unchecked: bool, } impl Default for Info { @@ -188,7 +180,6 @@ impl Default for Info { compress: false, refresh: None, properties: None, - unchecked: false, } } } @@ -228,15 +219,16 @@ impl Info { self } - /// The timescale a moq-lite hop serves this track at: none for an - /// [unchecked](Self::unchecked) one. + /// The timescale served on a moq-lite hop, preserving moq-transport timestamps only + /// when the hop negotiated the `ietf` extension. #[cfg_attr(not(feature = "async"), allow(dead_code))] // private: passthrough - pub(crate) fn lite_timescale(&self) -> Option { - self.timescale.filter(|_| !self.unchecked) + pub(crate) fn lite_timescale(&self, ietf: bool) -> Option { + self.timescale.filter(|_| self.ietf().is_none() || ietf) } - /// The draft the track's moq-transport publisher spoke, `None` for a moq-lite one. + /// The original moq-transport publisher's draft, whose timestamps may rewind, or + /// `None` for a native moq-lite track. // private: passthrough pub(crate) fn ietf(&self) -> Option { self.properties.as_ref().map(|properties| properties.version) @@ -848,9 +840,9 @@ impl TrackState { /// so the watermark caps it instead. /// /// Usually one group lookup each way; only unstarted or partial neighbors are skipped. - /// `None` on an [unchecked](Info::unchecked) track, whose timestamps may rewind. + /// `None` on a track with moq-transport provenance, whose timestamps may rewind. pub(super) fn neighbors(&self, sequence: u64) -> Option { - if self.info.as_ref().is_some_and(|info| info.unchecked) { + if self.info.as_ref().is_some_and(|info| info.ietf().is_some()) { return None; } let bounds = self.bounds.lock().unwrap_or_else(PoisonError::into_inner);