b118eb0c0d
A relay with a door on the room's namespace refuses an object of any track whose receipts never come, the catalog among them. The gateway now signs the catalog as it signs the media, on catalog/receipts, and a WHEP session reads the catalog only once its receipt verifies under a trusted first hop. One b=AS line per section: some WebRTC stacks read one bandwidth line only. The end-to-end tests (WHIP to a relay subscriber, WHEP playback, refusals, trickle ICE and an ICE restart) and the rooms corpus test pass on the build host. Co-Authored-By: Claude Opus 5.5 <[email protected]>
459 lines
20 KiB
Rust
459 lines
20 KiB
Rust
//! Frames into MoQ groups and objects, and each group's receipts signed as the first hop
|
||
//! (`extensions/live-ingest.md` §5 to §7).
|
||
//!
|
||
//! - **Objects.** Each frame is one object, one `WAI1` envelope ([`crate::frame`]).
|
||
//! - **Groups.** Video: a group begins at each keyframe, and holds it and the frames up
|
||
//! to the next. Audio: a group holds the frames whose mapped time falls in one window
|
||
//! of `audio_group_ms`, the windows laid end to end from the track's first frame. Group
|
||
//! IDs ascend by one; a track's first group's ID is its first frame's room-clock time in
|
||
//! microseconds, so a publisher that reconnects continues above the IDs it used before.
|
||
//! - **Receipts.** When a group ends, the gateway signs its group receipt over every
|
||
//! object of it and publishes the group receipt and the object receipts as group *g* of
|
||
//! the track's receipts track (moq-streaming-format §4.1): after the group's last object,
|
||
//! since the root covers every object. Each receipt chains to the track's previous group.
|
||
//! - **Time.** [`Timeline`]: the RTP timestamp to room clock mapping.
|
||
|
||
use crate::frame::{self, Codec, FrameInfo, GrantRef, Header, Live};
|
||
use ed25519_dalek::SigningKey;
|
||
use joule_moq::carriage::ReceiptBundle;
|
||
use joule_moq::receipt::{ObjectReceipt, ReceiptBuilder};
|
||
use joule_moq::{GroupId, ObjectId, TrackAlias};
|
||
use std::sync::Arc;
|
||
|
||
/// The RTP timestamp to room clock mapping of one track (`extensions/live-ingest.md` §7).
|
||
///
|
||
/// The first frame anchors it: its room-clock time is its arrival at the gateway, `A`,
|
||
/// and its extended RTP timestamp is `R0`. A later frame with timestamp `R` maps to
|
||
/// `A + ⌊(R − R0) · 10^6 / rate⌋` µs. A frame whose mapped time lies more than
|
||
/// `max_skew` from its own arrival, or whose timestamp is below `R0`, re-anchors the
|
||
/// mapping at itself. Mapped times never decrease on a track. So every frame's time lies
|
||
/// within `max_skew` of its arrival at the gateway.
|
||
#[derive(Clone, Debug)]
|
||
pub struct Timeline {
|
||
pub rate: u64,
|
||
pub max_skew_us: u64,
|
||
anchor: Option<(u64, u64)>,
|
||
last: u64,
|
||
pub reanchors: u64,
|
||
}
|
||
|
||
impl Timeline {
|
||
pub fn new(rate: u64, max_skew_us: u64) -> Self {
|
||
Timeline { rate, max_skew_us, anchor: None, last: 0, reanchors: 0 }
|
||
}
|
||
|
||
/// The room-clock time of a frame with extended RTP timestamp `rtp` that arrived at
|
||
/// room-clock `arrival`.
|
||
pub fn map(&mut self, rtp: u64, arrival: u64) -> u64 {
|
||
let t = match self.anchor {
|
||
Some((a, r0)) if rtp >= r0 => {
|
||
let t = a as u128 + (rtp - r0) as u128 * 1_000_000 / self.rate as u128;
|
||
let t = t.min(u64::MAX as u128) as u64;
|
||
if t.abs_diff(arrival) > self.max_skew_us {
|
||
self.reanchors += 1;
|
||
self.anchor = Some((arrival, rtp));
|
||
arrival
|
||
} else {
|
||
t
|
||
}
|
||
}
|
||
Some(_) => {
|
||
self.reanchors += 1;
|
||
self.anchor = Some((arrival, rtp));
|
||
arrival
|
||
}
|
||
None => {
|
||
self.anchor = Some((arrival, rtp));
|
||
arrival
|
||
}
|
||
};
|
||
self.last = self.last.max(t);
|
||
self.last
|
||
}
|
||
}
|
||
|
||
/// Where one media track and its receipts track are published.
|
||
#[derive(Clone, Debug)]
|
||
pub struct TrackOut {
|
||
/// A media kind of WAI live-rooms §6.1.
|
||
pub media: u8,
|
||
pub media_alias: u64,
|
||
pub receipts_alias: u64,
|
||
/// The `track_alias` the receipts sign, the catalog's `wai_receipt_alias`.
|
||
pub receipt_alias: u64,
|
||
pub priority: u8,
|
||
}
|
||
|
||
/// What a packager asks of its publisher.
|
||
#[derive(Clone, Debug)]
|
||
pub enum Emit {
|
||
/// One object of a media track's group. `first` opens the group's subgroup stream.
|
||
Object { alias: u64, group: u64, object: u64, first: bool, priority: u8, bytes: Vec<u8> },
|
||
/// The group's last object was sent: its stream ends with a FIN.
|
||
EndGroup { alias: u64, group: u64 },
|
||
/// A group of one object on a track of its own stream: a receipts bundle, or a catalog.
|
||
Single { alias: u64, group: u64, priority: u8, bytes: Vec<u8> },
|
||
}
|
||
|
||
/// What a frame did, beside what it emitted.
|
||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||
pub enum Verdict {
|
||
Carried,
|
||
/// Dropped: no group is open and the frame is not a keyframe. Ask for one.
|
||
NeedKeyframe,
|
||
/// Carried, and its group has run past `max_video_group_ms`: ask for a keyframe.
|
||
GroupLong,
|
||
/// Dropped: larger than `max_frame_bytes`.
|
||
TooLarge,
|
||
/// Dropped: a keyframe whose header gives no picture size, which its framing needs.
|
||
Unreadable,
|
||
}
|
||
|
||
/// Counts of one track, in the shape of a hop record's track entries (`crate::record`).
|
||
#[derive(Clone, Debug, Default, serde::Serialize)]
|
||
pub struct TrackCounts {
|
||
pub frames_in: u64,
|
||
pub bytes_in: u64,
|
||
pub objects_out: u64,
|
||
pub bytes_out: u64,
|
||
pub groups: u64,
|
||
pub dropped_before_keyframe: u64,
|
||
pub dropped_too_large: u64,
|
||
pub dropped_unreadable: u64,
|
||
pub reanchors: u64,
|
||
}
|
||
|
||
struct OpenGroup {
|
||
id: u64,
|
||
hashes: Vec<[u8; 32]>,
|
||
start_t: u64,
|
||
last_t: u64,
|
||
}
|
||
|
||
/// What signs: the gateway's key and the labels its receipts carry.
|
||
#[derive(Clone)]
|
||
pub struct Signer {
|
||
pub key: Arc<SigningKey>,
|
||
pub signer_id: String,
|
||
pub origin_label: String,
|
||
}
|
||
|
||
/// One media track of a WHIP session.
|
||
pub struct Packager {
|
||
pub out: TrackOut,
|
||
pub codec: Option<Codec>,
|
||
pub header: Option<Header>,
|
||
pub timeline: Timeline,
|
||
pub counts: TrackCounts,
|
||
group: Option<OpenGroup>,
|
||
next_group: Option<u64>,
|
||
prev_link: Option<[u8; 32]>,
|
||
last_ticks: Option<u64>,
|
||
grant: GrantRef,
|
||
signer: Signer,
|
||
t0_us: u64,
|
||
audio_group_us: u64,
|
||
max_video_group_us: u64,
|
||
max_frame_bytes: usize,
|
||
}
|
||
|
||
impl Packager {
|
||
#[allow(clippy::too_many_arguments)]
|
||
pub fn new(out: TrackOut, rate: u64, max_skew_us: u64, grant: GrantRef, signer: Signer, t0_us: u64, audio_group_us: u64, max_video_group_us: u64, max_frame_bytes: usize) -> Self {
|
||
Packager {
|
||
out,
|
||
codec: None,
|
||
header: None,
|
||
timeline: Timeline::new(rate, max_skew_us),
|
||
counts: TrackCounts::default(),
|
||
group: None,
|
||
next_group: None,
|
||
prev_link: None,
|
||
last_ticks: None,
|
||
grant,
|
||
signer,
|
||
t0_us,
|
||
audio_group_us: audio_group_us.max(1),
|
||
max_video_group_us,
|
||
max_frame_bytes,
|
||
}
|
||
}
|
||
|
||
pub fn is_video(&self) -> bool {
|
||
self.out.media != crate::room::media::AUDIO
|
||
}
|
||
|
||
/// One frame, depacketised: its codec, bytes, extended RTP timestamp and room-clock
|
||
/// arrival, and whether the WebRTC stack read it as a keyframe. Returns whether the
|
||
/// catalog must be published again (the track's codec or picture size changed).
|
||
pub fn on_frame(&mut self, codec: Codec, data: &[u8], rtp: u64, arrival: u64, keyframe: bool, emits: &mut Vec<Emit>) -> (Verdict, bool) {
|
||
self.counts.frames_in += 1;
|
||
self.counts.bytes_in += data.len() as u64;
|
||
if data.len() > self.max_frame_bytes {
|
||
self.counts.dropped_too_large += 1;
|
||
return (Verdict::TooLarge, false);
|
||
}
|
||
let mut catalog = false;
|
||
if self.codec != Some(codec) {
|
||
// A new codec begins a new group, at its first keyframe.
|
||
self.end_group(emits);
|
||
self.codec = Some(codec);
|
||
self.header = None;
|
||
catalog = true;
|
||
}
|
||
let t = self.timeline.map(rtp, arrival);
|
||
self.counts.reanchors = self.timeline.reanchors;
|
||
let key = match codec {
|
||
Codec::Avc => keyframe || frame::avc_is_idr(data),
|
||
Codec::Opus => false,
|
||
_ => keyframe,
|
||
};
|
||
|
||
// The group this frame belongs to.
|
||
if self.is_video() {
|
||
if key {
|
||
let h = match codec {
|
||
Codec::Avc => frame::avc_header(data).or_else(|| self.header.clone()),
|
||
Codec::Vp8 => frame::vp8_header(data),
|
||
Codec::Vp9 => frame::vp9_header(data),
|
||
Codec::Av1 => frame::av1_header(data).or_else(|| self.header.clone()),
|
||
Codec::Opus => None,
|
||
};
|
||
match h {
|
||
Some(h) => {
|
||
if self.header.as_ref() != Some(&h) {
|
||
catalog = true;
|
||
self.header = Some(h);
|
||
}
|
||
}
|
||
None if codec != Codec::Avc => {
|
||
self.counts.dropped_unreadable += 1;
|
||
return (Verdict::Unreadable, catalog);
|
||
}
|
||
None => {}
|
||
}
|
||
self.end_group(emits);
|
||
self.open_group(t);
|
||
} else if self.group.is_none() {
|
||
self.counts.dropped_before_keyframe += 1;
|
||
return (Verdict::NeedKeyframe, catalog);
|
||
}
|
||
} else {
|
||
if self.header.is_none() {
|
||
let ch = frame::opus_channels(data).unwrap_or(2);
|
||
self.header = Some(Header { width: 0, height: 0, codec: Some("opus".into()), channels: ch });
|
||
catalog = true;
|
||
}
|
||
let window_ends = self.group.as_ref().map(|g| g.start_t + self.audio_group_us);
|
||
if window_ends.is_none_or(|e| t >= e) {
|
||
self.end_group(emits);
|
||
// Windows are laid end to end: the next one starts where the last ended,
|
||
// or at this frame after a pause longer than a window.
|
||
let start = match window_ends {
|
||
Some(e) if t < e + self.audio_group_us => e,
|
||
_ => t,
|
||
};
|
||
self.open_group(start);
|
||
}
|
||
}
|
||
|
||
let ticks = t as u128 * 90_000 / 1_000_000;
|
||
let ticks = ticks as u64;
|
||
let delta = self.last_ticks.map_or(3_000, |p| ticks.saturating_sub(p));
|
||
self.last_ticks = Some(ticks);
|
||
let (w, h) = self.header.as_ref().map_or((0, 0), |h| (h.width, h.height));
|
||
let video = self.is_video();
|
||
let payload = frame::payload(codec, data, FrameInfo { width: w, height: h, ticks_90k: ticks, delta_90k: delta });
|
||
let g = self.group.as_mut().expect("a group is open");
|
||
let first = g.hashes.is_empty();
|
||
let live = Live {
|
||
t,
|
||
rtp,
|
||
rate: codec.clock_rate(),
|
||
key: codec.is_video().then_some(key),
|
||
grant: first.then(|| self.grant.clone()),
|
||
};
|
||
let envelope = frame::envelope(&frame::manifest(codec, &live), &payload);
|
||
g.hashes.push(ObjectReceipt::hash_payload(&envelope));
|
||
g.last_t = t;
|
||
let object = g.hashes.len() as u64 - 1;
|
||
self.counts.objects_out += 1;
|
||
self.counts.bytes_out += envelope.len() as u64;
|
||
emits.push(Emit::Object { alias: self.out.media_alias, group: g.id, object, first, priority: self.out.priority, bytes: envelope });
|
||
let long = video && t.saturating_sub(g.start_t) > self.max_video_group_us;
|
||
(if long { Verdict::GroupLong } else { Verdict::Carried }, catalog)
|
||
}
|
||
|
||
fn open_group(&mut self, start_t: u64) {
|
||
let id = self.next_group.unwrap_or(start_t);
|
||
self.next_group = Some(id + 1);
|
||
self.group = Some(OpenGroup { id, hashes: Vec::new(), start_t, last_t: start_t });
|
||
self.counts.groups += 1;
|
||
}
|
||
|
||
/// End the open group, if any: its stream's FIN, then its receipts, signed.
|
||
pub fn end_group(&mut self, emits: &mut Vec<Emit>) {
|
||
let Some(g) = self.group.take() else { return };
|
||
emits.push(Emit::EndGroup { alias: self.out.media_alias, group: g.id });
|
||
if g.hashes.is_empty() {
|
||
return;
|
||
}
|
||
let bundle = self.sign(&g);
|
||
emits.push(Emit::Single { alias: self.out.receipts_alias, group: g.id, priority: self.out.priority, bytes: bundle });
|
||
}
|
||
|
||
fn sign(&mut self, g: &OpenGroup) -> Vec<u8> {
|
||
let start = rfc3339(self.t0_us.saturating_add(g.start_t));
|
||
let end = rfc3339(self.t0_us.saturating_add(g.last_t));
|
||
let (bundle, link) = sign_group(&self.signer, self.out.receipt_alias, g.id, &g.hashes, &start, &end, self.prev_link);
|
||
self.prev_link = Some(link);
|
||
bundle
|
||
}
|
||
}
|
||
|
||
/// The receipts of one group, as group `group` of its receipts track carries them: the
|
||
/// group receipt, signed by `signer` over every object's labelled leaf and chained to
|
||
/// `parent`, and each object's receipt with its Merkle proof. Returns the bundle and the
|
||
/// group receipt's link hash, the next group's parent.
|
||
#[allow(clippy::too_many_arguments)]
|
||
pub fn sign_group(signer: &Signer, receipt_alias: u64, group: u64, hashes: &[[u8; 32]], start: &str, end: &str, parent: Option<[u8; 32]>) -> (Vec<u8>, [u8; 32]) {
|
||
let alias = TrackAlias(receipt_alias);
|
||
let mut b = ReceiptBuilder::new(alias, GroupId(group));
|
||
for h in hashes {
|
||
// No meter backs the gateway: every figure is the unmetered marker, 0 with class
|
||
// tag 0 (WAI energy-measurement §2), never a measurement of zero joules.
|
||
b.add_labelled_object(*h, 0, signer.origin_label.clone(), start.to_string(), vec![0]).expect("the unmetered marker is a class");
|
||
}
|
||
let objects: Vec<ObjectReceipt> = hashes
|
||
.iter()
|
||
.enumerate()
|
||
.map(|(i, h)| ObjectReceipt {
|
||
track_alias: alias,
|
||
group_id: GroupId(group),
|
||
object_id: ObjectId(i as u64),
|
||
content_hash: *h,
|
||
joules_micro: 0,
|
||
origin: signer.origin_label.clone(),
|
||
merkle_proof: b.proof_for(i).expect("an index of the group"),
|
||
energy_class: Some(vec![0]),
|
||
})
|
||
.collect();
|
||
let g = b.finalize_labelled(&signer.key, signer.signer_id.clone(), end.to_string(), parent).expect("the unmetered marker composes");
|
||
let link = g.link_hash();
|
||
(ReceiptBundle { group: g, objects }.to_payload().expect("a bundle serialises"), link)
|
||
}
|
||
|
||
/// An instant in µs since 1970 as RFC 3339 UTC with microseconds.
|
||
pub fn rfc3339(unix_us: u64) -> String {
|
||
let secs = unix_us / 1_000_000;
|
||
let micros = unix_us % 1_000_000;
|
||
let days = (secs / 86_400) as i64;
|
||
let rem = secs % 86_400;
|
||
// Civil from days (Howard Hinnant's algorithm), 1970-01-01 = day 0.
|
||
let z = days + 719_468;
|
||
let era = z.div_euclid(146_097);
|
||
let doe = z.rem_euclid(146_097);
|
||
let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365;
|
||
let y = yoe + era * 400;
|
||
let doy = doe - (365 * yoe + yoe / 4 - yoe / 100);
|
||
let mp = (5 * doy + 2) / 153;
|
||
let d = doy - (153 * mp + 2) / 5 + 1;
|
||
let m = if mp < 10 { mp + 3 } else { mp - 9 };
|
||
let y = if m <= 2 { y + 1 } else { y };
|
||
format!("{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.{micros:06}Z", rem / 3600, rem / 60 % 60, rem % 60)
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod tests {
|
||
use super::*;
|
||
use joule_moq::receipt::ReceiptVerifier;
|
||
|
||
#[test]
|
||
fn rfc3339_dates() {
|
||
assert_eq!(rfc3339(0), "1970-01-01T00:00:00.000000Z");
|
||
assert_eq!(rfc3339(1_792_281_600_123_456), "2026-10-18T00:00:00.123456Z");
|
||
assert_eq!(rfc3339(951_782_400_000_000), "2000-02-29T00:00:00.000000Z");
|
||
}
|
||
|
||
#[test]
|
||
fn the_timeline_holds_its_bound() {
|
||
let mut tl = Timeline::new(90_000, 250_000);
|
||
assert_eq!(tl.map(1_000, 5_000_000), 5_000_000);
|
||
// 90 000 ticks later: one second later, arriving 40 ms late.
|
||
assert_eq!(tl.map(91_000, 6_040_000), 6_000_000);
|
||
// A frame whose mapping would stray 300 ms from its arrival re-anchors.
|
||
assert_eq!(tl.map(181_000, 7_300_000), 7_300_000);
|
||
assert_eq!(tl.reanchors, 1);
|
||
// Mapped times never decrease.
|
||
assert_eq!(tl.map(100, 7_000_000), 7_300_000);
|
||
}
|
||
|
||
fn packager(media: u8) -> Packager {
|
||
let signer = Signer { key: Arc::new(SigningKey::from_bytes(&[3; 32])), signer_id: "test".into(), origin_label: "wai-ingest".into() };
|
||
let out = TrackOut { media, media_alias: 2, receipts_alias: 3, receipt_alias: 77, priority: 1 };
|
||
let rate = if media == crate::room::media::AUDIO { 48_000 } else { 90_000 };
|
||
Packager::new(out, rate, 250_000, GrantRef { grant_hash: [9; 32], joule_ceiling_micro: u64::MAX, funds_ceiling: None }, signer, 1_000_000_000, 200_000, 2_000_000, 1 << 20)
|
||
}
|
||
|
||
fn avc(idr: bool, n: u8) -> Vec<u8> {
|
||
let mut f = vec![0, 0, 0, 1, 0x67, 0x42, 0xe0, 0x1f, 0, 0, 0, 1, 0x68, 0xce];
|
||
f.truncate(if idr { 14 } else { 0 });
|
||
f.extend_from_slice(&[0, 0, 0, 1, if idr { 0x65 } else { 0x41 }, n, n, n]);
|
||
f
|
||
}
|
||
|
||
#[test]
|
||
fn video_groups_begin_at_keyframes_and_their_receipts_verify() {
|
||
let mut p = packager(crate::room::media::VIDEO);
|
||
let mut e = Vec::new();
|
||
assert_eq!(p.on_frame(Codec::Avc, &avc(false, 1), 0, 10, false, &mut e).0, Verdict::NeedKeyframe);
|
||
assert!(e.is_empty());
|
||
for i in 0..6u64 {
|
||
let idr = i % 3 == 0;
|
||
let (v, _) = p.on_frame(Codec::Avc, &avc(idr, i as u8 + 2), 3000 * i, 1_000 + 33_333 * i, idr, &mut e);
|
||
assert_eq!(v, Verdict::Carried);
|
||
}
|
||
p.end_group(&mut e);
|
||
let objects: Vec<_> = e.iter().filter_map(|x| if let Emit::Object { group, object, bytes, .. } = x { Some((*group, *object, bytes.clone())) } else { None }).collect();
|
||
let bundles: Vec<_> = e.iter().filter_map(|x| if let Emit::Single { group, bytes, .. } = x { Some((*group, ReceiptBundle::from_payload(bytes).unwrap())) } else { None }).collect();
|
||
assert_eq!(objects.len(), 6);
|
||
assert_eq!(bundles.len(), 2, "two groups, each ended by the keyframe after it or by the end");
|
||
let first = objects[0].0;
|
||
// The dropped frame anchored the timeline at its arrival, 10 µs; the first IDR's RTP
|
||
// time maps there too.
|
||
assert_eq!(first, 10, "the first group's ID is its first frame's room-clock time");
|
||
assert_eq!(bundles[1].0, first + 1);
|
||
let v = ReceiptVerifier::with_anchors(vec![SigningKey::from_bytes(&[3; 32]).verifying_key()]);
|
||
for (g, b) in &bundles {
|
||
let vg = v.verify_group(&b.group).unwrap();
|
||
assert_eq!(b.group.track_alias, TrackAlias(77));
|
||
for (og, o, bytes) in objects.iter().filter(|o| o.0 == *g) {
|
||
let r = &b.objects[*o as usize];
|
||
r.check_moqt(TrackAlias(77), GroupId(*og), ObjectId(*o)).unwrap();
|
||
v.verify_payload(bytes, r, &vg).unwrap();
|
||
}
|
||
}
|
||
assert_eq!(bundles[1].1.group.parent_receipt_hash, Some(bundles[0].1.group.link_hash()), "each group chains to the one before");
|
||
// The first object of a group names the grant; the others do not.
|
||
let o0 = frame::open(&objects[0].2).unwrap();
|
||
assert!(o0.manifest["live"]["grant"].is_object() && o0.key);
|
||
let o1 = frame::open(&objects[1].2).unwrap();
|
||
assert!(o1.manifest["live"].get("grant").is_none() && !o1.key);
|
||
}
|
||
|
||
#[test]
|
||
fn audio_groups_are_fixed_windows() {
|
||
let mut p = packager(crate::room::media::AUDIO);
|
||
let mut e = Vec::new();
|
||
// 20 ms packets for 1 s: five windows of 200 ms.
|
||
for i in 0..50u64 {
|
||
p.on_frame(Codec::Opus, &[0xfc, i as u8, 1, 2], 960 * i, 2_000_000 + 20_000 * i, false, &mut e);
|
||
}
|
||
p.end_group(&mut e);
|
||
let groups: Vec<u64> = e.iter().filter_map(|x| if let Emit::Single { group, .. } = x { Some(*group) } else { None }).collect();
|
||
assert_eq!(groups.len(), 5);
|
||
assert!(groups.windows(2).all(|w| w[1] == w[0] + 1));
|
||
let per_group = e.iter().filter(|x| matches!(x, Emit::Object { .. })).count() / 5;
|
||
assert_eq!(per_group, 10);
|
||
}
|
||
}
|