- commit
- 19a847649deb04bc1ae156fa5c9b13a3e1c21dc0
- parent
- 093e2c5d81c23d46f7d25aa6df1fc1e4fd41341f
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 22 | +- |
| M | crates/ | 6 | +- |
| M | crates/ | 32 | ++- |
| A | crates/ | 464 | ++++++++++++++++++++++++++++++ |
| M | crates/ | 4 | +- |
| M | crates/ | 4 | +- |
| M | crates/ | 424 | ++++++++++++++++++++++----- |
| M | crates/ | 59 | +++- |
| M | crates/ | 30 | ++- |
| M | crates/ | 6 | + |
| M | crates/ | 88 | ++++++- |
| M | scripts/ | 31 | ++ |
| M | src-tauri/ | 179 | ++++++++++++ |
| M | src-tauri/ | 75 | +++- |
| M | src-tauri/ | 22 | +- |
| M | src-tauri/ | 239 | +++++++++++++++- |
| M | src-tauri/ | 2 | +- |
@@ -508,11 +508,16 @@ impl MoqProbeReceiver {
508508 endpoint: &Endpoint,509509 address: EndpointAddr,510510 session_id: u64,511+ generation_id: u64,512+ advertised_broadcast_name: Option<&str>,511513 timeout: Duration,512514 authentication: Option<MoqProbeAuthentication>,513515 ) -> Result<Self> {514- let broadcast_name = media_moq_broadcast_name(session_id)515- .context("deriving session-scoped MoQ broadcast name")?;516+ let broadcast_name = match advertised_broadcast_name {517+ Some(name) => name.to_owned(),518+ None => media_moq_broadcast_name(session_id)519+ .context("deriving session-scoped MoQ broadcast name")?,520+ };516521 let moq = Moq::new(endpoint.clone());517522 let mut session = tokio::time::timeout(timeout, moq.connect(address.clone()))518523 .await@@ -534,7 +539,7 @@ impl MoqProbeReceiver {
534539 SignedSubscriptionCapability::decode(&authentication.subscription_capability)?535540 .verify_binding(536541 authentication.expected_host,537- session_id,542+ generation_id,538543 *endpoint.id().as_bytes(),539544 SubscriptionTracks::VIDEO_H264,540545 1,@@ -544,7 +549,7 @@ impl MoqProbeReceiver {
544549 &broadcast,545550 timeout,546551 authentication.expected_host,547- session_id,552+ generation_id,548553 authentication.now_unix,549554 )550555 .await@@ -788,11 +793,15 @@ async fn run_control_v2_smoke(
788793 initial.revision > 0 && matches!(initial.focus, FocusStateV2::Vacant { .. }),789794 "initial control v2 snapshot was not a revisioned vacant slot"790795 );796+ let generation_id = initial.media.generation_id;797+ let broadcast_name = initial.media.broadcast_name.clone();791798 792799 let mut media = MoqProbeReceiver::connect(793800 endpoint,794801 address.clone(),795802 session_id,803+ generation_id,804+ Some(&broadcast_name),796805 Duration::from_secs(timeout_seconds),797806 Some(MoqProbeAuthentication {798807 expected_host: *address.id.as_bytes(),@@ -945,6 +954,9 @@ async fn run_control_v2_smoke(
945954 println!("input_alpn=sigil/input/2");946955 println!("media_authentication=ed25519-v1");947956 println!("media_generation_certified=ok");957+ println!("media_generation_id={generation_id}");958+ println!("media_broadcast_name={broadcast_name}");959+ println!("media_generation_scope=host");948960 println!("subscription_endpoint_binding=ok");949961 println!("frames={accepted}");950962 println!("initial_snapshot_revision={}", initial.revision);@@ -1629,6 +1641,8 @@ async fn main() -> Result<()> {
16291641 &endpoint,16301642 address.clone(),16311643 session_id,1644+ session_id,1645+ None,16321646 Duration::from_secs(args.timeout_seconds),16331647 None,16341648 )@@ -42,7 +42,8 @@ use crate::cursor::PointerPositionTracker;
4242 use crate::input::InputBackend;4343 use crate::server::{4444 AudioHandler, AuthorizedMoqHandler, ControlHandler, ControlV2Handler, InputHandler,45- InputOperations, InputV2Handler, MediaFeedbackHandler, MediaV3Handler, SessionRegistry,45+ InputOperations, InputV2Handler, MediaFeedbackHandler, MediaGenerationManager, MediaV3Handler,46+ SessionRegistry,4647 };4748 4849 const CONNECTION_IDLE_TIMEOUT: Duration = Duration::from_secs(5);@@ -1111,6 +1112,7 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
11111112 11121113 let moq_origin = Origin::random();11131114 let input_operations = Arc::new(InputOperations::new(input_backend.clone()));1115+ let media_generations = MediaGenerationManager::new(config.clone(), host_secret);11141116 let authorization_admin = match &authorization {11151117 AuthorizationPolicy::Required(store) => Some(11161118 authorization_admin::AuthorizationAdminServer::start(@@ -1128,8 +1130,8 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
11281130 .accept(11291131 CONTROL_ALPN_V2,11301132 ControlV2Handler {1131- config: config.clone(),11321133 sessions: Arc::clone(&sessions),1134+ generations: Arc::clone(&media_generations),11331135 authorization: authorization.clone(),11341136 input_operations: Arc::clone(&input_operations),11351137 host_secret,@@ -44,13 +44,18 @@ pub(crate) fn publish_goq_catalog_v2(
4444 broadcast: &mut BroadcastProducer,4545 generation_id: u64,4646 certificate: &SignedMediaGenerationCertificate,47+ audio_enabled: bool,4748 ) -> Result<GoqCatalogProducer> {4849 let track = broadcast4950 .create_track(hang::Catalog::default_track())5051 .context("creating authenticated catalog.json track")?;5152 let mut producer = moq_json::Producer::new(track, moq_json::Config::default());52- let document = GoqCatalogDocumentV2::video_h264(generation_id, certificate)53- .context("constructing authenticated Goq catalog")?;53+ let document = if audio_enabled {54+ GoqCatalogDocumentV2::video_h264_opus(generation_id, certificate)55+ } else {56+ GoqCatalogDocumentV2::video_h264(generation_id, certificate)57+ }58+ .context("constructing authenticated Goq catalog")?;5459 document5560 .validate()5661 .context("validating authenticated Goq catalog document")?;@@ -134,7 +139,7 @@ mod tests {
134139 })135140 .unwrap();136141 let certificate = certificate();137- let catalog = publish_goq_catalog_v2(&mut broadcast, 42, &certificate).unwrap();142+ let catalog = publish_goq_catalog_v2(&mut broadcast, 42, &certificate, false).unwrap();138143 let document_track = broadcast139144 .consume()140145 .subscribe_track(&hang::Catalog::default_track())@@ -148,4 +153,25 @@ mod tests {
148153 assert_eq!(document.validate().unwrap(), certificate);149154 catalog.finish().unwrap();150155 }156+157+ 158+ async fn authenticated_catalog_can_advertise_video_and_audio_tracks() {159+ let mut broadcast = Broadcast::new().produce();160+ let _video = broadcast161+ .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))162+ .unwrap();163+ let _audio = broadcast164+ .create_track(Track::new(sigil_protocol::MOQ_AUDIO_OPUS_TRACK))165+ .unwrap();166+ let certificate = certificate();167+ let catalog = publish_goq_catalog_v2(&mut broadcast, 42, &certificate, true).unwrap();168+ let track = broadcast169+ .consume()170+ .subscribe_track(&hang::Catalog::default_track())171+ .unwrap();172+ let mut consumer = moq_json::Consumer::<GoqCatalogDocumentV2>::new(track);173+ let document = consumer.next().await.unwrap().unwrap();174+ assert!(document.goq.audio.is_some());175+ catalog.finish().unwrap();176+ }151177 }@@ -0,0 +1,464 @@
1+use std::sync::atomic::{AtomicU64, Ordering};2+use std::sync::{Arc, Mutex, Weak};3+use std::time::Duration;4+5+use anyhow::{Context, Result};6+use moq_net::{Broadcast, BroadcastConsumer, BroadcastProducer, Error as MoqError, Track};7+use sigil_protocol::{8+ MOQ_AUDIO_OPUS_TRACK, MOQ_AUDIO_TRACK_PRIORITY, MOQ_VIDEO_H264_TRACK, MOQ_VIDEO_TRACK_PRIORITY,9+ MediaGenerationSigningKey, SignedMediaGenerationCertificate,10+ media_generation_moq_broadcast_name,11+};12+use tokio::sync::{Notify, watch};13+use tracing::{debug, info, warn};14+15+use super::moq::{run_generation_audio_publisher, run_generation_video_publisher};16+use super::session::{MediaV3Telemetry, SourceTaskGuard};17+use crate::audio::spawn_pipewire_audio;18+use crate::authorization::unix_timestamp_now;19+use crate::clock::SessionClock;20+use crate::config::{HostConfig, VideoSource};21+use crate::moq_catalog::{GoqCatalogProducer, publish_goq_catalog_v2};22+use crate::source::{23+ EncodedFrame, EncodedSource, EncoderControl, spawn_gamescope_pipewire_after_static_preflight,24+ spawn_test_pattern,25+};26+27+const GENERATION_STOP_TIMEOUT: Duration = Duration::from_secs(2);28+29+30+pub(super) enum MediaGenerationLifecycle {31+ Idle,32+ Starting,33+ Active,34+ Stopping,35+}36+37+38+struct GenerationState {39+ lifecycle: MediaGenerationLifecycle,40+ active: Option<ActiveGeneration>,41+ leases: usize,42+}43+44+impl Default for GenerationState {45+ fn default() -> Self {46+ Self {47+ lifecycle: MediaGenerationLifecycle::Idle,48+ active: None,49+ leases: 0,50+ }51+ }52+}53+54+55+pub(crate) struct MediaGenerationManager {56+ config: HostConfig,57+ host_secret: [u8; 32],58+ next_generation_id: AtomicU64,59+ state: tokio::sync::Mutex<GenerationState>,60+ state_changed: Notify,61+ starts: AtomicU64,62+ stops: AtomicU64,63+}64+65+impl MediaGenerationManager {66+ pub(crate) fn new(config: HostConfig, host_secret: [u8; 32]) -> Arc<Self> {67+ Arc::new(Self {68+ config,69+ host_secret,70+ next_generation_id: AtomicU64::new(0),71+ state: tokio::sync::Mutex::new(GenerationState::default()),72+ state_changed: Notify::new(),73+ starts: AtomicU64::new(0),74+ stops: AtomicU64::new(0),75+ })76+ }77+78+ pub(super) async fn acquire(self: &Arc<Self>) -> Result<MediaGenerationLease> {79+ loop {80+ let mut state = self.state.lock().await;81+ match state.lifecycle {82+ MediaGenerationLifecycle::Active => {83+ state.leases = state84+ .leases85+ .checked_add(1)86+ .context("media generation lease count overflowed")?;87+ let shared = state88+ .active89+ .as_ref()90+ .context("active generation has no resources")?91+ .shared92+ .clone();93+ return Ok(MediaGenerationLease::new(self, shared));94+ }95+ MediaGenerationLifecycle::Idle => {96+ state.lifecycle = MediaGenerationLifecycle::Starting;97+ drop(state);98+ let generation_id = self99+ .next_generation_id100+ .fetch_add(1, Ordering::Relaxed)101+ .checked_add(1)102+ .context("media generation id overflowed")?;103+ match ActiveGeneration::start(&self.config, self.host_secret, generation_id)104+ .await105+ {106+ Ok(active) => {107+ let shared = active.shared.clone();108+ let mut state = self.state.lock().await;109+ state.active = Some(active);110+ state.leases = 1;111+ state.lifecycle = MediaGenerationLifecycle::Active;112+ self.starts.fetch_add(1, Ordering::Relaxed);113+ self.state_changed.notify_waiters();114+ return Ok(MediaGenerationLease::new(self, shared));115+ }116+ Err(error) => {117+ let mut state = self.state.lock().await;118+ state.lifecycle = MediaGenerationLifecycle::Idle;119+ self.state_changed.notify_waiters();120+ return Err(error);121+ }122+ }123+ }124+ MediaGenerationLifecycle::Starting | MediaGenerationLifecycle::Stopping => {125+ let changed = self.state_changed.notified();126+ drop(state);127+ changed.await;128+ }129+ }130+ }131+ }132+133+ async fn release(&self, generation_id: u64) {134+ let active = {135+ let mut state = self.state.lock().await;136+ if state.lifecycle != MediaGenerationLifecycle::Active137+ || state138+ .active139+ .as_ref()140+ .map(|active| active.shared.generation_id)141+ != Some(generation_id)142+ {143+ return;144+ }145+ state.leases = state.leases.saturating_sub(1);146+ if state.leases != 0 {147+ return;148+ }149+ state.lifecycle = MediaGenerationLifecycle::Stopping;150+ state.active.take()151+ };152+ if let Some(active) = active {153+ active.stop().await;154+ self.stops.fetch_add(1, Ordering::Relaxed);155+ }156+ let mut state = self.state.lock().await;157+ state.lifecycle = MediaGenerationLifecycle::Idle;158+ self.state_changed.notify_waiters();159+ }160+161+ 162+ async fn lifecycle(&self) -> MediaGenerationLifecycle {163+ self.state.lock().await.lifecycle164+ }165+}166+167+pub(super) struct GenerationShared {168+ pub(super) generation_id: u64,169+ pub(super) broadcast_name: String,170+ broadcast: Mutex<Option<BroadcastConsumer>>,171+ pub(super) certificate: SignedMediaGenerationCertificate,172+ pub(super) session_clock: SessionClock,173+ pub(super) pointer_surface_dimensions: Option<sigil_protocol::PointerSurfaceDimensions>,174+ pub(super) encoder_control: Option<EncoderControl>,175+ pub(super) telemetry: Arc<MediaV3Telemetry>,176+ pub(super) audio_enabled: bool,177+ pub(super) keyframe_requests: watch::Sender<Option<sigil_protocol::MediaControlRequestV3>>,178+}179+180+impl std::fmt::Debug for GenerationShared {181+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {182+ formatter183+ .debug_struct("GenerationShared")184+ .field("generation_id", &self.generation_id)185+ .field("broadcast_name", &self.broadcast_name)186+ .field("audio_enabled", &self.audio_enabled)187+ .finish_non_exhaustive()188+ }189+}190+191+impl GenerationShared {192+ pub(super) fn consumer(&self) -> Result<BroadcastConsumer> {193+ self.broadcast194+ .lock()195+ .map_err(|_| anyhow::anyhow!("media generation broadcast lock poisoned"))?196+ .as_ref()197+ .cloned()198+ .context("media generation broadcast is stopping")199+ }200+}201+202+203+pub(super) struct MediaGenerationLease {204+ manager: Weak<MediaGenerationManager>,205+ pub(super) shared: Arc<GenerationShared>,206+ released: bool,207+}208+209+impl MediaGenerationLease {210+ fn new(manager: &Arc<MediaGenerationManager>, shared: Arc<GenerationShared>) -> Self {211+ Self {212+ manager: Arc::downgrade(manager),213+ shared,214+ released: false,215+ }216+ }217+218+ pub(super) async fn release(mut self) {219+ self.release_inner().await;220+ self.released = true;221+ }222+223+ async fn release_inner(&self) {224+ if let Some(manager) = self.manager.upgrade() {225+ manager.release(self.shared.generation_id).await;226+ }227+ }228+}229+230+impl Drop for MediaGenerationLease {231+ fn drop(&mut self) {232+ if self.released {233+ return;234+ }235+ let Some(manager) = self.manager.upgrade() else {236+ return;237+ };238+ let generation_id = self.shared.generation_id;239+ if let Ok(runtime) = tokio::runtime::Handle::try_current() {240+ runtime.spawn(async move { manager.release(generation_id).await });241+ }242+ }243+}244+245+struct ActiveGeneration {246+ shared: Arc<GenerationShared>,247+ shutdown: watch::Sender<bool>,248+ video_publisher: tokio::task::JoinHandle<Result<()>>,249+ audio_publisher: Option<tokio::task::JoinHandle<Result<()>>>,250+ video_source: SourceTaskGuard,251+ audio_source: Option<SourceTaskGuard>,252+ _frame_receiver: watch::Receiver<Option<EncodedFrame>>,253+ catalog: Option<GoqCatalogProducer>,254+ broadcast: Option<BroadcastProducer>,255+}256+257+impl std::fmt::Debug for ActiveGeneration {258+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {259+ formatter260+ .debug_struct("ActiveGeneration")261+ .field("generation_id", &self.shared.generation_id)262+ .field("audio_enabled", &self.shared.audio_enabled)263+ .finish_non_exhaustive()264+ }265+}266+267+impl ActiveGeneration {268+ async fn start(config: &HostConfig, host_secret: [u8; 32], generation_id: u64) -> Result<Self> {269+ let session_clock = SessionClock::start();270+ let source = match config.source {271+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), session_clock)),272+ VideoSource::GamescopePipewire => {273+ let primary =274+ spawn_gamescope_pipewire_after_static_preflight(config.clone(), session_clock)275+ .await?;276+ super::select_gamescope_startup_source(config.clone(), session_clock, primary).await277+ }278+ }?;279+ let EncodedSource {280+ frames,281+ current_gop,282+ task,283+ pointer_surface_dimensions,284+ encoder_control,285+ } = source;286+ let video_source = SourceTaskGuard::new(task);287+288+ let mut generation_secret = [0_u8; 32];289+ getrandom::fill(&mut generation_secret)290+ .context("generating shared media authentication key")?;291+ let signing_key = MediaGenerationSigningKey::from_bytes(&generation_secret);292+ generation_secret.fill(0);293+ let issued_at_unix = unix_timestamp_now()?;294+ let certificate = signing_key.certify(295+ *iroh::SecretKey::from_bytes(&host_secret)296+ .public()297+ .as_bytes(),298+ &host_secret,299+ generation_id,300+ issued_at_unix,301+ issued_at_unix.saturating_add(60 * 60),302+ )?;303+304+ let mut broadcast = Broadcast::new().produce();305+ let video_track = broadcast306+ .create_track(Track {307+ name: MOQ_VIDEO_H264_TRACK.to_owned(),308+ priority: MOQ_VIDEO_TRACK_PRIORITY,309+ })310+ .context("creating generation H.264 track")?;311+ let audio_enabled = config.audio.is_some();312+ let audio_track = if audio_enabled {313+ Some(314+ broadcast315+ .create_track(Track {316+ name: MOQ_AUDIO_OPUS_TRACK.to_owned(),317+ priority: MOQ_AUDIO_TRACK_PRIORITY,318+ })319+ .context("creating generation Opus track")?,320+ )321+ } else {322+ None323+ };324+ let catalog =325+ publish_goq_catalog_v2(&mut broadcast, generation_id, &certificate, audio_enabled)?;326+ let broadcast_consumer = broadcast.consume();327+ let broadcast_name = media_generation_moq_broadcast_name(generation_id)?;328+ let telemetry = Arc::new(MediaV3Telemetry::default());329+ let (keyframe_requests, keyframes) = watch::channel(None);330+ let (shutdown, shutdown_rx) = watch::channel(false);331+ let video_publisher = tokio::spawn(run_generation_video_publisher(332+ config.clone(),333+ current_gop,334+ keyframes,335+ video_track,336+ generation_id,337+ signing_key.clone(),338+ encoder_control.clone(),339+ Arc::clone(&telemetry),340+ shutdown_rx.clone(),341+ ));342+343+ let (audio_source, audio_publisher) = if let Some(audio_track) = audio_track {344+ let (audio_packets, task) =345+ match spawn_pipewire_audio(config.clone(), session_clock).await {346+ Ok(source) => source,347+ Err(error) => {348+ video_publisher.abort();349+ video_source.abort_and_wait().await;350+ return Err(error).context("starting generation Opus source");351+ }352+ };353+ (354+ Some(SourceTaskGuard::new(task)),355+ Some(tokio::spawn(run_generation_audio_publisher(356+ audio_packets,357+ audio_track,358+ generation_id,359+ signing_key,360+ shutdown_rx,361+ ))),362+ )363+ } else {364+ (None, None)365+ };366+367+ let shared = Arc::new(GenerationShared {368+ generation_id,369+ broadcast_name,370+ broadcast: Mutex::new(Some(broadcast_consumer)),371+ certificate,372+ session_clock,373+ pointer_surface_dimensions,374+ encoder_control,375+ telemetry,376+ audio_enabled,377+ keyframe_requests,378+ });379+ info!(380+ generation_id,381+ video_sources = 1,382+ video_encoders = 1,383+ audio_sources = usize::from(audio_enabled),384+ audio_encoders = usize::from(audio_enabled),385+ publishers = 1,386+ "shared media generation started"387+ );388+ Ok(Self {389+ shared,390+ shutdown,391+ video_publisher,392+ audio_publisher,393+ video_source,394+ audio_source,395+ _frame_receiver: frames,396+ catalog: Some(catalog),397+ broadcast: Some(broadcast),398+ })399+ }400+401+ async fn stop(mut self) {402+ let generation_id = self.shared.generation_id;403+ if let Ok(mut broadcast) = self.shared.broadcast.lock() {404+ *broadcast = None;405+ }406+ self.shutdown.send_replace(true);407+ wait_or_abort_task(&mut self.video_publisher, "video publisher").await;408+ if let Some(mut audio_publisher) = self.audio_publisher.take() {409+ wait_or_abort_task(&mut audio_publisher, "audio publisher").await;410+ }411+ if let Some(catalog) = self.catalog.take()412+ && let Err(error) = catalog.finish()413+ {414+ warn!(generation_id, %error, "finishing shared media catalog failed");415+ }416+ if let Some(mut broadcast) = self.broadcast.take() {417+ let _ = broadcast.abort(MoqError::Cancel);418+ }419+ if let Some(audio_source) = self.audio_source.take() {420+ audio_source.abort_and_wait().await;421+ }422+ self.video_source423+ .wait_or_abort(super::SOURCE_REAP_GRACE_TIMEOUT)424+ .await;425+ info!(426+ generation_id,427+ "shared media generation stopped with all sources cleaned up"428+ );429+ }430+}431+432+async fn wait_or_abort_task(task: &mut tokio::task::JoinHandle<Result<()>>, name: &'static str) {433+ match tokio::time::timeout(GENERATION_STOP_TIMEOUT, &mut *task).await {434+ Ok(Ok(Ok(()))) => {}435+ Ok(Ok(Err(error))) => warn!(%error, task = name, "media generation task ended with error"),436+ Ok(Err(error)) => warn!(%error, task = name, "media generation task failed"),437+ Err(_) => {438+ debug!(task = name, "aborting stalled media generation task");439+ task.abort();440+ let _ = task.await;441+ }442+ }443+}444+445+446+mod tests {447+ use super::*;448+449+ 450+ fn lifecycle_is_explicit_and_bounded() {451+ let state = GenerationState::default();452+ assert_eq!(state.lifecycle, MediaGenerationLifecycle::Idle);453+ assert_eq!(state.leases, 0);454+ assert!(state.active.is_none());455+ }456+457+ 458+ async fn manager_begins_idle_without_starting_capture() {459+ let manager = MediaGenerationManager::new(super::super::moq_test_config(), [7; 32]);460+ assert_eq!(manager.lifecycle().await, MediaGenerationLifecycle::Idle);461+ assert_eq!(manager.starts.load(Ordering::Relaxed), 0);462+ assert_eq!(manager.stops.load(Ordering::Relaxed), 0);463+ }464+}@@ -67,8 +67,8 @@ pub struct ControlHandler {
6767 6868 6969 pub struct ControlV2Handler {70- pub config: HostConfig,7170 pub sessions: Arc<SessionRegistry>,71+ pub generations: Arc<MediaGenerationManager>,7272 pub authorization: AuthorizationPolicy,7373 pub input_operations: Arc<InputOperations>,7474 pub host_secret: [u8; 32],@@ -267,8 +267,8 @@ impl ProtocolHandler for ControlV2Handler {
267267 let remote = connection.remote_id();268268 if let Err(error) = serve_control_moq_v2(269269 connection,270- self.config.clone(),271270 &self.sessions,271+ &self.generations,272272 &self.authorization,273273 &self.input_operations,274274 self.host_secret,@@ -36,7 +36,7 @@ use crate::clock::SessionClock;
3636 use crate::config::{GamescopeEncoderBackend, HostConfig, VaapiRateControl, VideoSource};3737 use crate::cursor::{PointerPositionTracker, PointerState};3838 use crate::input::{InputBackend, InputDisposition};39-use crate::moq_catalog::{publish_goq_catalog, publish_goq_catalog_v2};39+use crate::moq_catalog::publish_goq_catalog;4040 use crate::source::{4141 EncodedFrame, EncodedGop, EncodedSource, EncoderControl,4242 spawn_gamescope_pipewire_after_static_preflight, spawn_test_pattern,@@ -49,6 +49,7 @@ const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);
4949 const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);5050 5151 mod adaptive;52+mod generation;5253 mod handlers;5354 mod media_v3;5455 mod moq;@@ -57,6 +58,7 @@ mod startup;
5758 5859 pub(crate) use adaptive::VideoDimensions;5960 use adaptive::serve_media_feedback;61+pub(crate) use generation::MediaGenerationManager;6062 pub use handlers::{6163 AudioHandler, AuthorizedMoqHandler, ControlHandler, ControlV2Handler, InputHandler,6264 InputOperations, InputV2Handler, MediaFeedbackHandler, MediaV3Handler,@@ -17,7 +17,7 @@ enum MoqGroupDecision {
1717 /// Owns the single bounded live MoQ track. One configured H.264 GOP maps to1818 /// one native MoQ group; its application frame sequence remains inside the1919 /// encoded object envelope and is never reused as the transport group id.20-struct MoqGroupPublisher {20+pub(super) struct MoqGroupPublisher {2121 track: TrackProducer,2222 current_group: Option<GroupProducer>,2323 cursor: MediaV3GroupCursor,@@ -195,6 +195,279 @@ impl MoqGroupPublisher {
195195 }196196 }197197 198+const MOQ_AUDIO_PACKETS_PER_GROUP: usize = 5;199+const MAX_MOQ_AUDIO_GROUP_BYTES: usize =200+ MOQ_AUDIO_PACKETS_PER_GROUP * (AUDIO_HEADER_LEN + MAX_AUDIO_PAYLOAD_LEN + 256);201+202+struct MoqAudioPublisher {203+ track: TrackProducer,204+ current_group: Option<GroupProducer>,205+ object_count: usize,206+ object_bytes: usize,207+ generation_id: u64,208+ signing_key: MediaGenerationSigningKey,209+}210+211+impl MoqAudioPublisher {212+ fn new(213+ track: TrackProducer,214+ generation_id: u64,215+ signing_key: MediaGenerationSigningKey,216+ ) -> Self {217+ Self {218+ track,219+ current_group: None,220+ object_count: 0,221+ object_bytes: 0,222+ generation_id,223+ signing_key,224+ }225+ }226+227+ fn publish(&mut self, packet: crate::audio::EncodedAudioPacket) -> Result<()> {228+ let flags = if packet.discontinuity {229+ AudioFlags::DISCONTINUITY230+ } else {231+ AudioFlags::NONE232+ };233+ let payload = AudioPacket::new(234+ AudioPacketHeader::opus(235+ packet.payload.len(),236+ packet.sequence,237+ packet.capture_timestamp_us,238+ packet.pts_us,239+ flags,240+ )?,241+ packet.payload.as_ref().to_vec(),242+ )?243+ .encode_datagram()?;244+245+ if self.current_group.is_none()246+ || self.object_count >= MOQ_AUDIO_PACKETS_PER_GROUP247+ || packet.discontinuity248+ {249+ self.finish_group()?;250+ self.current_group = Some(251+ self.track252+ .append_group()253+ .context("creating bounded MoQ Opus group")?,254+ );255+ }256+ let group = self257+ .current_group258+ .as_mut()259+ .context("Opus packet has no active MoQ group")?;260+ let object_id = u32::try_from(self.object_count).context("audio object id overflowed")?;261+ let object = self.signing_key.authenticate(262+ MediaObjectCoordinates {263+ generation_id: self.generation_id,264+ track: MediaTrack::AudioOpus,265+ group_id: group.sequence,266+ object_id,267+ flags: u16::from(flags.bits()),268+ },269+ &payload,270+ )?;271+ let next_bytes = self272+ .object_bytes273+ .checked_add(object.as_bytes().len())274+ .filter(|bytes| *bytes <= MAX_MOQ_AUDIO_GROUP_BYTES)275+ .context("bounded MoQ Opus group exceeded its byte limit")?;276+ group277+ .write_frame(object.into_bytes())278+ .context("writing authenticated Opus object")?;279+ self.object_count += 1;280+ self.object_bytes = next_bytes;281+ if self.object_count == MOQ_AUDIO_PACKETS_PER_GROUP {282+ self.finish_group()?;283+ }284+ Ok(())285+ }286+287+ fn finish_group(&mut self) -> Result<()> {288+ if let Some(mut group) = self.current_group.take() {289+ group.finish().context("finishing bounded MoQ Opus group")?;290+ }291+ self.object_count = 0;292+ self.object_bytes = 0;293+ Ok(())294+ }295+296+ fn abort(mut self) {297+ if let Some(mut group) = self.current_group.take() {298+ let _ = group.abort(MoqError::Cancel);299+ }300+ let _ = self.track.abort(MoqError::Cancel);301+ }302+}303+304+305+pub(super) async fn run_generation_video_publisher(306+ config: HostConfig,307+ mut current_gop_receiver: tokio::sync::watch::Receiver<Option<EncodedGop>>,308+ mut control_requests: tokio::sync::watch::Receiver<Option<MediaControlRequestV3>>,309+ track: TrackProducer,310+ generation_id: u64,311+ signing_key: MediaGenerationSigningKey,312+ encoder_control: Option<EncoderControl>,313+ telemetry: Arc<MediaV3Telemetry>,314+ mut shutdown: tokio::sync::watch::Receiver<bool>,315+) -> Result<()> {316+ let maximum_replay_age = maximum_media_replay_age(config.framerate);317+ let mut replay_cursor = MediaReplayCursor::default();318+ let mut publisher = MoqGroupPublisher::authenticated(track, generation_id, signing_key);319+ let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));320+ let log_identity = iroh::SecretKey::from_bytes(&[1; 32]).public();321+322+ let result = async {323+ loop {324+ tokio::select! {325+ biased;326+ changed = shutdown.changed() => {327+ if changed.is_err() || *shutdown.borrow_and_update() {328+ return Ok(());329+ }330+ }331+ changed = control_requests.changed() => {332+ if changed.is_err() {333+ continue;334+ }335+ let Some(request) = *control_requests.borrow_and_update() else {336+ continue;337+ };338+ let through_sequence = current_gop_receiver339+ .borrow()340+ .as_ref()341+ .and_then(|gop| gop.frames.last())342+ .map(|frame| frame.sequence);343+ let cancelled_group = apply_moq_keyframe_request(344+ &mut publisher,345+ &mut replay_cursor,346+ through_sequence,347+ request.reason,348+ );349+ if cancelled_group.is_some() {350+ telemetry.scheduler_cancellations.fetch_add(1, Ordering::Relaxed);351+ }352+ let disposition = forced_idr.request(request.reason);353+ debug!(generation_id, request_id = request.request_id, ?disposition, ?cancelled_group, "shared generation accepted keyframe request");354+ }355+ acknowledgement = forced_idr.acknowledgements.join_next(),356+ if forced_idr.pending_revision.is_some() =>357+ {358+ forced_idr.complete(acknowledgement, log_identity, "shared-iroh-moq");359+ }360+ changed = current_gop_receiver.changed() => {361+ changed.context("shared encoded video source stopped")?;362+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {363+ continue;364+ };365+ publish_generation_gop(366+ &config,367+ &mut publisher,368+ &mut replay_cursor,369+ current_gop,370+ maximum_replay_age,371+ &telemetry,372+ )?;373+ }374+ }375+ }376+ }377+ .await;378+ forced_idr379+ .abort_and_drain(log_identity, "shared-iroh-moq")380+ .await;381+ publisher.abort();382+ result383+}384+385+fn publish_generation_gop(386+ config: &HostConfig,387+ publisher: &mut MoqGroupPublisher,388+ replay_cursor: &mut MediaReplayCursor,389+ current_gop: EncodedGop,390+ maximum_replay_age: Duration,391+ telemetry: &MediaV3Telemetry,392+) -> Result<()> {393+ let initial_replay_started_at = replay_cursor.last_sequence.is_none().then(Instant::now);394+ let replay_through_sequence = current_gop395+ .frames396+ .last()397+ .map(|frame| frame.sequence)398+ .context("shared current GOP snapshot is empty")?;399+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {400+ let replay_discontinuity = match replay_cursor.classify(401+ &frame,402+ replay_through_sequence,403+ initial_replay_started_at,404+ Instant::now(),405+ maximum_replay_age,406+ ) {407+ MediaReplayDecision::Send { discontinuity } => discontinuity,408+ MediaReplayDecision::SkipUntilKeyframe409+ | MediaReplayDecision::DiscardStaleSuffix { .. } => {410+ if publisher.request_keyframe().is_some() {411+ telemetry412+ .scheduler_cancellations413+ .fetch_add(1, Ordering::Relaxed);414+ }415+ replay_cursor.enter_resync_through(Some(replay_through_sequence));416+ break;417+ }418+ };419+ match publisher420+ .publish(config, &frame, replay_discontinuity)421+ .inspect_err(|_| {422+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);423+ })? {424+ MoqGroupDecision::Published {425+ cancelled_previous, ..426+ } => {427+ if cancelled_previous {428+ telemetry429+ .scheduler_cancellations430+ .fetch_add(1, Ordering::Relaxed);431+ }432+ replay_cursor.commit_sent(&frame);433+ }434+ MoqGroupDecision::SkipUntilKeyframe | MoqGroupDecision::EnterResync => {435+ replay_cursor.enter_resync_through(Some(replay_through_sequence));436+ break;437+ }438+ }439+ }440+ Ok(())441+}442+443+pub(super) async fn run_generation_audio_publisher(444+ mut packets: tokio::sync::mpsc::Receiver<crate::audio::EncodedAudioPacket>,445+ track: TrackProducer,446+ generation_id: u64,447+ signing_key: MediaGenerationSigningKey,448+ mut shutdown: tokio::sync::watch::Receiver<bool>,449+) -> Result<()> {450+ let mut publisher = MoqAudioPublisher::new(track, generation_id, signing_key);451+ let result = async {452+ loop {453+ tokio::select! {454+ changed = shutdown.changed() => {455+ if changed.is_err() || *shutdown.borrow_and_update() {456+ return Ok(());457+ }458+ }459+ packet = packets.recv() => {460+ let packet = packet.context("shared encoded Opus source stopped")?;461+ publisher.publish(packet)?;462+ }463+ }464+ }465+ }466+ .await;467+ publisher.abort();468+ result469+}470+198471 fn apply_moq_keyframe_request(199472 publisher: &mut MoqGroupPublisher,200473 replay_cursor: &mut MediaReplayCursor,@@ -419,8 +692,8 @@ pub(super) async fn serve_control_moq(
419692 420693 pub(super) async fn serve_control_moq_v2(421694 connection: Connection,422- config: HostConfig,423695 sessions: &Arc<SessionRegistry>,696+ generations: &Arc<MediaGenerationManager>,424697 authorization: &AuthorizationPolicy,425698 input_operations: &Arc<InputOperations>,426699 host_secret: [u8; 32],@@ -468,56 +741,36 @@ pub(super) async fn serve_control_moq_v2(
468741 }469742 };470743 471- let source = match config.source {472- VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),473- VideoSource::GamescopePipewire => {474- let primary = spawn_gamescope_pipewire_after_static_preflight(475- config.clone(),476- lease.session_clock,477- )478- .await?;479- select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await480- }481- };482- let EncodedSource {483- frames: frame_receiver,484- current_gop: mut current_gop_receiver,485- task: source_task,486- pointer_surface_dimensions,487- encoder_control,488- } = match source {489- Ok(source) => source,744+ let generation = match generations.acquire().await {745+ Ok(generation) => generation,490746 Err(error) => {491- send_rejection_v2(&mut send, "video source is unavailable").await?;747+ send_rejection_v2(&mut send, "shared media generation is unavailable").await?;492748 return Err(error);493749 }494750 };495- let source_task = SourceTaskGuard::new(source_task);496- sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;497-498- let mut generation_secret = [0_u8; 32];499- getrandom::fill(&mut generation_secret)500- .context("generating ephemeral media authentication key")?;501- let generation_signing_key = MediaGenerationSigningKey::from_bytes(&generation_secret);502- generation_secret.fill(0);503- let issued_at_unix = unix_timestamp_now()?;504- let certificate = generation_signing_key.certify(505- *iroh::SecretKey::from_bytes(&host_secret)506- .public()507- .as_bytes(),508- &host_secret,751+ let shared = Arc::clone(&generation.shared);752+ let initial_snapshot = sessions.bind_v2_generation(753+ remote,509754 lease.session_id,510- issued_at_unix,511- issued_at_unix.saturating_add(60 * 60),755+ shared.generation_id,756+ shared.session_clock,757+ Arc::clone(&shared.telemetry),758+ shared.encoder_control.clone(),512759 )?;760+ let issued_at_unix = unix_timestamp_now()?;513761 let mut subscription_nonce = [0_u8; 32];514762 getrandom::fill(&mut subscription_nonce).context("generating subscription capability nonce")?;763+ let subscription_tracks = if shared.audio_enabled {764+ SubscriptionTracks::ALL765+ } else {766+ SubscriptionTracks::VIDEO_H264767+ };515768 let subscription_capability = SignedSubscriptionCapability::issue(516769 SubscriptionClaims::new(517- certificate.claims.host_node_id,518- lease.session_id,770+ shared.certificate.claims.host_node_id,771+ shared.generation_id,519772 *remote.as_bytes(),520- SubscriptionTracks::VIDEO_H264,773+ subscription_tracks,521774 lease.authorization_revision,522775 issued_at_unix,523776 issued_at_unix.saturating_add(15 * 60),@@ -528,32 +781,27 @@ pub(super) async fn serve_control_moq_v2(
528781 )?;529782 let subscription_token = subscription_capability.encode();530783 531- let mut broadcast = Broadcast::new().produce();532- let track = broadcast533- .create_track(Track {534- name: MOQ_VIDEO_H264_TRACK.to_owned(),535- priority: MOQ_VIDEO_TRACK_PRIORITY,536- })537- .context("creating static MoQ H.264 track for control v2")?;538- let catalog = publish_goq_catalog_v2(&mut broadcast, lease.session_id, &certificate)?;539- let broadcast_name = media_moq_broadcast_name(lease.session_id)?;784+ let broadcast_name = shared.broadcast_name.clone();540785 let attachment = sessions.expect_moq_v2(541786 remote,542787 lease.session_id,543788 broadcast_name.clone(),544- broadcast.consume(),789+ shared.consumer()?,545790 subscription_capability,546791 )?;547792 548- let negotiated = MEDIA_CAPABILITIES793+ let mut supported = vec![Capability::VideoH264];794+ if shared.audio_enabled {795+ supported.push(Capability::AudioOpus);796+ }797+ let negotiated = supported549798 .iter()550799 .copied()551800 .filter(|capability| hello.capabilities.contains(capability))552801 .collect();553- let mut control_hello =554- HostHelloV2::accepted(lease.session_id, negotiated, lease.initial_snapshot.clone())555- .with_media_subscription_capability(subscription_token);556- if let Some(dimensions) = pointer_surface_dimensions {802+ let mut control_hello = HostHelloV2::accepted(lease.session_id, negotiated, initial_snapshot)803+ .with_media_subscription_capability(subscription_token);804+ if let Some(dimensions) = shared.pointer_surface_dimensions {557805 control_hello = control_hello.with_pointer_surface_dimensions(dimensions);558806 }559807 write_host_hello_v2(&mut send, &control_hello).await?;@@ -579,27 +827,18 @@ pub(super) async fn serve_control_moq_v2(
579827 .await580828 .context("timed out waiting for authorized MoQ v2 attachment")??;581829 582- let session_result = run_control_moq_session(830+ let session_result = run_generation_control_session(583831 &connection,584- &config,585- &mut current_gop_receiver,586- MoqControlReader::V2 {587- send,588- recv,589- sessions: Arc::clone(sessions),590- input_operations: Arc::clone(input_operations),591- session_id: lease.session_id,592- },832+ send,833+ recv,834+ Arc::clone(sessions),835+ Arc::clone(input_operations),593836 remote,837+ lease.session_id,594838 closed,595- track,596- &mut broadcast,597- encoder_control,598- Arc::clone(&lease.media_v3_telemetry),599- Some((lease.session_id, generation_signing_key)),839+ shared.keyframe_requests.clone(),600840 )601841 .await;602- let catalog_result = catalog.finish();603842 604843 if sessions.invalidate_v2_focus(605844 remote,@@ -608,14 +847,43 @@ pub(super) async fn serve_control_moq_v2(
608847 )? {609848 input_operations.reset()?;610849 }611- drop(current_gop_receiver);612- drop(frame_receiver);613- source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;614850 drop(lease);851+ generation.release().await;615852 info!(%remote, "MoQ control v2 client released");616- match session_result {617- Err(error) => Err(error),618- Ok(()) => catalog_result,853+ session_result854+}855+856+857+async fn run_generation_control_session(858+ connection: &Connection,859+ send: iroh::endpoint::SendStream,860+ recv: iroh::endpoint::RecvStream,861+ sessions: Arc<SessionRegistry>,862+ input_operations: Arc<InputOperations>,863+ remote: EndpointId,864+ session_id: u64,865+ mut moq_closed: tokio::sync::oneshot::Receiver<()>,866+ keyframes: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,867+) -> Result<()> {868+ let mut control = tokio::spawn(forward_control_v2_requests(869+ send,870+ recv,871+ sessions,872+ input_operations,873+ remote,874+ session_id,875+ keyframes,876+ ));877+ tokio::select! {878+ result = &mut control => result.context("shared-generation control task failed")?,879+ reason = connection.closed() => {880+ debug!(%remote, ?reason, "shared-generation control connection closed");881+ Ok(())882+ }883+ result = &mut moq_closed => {884+ debug!(%remote, ?result, "shared-generation MoQ attachment closed");885+ Ok(())886+ }619887 }620888 }621889 @@ -11,7 +11,7 @@ use sigil_protocol::{
1111 ControllerSlot, FocusCommandActionV2, FocusCommandV2, FocusStateV2, FocusTransitionReasonV2,1212 InvitationGrants, KeyframeRequestReasonV3, MediaGenerationDescriptorV2, SessionSnapshotV2,1313 SignedSubscriptionCapability, SubscriptionTracks, ViewerPresenceId, ViewerPresenceV2,14- media_moq_broadcast_name,14+ media_generation_moq_broadcast_name, media_moq_broadcast_name,1515 };1616 use tracing::{debug, warn};1717 @@ -77,6 +77,8 @@ struct ActiveSession {
7777 session_id: u64,7878 nonce: [u8; 16],7979 session_clock: SessionClock,80+ media_generation_id: u64,81+ media_broadcast_name: String,8082 grants: InvitationGrants,8183 viewer_handle: Option<String>,8284 authorization_revision: u64,@@ -400,6 +402,8 @@ impl SessionRegistry {
400402 session_id,401403 nonce,402404 session_clock,405+ media_generation_id: session_id,406+ media_broadcast_name: media_moq_broadcast_name(session_id)?,403407 grants,404408 viewer_handle: None,405409 authorization_revision: 1,@@ -483,6 +487,8 @@ impl SessionRegistry {
483487 session_id,484488 nonce,485489 session_clock,490+ media_generation_id: session_id,491+ media_broadcast_name: media_moq_broadcast_name(session_id)?,486492 grants: authorized.grants,487493 viewer_handle: Some(authorized.handle),488494 authorization_revision: authorized.authorization_revision,@@ -514,6 +520,42 @@ impl SessionRegistry {
514520 })515521 }516522 523+ pub(super) fn bind_v2_generation(524+ &self,525+ remote: EndpointId,526+ session_id: u64,527+ generation_id: u64,528+ session_clock: SessionClock,529+ telemetry: Arc<MediaV3Telemetry>,530+ encoder_control: Option<EncoderControl>,531+ ) -> Result<SessionSnapshotV2> {532+ let mut active = self.active.lock().expect("session registry poisoned");533+ let session = active534+ .as_mut()535+ .filter(|session| {536+ session.mode == SessionMode::V2Single537+ && session.media_active538+ && session.remote == remote539+ && session.session_id == session_id540+ })541+ .context("generation binding does not match the active v2 viewer")?;542+ ensure!(generation_id != 0, "media generation id must be non-zero");543+ session.session_clock = session_clock;544+ session.media_generation_id = generation_id;545+ session.media_broadcast_name = media_generation_moq_broadcast_name(generation_id)?;546+ session.media_v3_telemetry = telemetry;547+ session.encoder_control = encoder_control;548+ let v2_state = self.v2_state.lock().expect("v2 session state poisoned");549+ let snapshot = v2_snapshot_for(550+ session,551+ v2_state552+ .as_ref()553+ .context("active v2 viewer is missing snapshot state")?,554+ )?;555+ self.v2_snapshot.send_replace(Some(snapshot.clone()));556+ Ok(snapshot)557+ }558+517559 pub(super) fn claim_input(518560 self: &Arc<Self>,519561 remote: EndpointId,@@ -1019,7 +1061,10 @@ impl SessionRegistry {
10191061 let session = active10201062 .as_mut()10211063 .filter(|session| {1022- session.media_active && session.remote == remote && session.nonce == nonce1064+ session.mode == SessionMode::LegacyExclusive1065+ && session.media_active1066+ && session.remote == remote1067+ && session.nonce == nonce10231068 })10241069 .context("audio connection does not match the active media session")?;10251070 ensure!(@@ -1094,7 +1139,7 @@ impl SessionRegistry {
10941139 })10951140 .context("authenticated MoQ expectation does not match the active v2 session")?;10961141 ensure!(1097- subscription_capability.claims.media_generation_id == session_id1142+ subscription_capability.claims.media_generation_id == session.media_generation_id10981143 && subscription_capability.claims.subscriber_endpoint_id == *remote.as_bytes()10991144 && subscription_capability11001145 .claims@@ -1157,9 +1202,9 @@ impl SessionRegistry {
11571202 capability11581203 .verify_binding(11591204 capability.claims.host_node_id,1160- session.session_id,1205+ session.media_generation_id,11611206 *remote.as_bytes(),1162- SubscriptionTracks::VIDEO_H264,1207+ capability.claims.tracks,11631208 session.authorization_revision,11641209 now_unix,11651210 )@@ -1280,8 +1325,8 @@ fn v2_snapshot_for(session: &ActiveSession, v2: &V2SessionState) -> Result<Sessi
12801325 focus: v2.focus.clone(),12811326 transition_reason: v2.transition_reason,12821327 media: MediaGenerationDescriptorV2 {1283- generation_id: session.session_id,1284- broadcast_name: media_moq_broadcast_name(session.session_id)?,1328+ generation_id: session.media_generation_id,1329+ broadcast_name: session.media_broadcast_name.clone(),12851330 },12861331 };12871332 snapshot.validate()?;@@ -75,11 +75,12 @@ pub use media_v3::{
7575 write_media_control_request_v3, write_media_object_v3,7676 };7777 pub use moq_catalog::{78- GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES,79- MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1, MOQ_CATALOG_EXTENSION_VERSION_V1,80- MOQ_CATALOG_EXTENSION_VERSION_V2, MOQ_GOP_GROUP_FORMAT_V1, MOQ_MEDIA_OBJECT_FORMAT_V1,81- MOQ_VIDEO_TRACK_PRIORITY, MoqCatalogExtensionV1, MoqCatalogExtensionV2,82- MoqTrackAuthenticationV2, MoqTrackDescriptorV1, MoqVideoCatalogV1, MoqVideoCatalogV2,78+ GoqCatalogDocument, GoqCatalogDocumentV2, MAX_MOQ_CATALOG_BYTES, MOQ_AUDIO_GROUP_FORMAT_V1,79+ MOQ_AUDIO_OBJECT_FORMAT_V1, MOQ_AUDIO_TRACK_PRIORITY, MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1,80+ MOQ_CATALOG_EXTENSION_VERSION_V1, MOQ_CATALOG_EXTENSION_VERSION_V2, MOQ_GOP_GROUP_FORMAT_V1,81+ MOQ_MEDIA_OBJECT_FORMAT_V1, MOQ_VIDEO_TRACK_PRIORITY, MoqAudioCatalogV2, MoqCatalogExtensionV1,82+ MoqCatalogExtensionV2, MoqTrackAuthenticationV2, MoqTrackDescriptorV1, MoqVideoCatalogV1,83+ MoqVideoCatalogV2,8384 };8485 pub use subscription::{8586 MAX_SUBSCRIPTION_CAPABILITY_TOKEN_LEN, MAX_SUBSCRIPTION_TTL_SECS,@@ -108,6 +109,8 @@ pub const CONTROL_ALPN_V1: &[u8] = b"sigil/control/1";
108109 pub const CONTROL_ALPN_V2: &[u8] = b"sigil/control/2";109110 /// Static upstream MoQ track carrying bounded H.264 access-unit objects.110111 pub const MOQ_VIDEO_H264_TRACK: &str = "video/h264";112+/// Static upstream MoQ track carrying authenticated 20 ms Opus packets.113+pub const MOQ_AUDIO_OPUS_TRACK: &str = "audio/opus";111114 /// ALPN for the v1 low-latency Opus datagram connection.112115 pub const AUDIO_ALPN_V1: &[u8] = b"sigil/audio/1";113116 @@ -142,6 +145,17 @@ pub fn media_moq_broadcast_name(session_id: u64) -> Result<String> {
142145 Ok(format!("sigil/session/{session_id}/video"))143146 }144147 148+/// Derive the host-generation-scoped MoQ broadcast name used by control v2.149+pub fn media_generation_moq_broadcast_name(generation_id: u64) -> Result<String> {150+ if generation_id == 0 {151+ return Err(ProtocolError::InvalidMessage {152+ message_type: "MoQ generation broadcast name",153+ reason: "generation id must be non-zero",154+ });155+ }156+ Ok(format!("sigil/generation/{generation_id}/media"))157+}158+145159 146160 mod tests {147161 use super::*;@@ -160,10 +174,16 @@ mod tests {
160174 161175 fn moq_media_namespace_is_session_scoped_and_stable() {162176 assert_eq!(MOQ_VIDEO_H264_TRACK, "video/h264");177+ assert_eq!(MOQ_AUDIO_OPUS_TRACK, "audio/opus");163178 assert_eq!(164179 media_moq_broadcast_name(42).unwrap(),165180 "sigil/session/42/video"166181 );167182 assert!(media_moq_broadcast_name(0).is_err());183+ assert_eq!(184+ media_generation_moq_broadcast_name(42).unwrap(),185+ "sigil/generation/42/media"186+ );187+ assert!(media_generation_moq_broadcast_name(0).is_err());168188 }169189 }@@ -319,6 +319,12 @@ impl SignedMediaGenerationCertificate {
319319 320320 pub struct MediaGenerationSigningKey(SigningKey);321321 322+impl Clone for MediaGenerationSigningKey {323+ fn clone(&self) -> Self {324+ Self(self.0.clone())325+ }326+}327+322328 impl std::fmt::Debug for MediaGenerationSigningKey {323329 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {324330 formatter.write_str("MediaGenerationSigningKey([REDACTED])")@@ -1,8 +1,8 @@
11 use serde::{Deserialize, Serialize};22 33 use crate::{4- MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX, MOQ_VIDEO_H264_TRACK, MediaAuthMode, ProtocolError,5- Result, SignedMediaGenerationCertificate,4+ MEDIA_GENERATION_CERTIFICATE_TOKEN_PREFIX, MOQ_AUDIO_OPUS_TRACK, MOQ_VIDEO_H264_TRACK,5+ MediaAuthMode, ProtocolError, Result, SignedMediaGenerationCertificate,66 };77 88 pub const MOQ_CATALOG_EXTENSION_VERSION_V1: u16 = 1;@@ -12,7 +12,10 @@ pub const MAX_MOQ_CATALOG_BYTES: usize = 4 * 1024;
1212 pub const MOQ_MEDIA_OBJECT_FORMAT_V1: &str = "sigil/media-frame/1";1313 pub const MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1: &str = "sigil/authenticated-media-object/1";1414 pub const MOQ_GOP_GROUP_FORMAT_V1: &str = "sigil/moq-gop/1";15+pub const MOQ_AUDIO_OBJECT_FORMAT_V1: &str = "sigil/audio-packet/1";16+pub const MOQ_AUDIO_GROUP_FORMAT_V1: &str = "sigil/moq-audio-group/1";1517 pub const MOQ_VIDEO_TRACK_PRIORITY: u8 = u8::MAX;18+pub const MOQ_AUDIO_TRACK_PRIORITY: u8 = u8::MAX - 1;1619 1720 1821 @@ -112,12 +115,27 @@ pub struct MoqVideoCatalogV2 {
112115 pub authentication: MoqTrackAuthenticationV2,113116 }114117 118+119+120+pub struct MoqAudioCatalogV2 {121+ pub track: MoqTrackDescriptorV1,122+ pub codec: String,123+ pub sample_rate: u32,124+ pub channels: u8,125+ pub frame_samples: u16,126+ pub payload_format: String,127+ pub group_format: String,128+ pub authentication: MoqTrackAuthenticationV2,129+}130+115131 116132 117133 pub struct MoqCatalogExtensionV2 {118134 pub version: u16,119135 pub generation_id: u64,120136 pub video: MoqVideoCatalogV2,137+ 138+ pub audio: Option<MoqAudioCatalogV2>,121139 }122140 123141 impl MoqCatalogExtensionV2 {@@ -142,11 +160,38 @@ impl MoqCatalogExtensionV2 {
142160 generation_certificate: certificate.encode(),143161 },144162 },163+ audio: None,145164 };146165 extension.validate()?;147166 Ok(extension)148167 }149168 169+ pub fn video_h264_opus(170+ generation_id: u64,171+ certificate: &SignedMediaGenerationCertificate,172+ ) -> Result<Self> {173+ let mut extension = Self::video_h264(generation_id, certificate)?;174+ extension.audio = Some(MoqAudioCatalogV2 {175+ track: MoqTrackDescriptorV1 {176+ name: MOQ_AUDIO_OPUS_TRACK.to_owned(),177+ priority: MOQ_AUDIO_TRACK_PRIORITY,178+ },179+ codec: "opus".to_owned(),180+ sample_rate: crate::OPUS_SAMPLE_RATE,181+ channels: crate::OPUS_CHANNELS,182+ frame_samples: crate::OPUS_FRAME_SAMPLES,183+ payload_format: MOQ_AUDIO_OBJECT_FORMAT_V1.to_owned(),184+ group_format: MOQ_AUDIO_GROUP_FORMAT_V1.to_owned(),185+ authentication: MoqTrackAuthenticationV2 {186+ mode: MediaAuthMode::Ed25519.label().to_owned(),187+ object_format: MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1.to_owned(),188+ generation_certificate: certificate.encode(),189+ },190+ });191+ extension.validate()?;192+ Ok(extension)193+ }194+150195 pub fn validate(&self) -> Result<SignedMediaGenerationCertificate> {151196 if self.version != MOQ_CATALOG_EXTENSION_VERSION_V2 || self.generation_id == 0 {152197 return invalid_v2("unsupported extension version or zero generation");@@ -169,6 +214,22 @@ impl MoqCatalogExtensionV2 {
169214 {170215 return invalid_v2("unknown or malformed media authentication contract");171216 }217+ if let Some(audio) = &self.audio218+ && (audio.track.name != MOQ_AUDIO_OPUS_TRACK219+ || audio.track.priority != MOQ_AUDIO_TRACK_PRIORITY220+ || audio.codec != "opus"221+ || audio.sample_rate != crate::OPUS_SAMPLE_RATE222+ || audio.channels != crate::OPUS_CHANNELS223+ || audio.frame_samples != crate::OPUS_FRAME_SAMPLES224+ || audio.payload_format != MOQ_AUDIO_OBJECT_FORMAT_V1225+ || audio.group_format != MOQ_AUDIO_GROUP_FORMAT_V1226+ || audio.authentication.mode != MediaAuthMode::Ed25519.label()227+ || audio.authentication.object_format != MOQ_AUTHENTICATED_MEDIA_OBJECT_FORMAT_V1228+ || audio.authentication.generation_certificate229+ != self.video.authentication.generation_certificate)230+ {231+ return invalid_v2("audio track contract does not match authenticated Opus");232+ }172233 let certificate = SignedMediaGenerationCertificate::decode(173234 &self.video.authentication.generation_certificate,174235 )?;@@ -238,6 +299,16 @@ impl GoqCatalogDocumentV2 {
238299 })239300 }240301 302+ pub fn video_h264_opus(303+ generation_id: u64,304+ certificate: &SignedMediaGenerationCertificate,305+ ) -> Result<Self> {306+ Ok(Self {307+ media: hang::Catalog::default(),308+ goq: MoqCatalogExtensionV2::video_h264_opus(generation_id, certificate)?,309+ })310+ }311+241312 pub fn validate(&self) -> Result<SignedMediaGenerationCertificate> {242313 if self.media != hang::Catalog::default() {243314 return Err(ProtocolError::InvalidMessage {@@ -363,4 +434,17 @@ mod tests {
363434 value["goq"]["video"]["authentication"]["key"] = serde_json::json!("secret");364435 assert!(serde_json::from_value::<GoqCatalogDocumentV2>(value).is_err());365436 }437+438+ 439+ fn authenticated_v2_catalog_advertises_opus_on_the_shared_clock() {440+ let base = v2_document();441+ let certificate = base.validate().unwrap();442+ let document = GoqCatalogDocumentV2::video_h264_opus(42, &certificate).unwrap();443+ let audio = document.goq.audio.as_ref().unwrap();444+ assert_eq!(audio.track.name, MOQ_AUDIO_OPUS_TRACK);445+ assert_eq!(audio.sample_rate, crate::OPUS_SAMPLE_RATE);446+ assert_eq!(audio.channels, crate::OPUS_CHANNELS);447+ assert_eq!(audio.frame_samples, crate::OPUS_FRAME_SAMPLES);448+ assert_eq!(document.validate().unwrap(), certificate);449+ }366450 }@@ -25,6 +25,7 @@ relay_only=false
2525 expected_path_mode=direct2626 control_v2=false2727 viewers=128+assert_shared_generation=false2829 2930 usage() {3031 cat <<'EOF'@@ -47,6 +48,8 @@ Options:
4748 control, media, and input paths to stay relayed4849 --control-v2 Exercise explicit control/2 and input/2 focus4950 --viewers COUNT Viewer count for control-v2 (Phase 1 requires 1)51+ --assert-shared-generation Require host-scoped generation naming, one52+ producer start, and final-viewer cleanup5053 --help Show this help5154 EOF5255 }@@ -121,6 +124,10 @@ while [[ $# -gt 0 ]]; do
121124 viewers="$2"122125 shift 2123126 ;;127+ --assert-shared-generation)128+ assert_shared_generation=true129+ shift130+ ;;124131 --help|-h)125132 usage126133 exit 0@@ -541,6 +548,14 @@ if [[ "$control_v2" == true ]]; then
541548 grep -Fxq 'input_alpn=sigil/input/2' "$v2_log" || die "input-v2 ALPN evidence is missing"542549 grep -Fxq 'slot_0_input=ok' "$v2_log" || die "slot-0 input evidence is missing"543550 grep -Fxq 'focus_release=ok' "$v2_log" || die "focus release evidence is missing"551+ if [[ "$assert_shared_generation" == true ]]; then552+ grep -Fxq 'media_generation_scope=host' "$v2_log" \553+ || die "host-scoped media generation evidence is missing"554+ grep -Eq '^media_generation_id=[1-9][0-9]*$' "$v2_log" \555+ || die "media generation ID evidence is missing"556+ grep -Eq '^media_broadcast_name=sigil/generation/[1-9][0-9]*/media$' "$v2_log" \557+ || die "generation-scoped broadcast evidence is missing"558+ fi544559 awk -F= '$1 == "initial_snapshot_revision" && $2 ~ /^[1-9][0-9]*$/ { found=1 } END { exit !found }' \545560 "$v2_log" || die "initial snapshot evidence is missing"546561 awk -F= '$1 == "focus_generation" && $2 ~ /^[1-9][0-9]*$/ { found=1 } END { exit !found }' \@@ -549,6 +564,21 @@ if [[ "$control_v2" == true ]]; then
549564 || die "control-v2 session was not released"550565 wait_for_log_count "$host_log" 'input v2 client released' 1 "$host_pid" 'sigil' \551566 || die "input-v2 session was not neutralized and released"567+ if [[ "$assert_shared_generation" == true ]]; then568+ clean_host_log="$tmp_root/host-shared-generation.log"569+ sed $'s/\033\[[0-9;]*m//g' "$host_log" >"$clean_host_log"570+ wait_for_log_count "$host_log" 'shared media generation stopped with all sources cleaned up' \571+ 1 "$host_pid" 'sigil' || die "shared generation did not clean up after the final viewer"572+ [[ "$(grep -Fc -- 'shared media generation started' "$host_log" || true)" -eq 1 ]] \573+ || die "control-v2 started more than one shared media generation"574+ [[ "$(grep -Fc -- 'shared media generation stopped with all sources cleaned up' "$host_log" || true)" -eq 1 ]] \575+ || die "control-v2 stopped the shared media generation more than once"576+ for resource_field in \577+ 'video_sources=1' 'video_encoders=1' 'audio_sources=0' 'audio_encoders=0' 'publishers=1'; do578+ grep -F 'shared media generation started' "$clean_host_log" | grep -Fq -- "$resource_field" \579+ || die "shared generation resource-count evidence is missing $resource_field"580+ done581+ fi552582 553583 # The unchanged v1 path remains explicitly usable after a v2 generation.554584 if ! run_bounded "$command_timeout_seconds" "$v1_log" \@@ -574,6 +604,7 @@ if [[ "$control_v2" == true ]]; then
574604 printf 'input_v2=ok\n'575605 printf 'neutralization=reset-before-input-lease-release\n'576606 printf 'v1_compatibility=ok\n'607+ printf 'shared_generation=%s\n' "$([[ "$assert_shared_generation" == true ]] && printf ok || printf not-requested)"577608 exit 0578609 fi579610 @@ -12,6 +12,7 @@ use tauri::{
1212 1313 use super::transport::negotiate_v1;1414 use crate::commands::state::{AUDIO_DELIVERY_CAPACITY, AppState, AudioDeliveryState};15+use crate::media::moq_receiver::MoqAudioReceiver;1516 1617 // Three 20 ms Opus packets cap the Rust→webview handoff at 60 ms. The1718 // AudioWorklet owns a separate fixed ring and never feeds back into transport.@@ -368,6 +369,184 @@ pub(crate) async fn try_start_audio(
368369 Ok(audio_connection)369370 }370371 372+pub(crate) struct MoqAudioStartRequest {373+ pub(crate) audio_supported: bool,374+ pub(crate) audio_channel: Channel<Response>,375+ pub(crate) audio_deliveries: Arc<StdMutex<AudioDeliveryState>>,376+ pub(crate) connection_generation: Arc<AtomicU64>,377+ pub(crate) generation: u64,378+}379+380+pub(crate) fn try_start_moq_audio(381+ app: AppHandle,382+ mut receiver: MoqAudioReceiver,383+ request: MoqAudioStartRequest,384+) -> Result<(), String> {385+ if !request.audio_supported {386+ return Err("WebCodecs Opus AudioDecoder is unavailable".to_string());387+ }388+ tokio::spawn(async move {389+ let mut reorder = AudioReorderBuffer::default();390+ let mut transport_received_total = 0_u64;391+ let mut sequence_dropped_total = 0_u64;392+ let mut frontend_dropped_total = 0_u64;393+ let mut frontend_sent_total = 0_u64;394+ let mut pending_discontinuity = false;395+ let mut last_stats = Instant::now();396+397+ loop {398+ let packet = match receiver.next().await {399+ Ok(Some(packet)) => packet,400+ Ok(None) => {401+ emit_audio_event_if_current(402+ &app,403+ "audio-state",404+ &request.connection_generation,405+ &request.audio_deliveries,406+ request.generation,407+ |_| {408+ serde_json::json!({409+ "generation": request.generation,410+ "available": false,411+ "error": "Authenticated MoQ audio track ended"412+ })413+ },414+ );415+ break;416+ }417+ Err(error) => {418+ emit_audio_event_if_current(419+ &app,420+ "audio-state",421+ &request.connection_generation,422+ &request.audio_deliveries,423+ request.generation,424+ |_| {425+ serde_json::json!({426+ "generation": request.generation,427+ "available": false,428+ "error": error429+ })430+ },431+ );432+ break;433+ }434+ };435+ if !audio_generation_is_current(&request.connection_generation, request.generation) {436+ break;437+ }438+ transport_received_total = transport_received_total.saturating_add(1);439+ let (packets, dropped) = match reorder.insert(packet) {440+ Ok(result) => result,441+ Err(error) => {442+ emit_audio_event_if_current(443+ &app,444+ "audio-state",445+ &request.connection_generation,446+ &request.audio_deliveries,447+ request.generation,448+ |_| {449+ serde_json::json!({450+ "generation": request.generation,451+ "available": false,452+ "error": error453+ })454+ },455+ );456+ break;457+ }458+ };459+ sequence_dropped_total = sequence_dropped_total.saturating_add(dropped);460+ pending_discontinuity |= dropped > 0;461+ for ordered in packets {462+ let delivery_id = match lock_audio_deliveries(&request.audio_deliveries)463+ .reserve(request.generation)464+ {465+ Ok(Some(delivery_id)) => delivery_id,466+ Ok(None) => {467+ frontend_dropped_total = frontend_dropped_total.saturating_add(1);468+ pending_discontinuity = true;469+ continue;470+ }471+ Err(_) => return,472+ };473+ let envelope = match encode_audio_channel_packet(474+ request.generation,475+ delivery_id,476+ ordered.packet,477+ ordered.discontinuity || pending_discontinuity,478+ ) {479+ Ok(envelope) => envelope,480+ Err(error) => {481+ lock_audio_deliveries(&request.audio_deliveries)482+ .release_failed_delivery(request.generation, delivery_id);483+ emit_audio_event_if_current(484+ &app,485+ "audio-state",486+ &request.connection_generation,487+ &request.audio_deliveries,488+ request.generation,489+ |_| {490+ serde_json::json!({491+ "generation": request.generation,492+ "available": false,493+ "error": error494+ })495+ },496+ );497+ return;498+ }499+ };500+ if request.audio_channel.send(Response::new(envelope)).is_err() {501+ lock_audio_deliveries(&request.audio_deliveries)502+ .release_failed_delivery(request.generation, delivery_id);503+ emit_audio_event_if_current(504+ &app,505+ "audio-state",506+ &request.connection_generation,507+ &request.audio_deliveries,508+ request.generation,509+ |_| {510+ serde_json::json!({511+ "generation": request.generation,512+ "available": false,513+ "error": "Audio webview channel closed"514+ })515+ },516+ );517+ return;518+ }519+ frontend_sent_total = frontend_sent_total.saturating_add(1);520+ pending_discontinuity = false;521+ }522+ if last_stats.elapsed() >= Duration::from_millis(250) {523+ if !emit_audio_event_if_current(524+ &app,525+ "audio-stats",526+ &request.connection_generation,527+ &request.audio_deliveries,528+ request.generation,529+ |deliveries| {530+ serde_json::json!({531+ "generation": request.generation,532+ "transport_received_total": transport_received_total,533+ "sequence_dropped_total": sequence_dropped_total,534+ "frontend_dropped_total": frontend_dropped_total,535+ "frontend_sent_total": frontend_sent_total,536+ "frontend_queue_depth": deliveries.depth(request.generation).unwrap_or(0),537+ "frontend_queue_capacity": AUDIO_DELIVERY_CAPACITY,538+ })539+ },540+ ) {541+ return;542+ }543+ last_stats = Instant::now();544+ }545+ }546+ });547+ Ok(())548+}549+371550 pub(crate) fn iroh_client_ack_audio(372551 state: State<'_, AppState>,373552 generation: u64,@@ -20,7 +20,8 @@ use crate::media::adaptive_feedback::{
2020 run_media_feedback_session,2121 };2222 use crate::media::audio_delivery::{23- AudioStartRequest, lock_audio_deliveries, next_audio_generation, try_start_audio,23+ AudioStartRequest, MoqAudioStartRequest, lock_audio_deliveries, next_audio_generation,24+ try_start_audio, try_start_moq_audio,2425 };2526 use crate::media::frame_channel::{close_generation_connection, next_media_generation};2627 use crate::media::input_delivery::{@@ -218,6 +219,9 @@ pub(crate) async fn connect_client(
218219 NegotiatedMediaStream::V2(negotiation) => Some(negotiation.initial_snapshot.clone()),219220 NegotiatedMediaStream::V1(_) => None,220221 };222+ let advertised_media = initial_session_snapshot223+ .as_ref()224+ .map(|snapshot| snapshot.media.clone());221225 let moq_authentication = match &media_negotiation {222226 NegotiatedMediaStream::V2(negotiation) => Some(MoqAuthenticationExpectation {223227 expected_host: *host_node_id.as_bytes(),@@ -240,13 +244,17 @@ pub(crate) async fn connect_client(
240244 }241245 let media_verification_failures =242246 lock_network_diagnostics(&network_diagnostics).media_verification_failure_counter();243- let (upstream_moq_media, frame_connection_for_stats) = if media_transport247+ let (upstream_moq_media, upstream_moq_audio, frame_connection_for_stats) = if media_transport244248 == MediaTransport::UpstreamMoq245249 {246- let (receiver, diagnostics_connection) = match open_upstream_moq_media(250+ let (receiver, audio_receiver, diagnostics_connection) = match open_upstream_moq_media(247251 &endpoint,248252 &addr,249253 media_session_id,254+ advertised_media.as_ref().map(|media| media.generation_id),255+ advertised_media256+ .as_ref()257+ .map(|media| media.broadcast_name.as_str()),250258 moq_authentication,251259 media_verification_failures,252260 )@@ -263,9 +271,9 @@ pub(crate) async fn connect_client(
263271 return Err(error);264272 }265273 };266- (Some(receiver), diagnostics_connection)274+ (Some(receiver), audio_receiver, diagnostics_connection)267275 } else {268- (None, frame_conn.clone())276+ (None, None, frame_conn.clone())269277 };270278 if !development_mode && let Some(expected_invitation) = invitation.as_deref() {271279 // An accepted media hello means Sigil durably committed the one-time@@ -336,26 +344,49 @@ pub(crate) async fn connect_client(
336344 337345 let audio_generation = next_audio_generation(&state.audio_connection_generation)?;338346 lock_audio_deliveries(&state.audio_deliveries).begin_generation(audio_generation)?;339- let audio_result = try_start_audio(340- app.clone(),341- &endpoint,342- AudioStartRequest {343- address: addr.clone(),344- handshake_nonce,345- media_session_id,346- audio_supported,347- audio_channel,348- audio_deliveries: Arc::clone(&state.audio_deliveries),349- connection_generation: Arc::clone(&state.audio_connection_generation),350- generation: audio_generation,351- },352- )353- .await;347+ let audio_result = if control_protocol == ControlProtocol::V2 {348+ let receiver = upstream_moq_audio349+ .ok_or_else(|| "Host did not advertise an authenticated MoQ Opus track".to_string());350+ match receiver {351+ Ok(receiver) => try_start_moq_audio(352+ app.clone(),353+ receiver,354+ MoqAudioStartRequest {355+ audio_supported,356+ audio_channel,357+ audio_deliveries: Arc::clone(&state.audio_deliveries),358+ connection_generation: Arc::clone(&state.audio_connection_generation),359+ generation: audio_generation,360+ },361+ )362+ .map(|()| None),363+ Err(error) => Err(error),364+ }365+ } else {366+ try_start_audio(367+ app.clone(),368+ &endpoint,369+ AudioStartRequest {370+ address: addr.clone(),371+ handshake_nonce,372+ media_session_id,373+ audio_supported,374+ audio_channel,375+ audio_deliveries: Arc::clone(&state.audio_deliveries),376+ connection_generation: Arc::clone(&state.audio_connection_generation),377+ generation: audio_generation,378+ },379+ )380+ .await381+ .map(Some)382+ };354383 let mut audio_connection_for_stats = None;355384 let (audio_available, connected_audio_generation, audio_error) = match audio_result {356385 Ok(connection) => {357- audio_connection_for_stats = Some(connection.clone());358- *state.audio_connection.lock().await = Some((audio_generation, connection));386+ if let Some(connection) = connection {387+ audio_connection_for_stats = Some(connection.clone());388+ *state.audio_connection.lock().await = Some((audio_generation, connection));389+ }359390 (true, Some(audio_generation), None)360391 }361392 Err(error) => {@@ -25,6 +25,7 @@ impl MoqCatalogMode {
2525 2626 pub(crate) struct MoqCatalogSelection {2727 pub(crate) track: TrackConsumer,28+ pub(crate) audio_track: Option<TrackConsumer>,2829 pub(crate) mode: MoqCatalogMode,2930 pub(crate) certificate: Option<SignedMediaGenerationCertificate>,3031 }@@ -44,6 +45,7 @@ fn subscribe_static_video_track(
4445 })?;4546 Ok(MoqCatalogSelection {4647 track,48+ audio_track: None,4749 mode: MoqCatalogMode::AbsentStaticTrackCompat,4850 certificate: None,4951 })@@ -83,6 +85,7 @@ pub(crate) async fn subscribe_goq_video_track(
8385 })?;8486 Ok(MoqCatalogSelection {8587 track,88+ audio_track: None,8689 mode: MoqCatalogMode::GoqV1,8790 certificate: None,8891 })@@ -112,8 +115,21 @@ pub(crate) async fn subscribe_authenticated_goq_video_track(
112115 .map_err(|error| {113116 format!("Failed to subscribe to authenticated Goq video track: {error}")114117 })?;118+ let audio_track = document119+ .goq120+ .audio121+ .as_ref()122+ .map(|audio| {123+ broadcast124+ .subscribe_track(&Track::new(audio.track.name.clone()))125+ .map_err(|error| {126+ format!("Failed to subscribe to authenticated Goq audio track: {error}")127+ })128+ })129+ .transpose()?;115130 Ok(MoqCatalogSelection {116131 track,132+ audio_track,117133 mode: MoqCatalogMode::GoqV2Authenticated,118134 certificate: Some(certificate),119135 })@@ -176,7 +192,7 @@ mod tests {
176192 )177193 .unwrap();178194 (179- GoqCatalogDocumentV2::video_h264(42, &certificate).unwrap(),195+ GoqCatalogDocumentV2::video_h264_opus(42, &certificate).unwrap(),180196 host.verifying_key().to_bytes(),181197 )182198 }@@ -227,6 +243,9 @@ mod tests {
227243 .create_track(Track::new(sigil_protocol::MOQ_VIDEO_H264_TRACK))228244 .unwrap();229245 let (document, host) = v2_document();246+ let _audio = broadcast247+ .create_track(Track::new(sigil_protocol::MOQ_AUDIO_OPUS_TRACK))248+ .unwrap();230249 let _catalog = publish_catalog_v2(&mut broadcast, &document);231250 let selection = subscribe_authenticated_goq_video_track(232251 &broadcast.consume(),@@ -238,6 +257,7 @@ mod tests {
238257 .await239258 .unwrap();240259 assert_eq!(selection.mode, MoqCatalogMode::GoqV2Authenticated);260+ assert!(selection.audio_track.is_some());241261 assert_eq!(selection.certificate.unwrap().claims.generation_id, 42);242262 243263 let mut wrong_host_broadcast = Broadcast::new().produce();@@ -10,11 +10,11 @@ use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};
1010 1111 use sigil_protocol::MOQ_VIDEO_H264_TRACK;1212 use sigil_protocol::{13- AuthenticatedMediaObject, FrameFlags, KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3,14- MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS, MAX_MEDIA_OBJECT_ID_V3, MediaCodec, MediaFrame,15- MediaObjectCoordinates, MediaTrack, SignedMediaGenerationCertificate,16- SignedSubscriptionCapability, SubscriptionTracks, decode_media_frame_object,17- media_moq_broadcast_name,13+ AudioFlags, AudioPacket, AudioPacketHeader, AuthenticatedMediaObject, FrameFlags,14+ KeyframeRequestReasonV3, MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,15+ MAX_MEDIA_OBJECT_ID_V3, MediaCodec, MediaFrame, MediaObjectCoordinates, MediaTrack,16+ SignedMediaGenerationCertificate, SignedSubscriptionCapability, SubscriptionTracks,17+ decode_media_frame_object, media_moq_broadcast_name,1818 };1919 use tauri::{AppHandle, Manager};2020 @@ -163,6 +163,168 @@ pub(crate) struct MoqMediaReceiver {
163163 verification_failures: Option<Arc<AtomicU64>>,164164 }165165 166+struct MoqAudioGroupCursor {167+ group: GroupConsumer,168+ sequence: u64,169+ object_count: usize,170+ object_bytes: usize,171+}172+173+pub(crate) struct MoqAudioReceiver {174+ track: TrackConsumer,175+ current_group: Option<MoqAudioGroupCursor>,176+ last_group_sequence: Option<u64>,177+ last_packet_sequence: Option<u64>,178+ discontinuity_pending: bool,179+ certificate: SignedMediaGenerationCertificate,180+ verification_failures: Arc<AtomicU64>,181+}182+183+impl MoqAudioReceiver {184+ fn new(185+ track: TrackConsumer,186+ certificate: SignedMediaGenerationCertificate,187+ verification_failures: Arc<AtomicU64>,188+ ) -> Self {189+ Self {190+ track,191+ current_group: None,192+ last_group_sequence: None,193+ last_packet_sequence: None,194+ discontinuity_pending: true,195+ certificate,196+ verification_failures,197+ }198+ }199+200+ pub(crate) async fn next(&mut self) -> Result<Option<AudioPacket>, String> {201+ loop {202+ if self.current_group.is_none() {203+ let Some(group) = self204+ .track205+ .next_group()206+ .await207+ .map_err(|error| format!("Upstream MoQ audio track failed: {error}"))?208+ else {209+ return Ok(None);210+ };211+ if self212+ .last_group_sequence213+ .is_some_and(|previous| previous.checked_add(1) != Some(group.sequence))214+ {215+ self.discontinuity_pending = true;216+ }217+ if self218+ .last_group_sequence219+ .is_some_and(|previous| group.sequence <= previous)220+ {221+ return Err("Upstream MoQ audio group sequence did not increase".to_string());222+ }223+ self.current_group = Some(MoqAudioGroupCursor {224+ sequence: group.sequence,225+ group,226+ object_count: 0,227+ object_bytes: 0,228+ });229+ }230+231+ let cursor = self232+ .current_group233+ .as_mut()234+ .expect("audio group initialized");235+ let object = match tokio::time::timeout(236+ CLIENT_MOQ_OBJECT_READ_TIMEOUT,237+ cursor.group.read_frame(),238+ )239+ .await240+ {241+ Err(_) => {242+ self.last_group_sequence = Some(cursor.sequence);243+ self.current_group = None;244+ self.discontinuity_pending = true;245+ continue;246+ }247+ Ok(Err(error)) if moq_group_error_is_recoverable(&error) => {248+ self.last_group_sequence = Some(cursor.sequence);249+ self.current_group = None;250+ self.discontinuity_pending = true;251+ continue;252+ }253+ Ok(Err(error)) => {254+ return Err(format!(255+ "Upstream MoQ audio group {} failed: {error}",256+ cursor.sequence257+ ));258+ }259+ Ok(Ok(None)) => {260+ self.last_group_sequence = Some(cursor.sequence);261+ self.current_group = None;262+ continue;263+ }264+ Ok(Ok(Some(object))) => object,265+ };266+ if cursor.object_count >= 5 {267+ return Err("Upstream MoQ audio group exceeded five packets".to_string());268+ }269+ cursor.object_bytes = cursor270+ .object_bytes271+ .checked_add(object.len())272+ .filter(|bytes| {273+ *bytes274+ <= 5 * (sigil_protocol::AUDIO_HEADER_LEN275+ + sigil_protocol::MAX_AUDIO_PAYLOAD_LEN276+ + 256)277+ })278+ .ok_or_else(|| "Upstream MoQ audio group exceeded its byte bound".to_string())?;279+ let coordinates = AuthenticatedMediaObject::coordinates(&object).map_err(|error| {280+ self.verification_failures.fetch_add(1, Ordering::Relaxed);281+ format!("Invalid authenticated MoQ audio header: {error}")282+ })?;283+ let expected = MediaObjectCoordinates {284+ generation_id: self.certificate.claims.generation_id,285+ track: MediaTrack::AudioOpus,286+ group_id: cursor.sequence,287+ object_id: u32::try_from(cursor.object_count)288+ .map_err(|_| "MoQ audio object index overflowed".to_string())?,289+ flags: coordinates.flags,290+ };291+ let payload = AuthenticatedMediaObject::verify(&object, &self.certificate, expected)292+ .map_err(|error| {293+ self.verification_failures.fetch_add(1, Ordering::Relaxed);294+ format!("Authenticated upstream MoQ audio verification failed: {error}")295+ })?;296+ let mut packet = AudioPacket::decode_datagram(payload)297+ .map_err(|error| format!("Invalid upstream MoQ Opus packet: {error}"))?;298+ if coordinates.flags != u16::from(packet.header.flags.bits()) {299+ self.verification_failures.fetch_add(1, Ordering::Relaxed);300+ return Err("Authenticated MoQ audio flags do not match the packet".to_string());301+ }302+ if self303+ .last_packet_sequence304+ .is_some_and(|previous| previous.checked_add(1) != Some(packet.header.sequence))305+ {306+ self.discontinuity_pending = true;307+ }308+ if self.discontinuity_pending309+ && !packet.header.flags.contains(AudioFlags::DISCONTINUITY)310+ {311+ packet.header = AudioPacketHeader::opus(312+ packet.payload.len(),313+ packet.header.sequence,314+ packet.header.capture_timestamp_us,315+ packet.header.pts_us,316+ AudioFlags::DISCONTINUITY,317+ )318+ .map_err(|error| error.to_string())?;319+ }320+ cursor.object_count += 1;321+ self.last_packet_sequence = Some(packet.header.sequence);322+ self.discontinuity_pending = false;323+ return Ok(Some(packet));324+ }325+ }326+}327+166328 impl MoqMediaReceiver {167329 fn new(168330 moq: Moq,@@ -545,11 +707,24 @@ pub(crate) async fn open_upstream_moq_media(
545707 endpoint: &Endpoint,546708 address: &iroh::EndpointAddr,547709 session_id: u64,710+ advertised_generation_id: Option<u64>,711+ advertised_broadcast_name: Option<&str>,548712 authentication: Option<MoqAuthenticationExpectation>,549713 verification_failures: Arc<AtomicU64>,550-) -> Result<(MoqMediaReceiver, iroh::endpoint::Connection), String> {551- let broadcast_name = media_moq_broadcast_name(session_id)552- .map_err(|error| format!("Invalid MoQ media session name: {error}"))?;714+) -> Result<715+ (716+ MoqMediaReceiver,717+ Option<MoqAudioReceiver>,718+ iroh::endpoint::Connection,719+ ),720+ String,721+> {722+ let generation_id = advertised_generation_id.unwrap_or(session_id);723+ let broadcast_name = match advertised_broadcast_name {724+ Some(name) => name.to_owned(),725+ None => media_moq_broadcast_name(session_id)726+ .map_err(|error| format!("Invalid MoQ media session name: {error}"))?,727+ };553728 let moq = Moq::new(endpoint.clone());554729 let mut session =555730 tokio::time::timeout(CLIENT_MOQ_CONNECT_TIMEOUT, moq.connect(address.clone()))@@ -581,7 +756,7 @@ pub(crate) async fn open_upstream_moq_media(
581756 capability582757 .verify_binding(583758 authentication.expected_host,584- session_id,759+ generation_id,585760 *endpoint.id().as_bytes(),586761 SubscriptionTracks::VIDEO_H264,587762 1,@@ -592,7 +767,7 @@ pub(crate) async fn open_upstream_moq_media(
592767 &broadcast,593768 CLIENT_MOQ_SUBSCRIBE_TIMEOUT,594769 authentication.expected_host,595- session_id,770+ generation_id,596771 authentication.now_unix,597772 )598773 .await?@@ -600,6 +775,14 @@ pub(crate) async fn open_upstream_moq_media(
600775 subscribe_goq_video_track(&broadcast, CLIENT_MOQ_SUBSCRIBE_TIMEOUT).await?601776 };602777 eprintln!("[client] moq catalog: {}", catalog.mode.label());778+ let audio = match (catalog.audio_track, catalog.certificate.clone()) {779+ (Some(track), Some(certificate)) => Some(MoqAudioReceiver::new(780+ track,781+ certificate,782+ Arc::clone(&verification_failures),783+ )),784+ _ => None,785+ };603786 Ok((604787 MoqMediaReceiver::new(605788 moq,@@ -609,6 +792,7 @@ pub(crate) async fn open_upstream_moq_media(
609792 catalog.certificate,610793 Some(verification_failures),611794 ),795+ audio,612796 diagnostics_connection,613797 ))614798 }@@ -716,6 +900,41 @@ mod tests {
716900 assert_eq!(failures.load(Ordering::Relaxed), 1);717901 }718902 903+ 904+ async fn authenticated_audio_receiver_preserves_verified_discontinuity() {905+ let mut producer = Track::new(sigil_protocol::MOQ_AUDIO_OPUS_TRACK).produce();906+ let (generation, certificate) = authentication_fixture();907+ let failures = Arc::new(AtomicU64::new(0));908+ let mut receiver =909+ MoqAudioReceiver::new(producer.consume(), certificate, Arc::clone(&failures));910+ let packet = AudioPacket::new(911+ AudioPacketHeader::opus(3, 7, 140_000, 140_000, AudioFlags::NONE).unwrap(),912+ vec![0xf8, 0xff, 0xfe],913+ )914+ .unwrap();915+ let payload = packet.encode_datagram().unwrap();916+ let mut group = producer.append_group().unwrap();917+ let object = generation918+ .authenticate(919+ MediaObjectCoordinates {920+ generation_id: 42,921+ track: MediaTrack::AudioOpus,922+ group_id: group.sequence,923+ object_id: 0,924+ flags: 0,925+ },926+ &payload,927+ )928+ .unwrap();929+ group.write_frame(object.into_bytes()).unwrap();930+ group.finish().unwrap();931+932+ let received = receiver.next().await.unwrap().unwrap();933+ assert_eq!(received.header.sequence, 7);934+ assert!(received.header.flags.contains(AudioFlags::DISCONTINUITY));935+ assert_eq!(failures.load(Ordering::Relaxed), 0);936+ }937+719938 720939 fn upstream_moq_group_gap_keyframe_is_the_recovery_barrier() {721940 let keyframe = media_object_frame(80, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));@@ -134,7 +134,7 @@ async fn negotiate_v2(
134134 nonce: [u8; 16],135135 invitation: Option<&str>,136136 ) -> Result<NegotiatedV2Stream, String> {137- let capabilities = vec![Capability::VideoH264];137+ let capabilities = vec![Capability::VideoH264, Capability::AudioOpus];138138 let mut hello = ClientHelloV2::new("portal/0.1.0", nonce, capabilities.clone());139139 if let Some(invitation) = invitation {140140 hello = hello.with_invitation(invitation);