- commit
- 1a9d5ab4ecd934cdc24e6913aecbc9fa521497c5
- parent
- 5f6d0fddb14aaaf361514e86b3d7bfc46b11f411
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 721 | +------------------------------ |
| A | crates/ | 727 | ++++++++++++++++++++++++++++++ |
@@ -47,13 +47,12 @@ const MAX_PENDING_HANDSHAKES: usize = 4;
4747 // sent, but never replay a suffix already more than two configured periods old.4848 const MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS: u64 = 2;4949 const SOURCE_REAP_GRACE_TIMEOUT: Duration = Duration::from_secs(1);50-const MOQ_ATTACHMENT_TIMEOUT: Duration = Duration::from_secs(10);51-const MOQ_REJECT_CODE: u32 = 0x534d;5250 const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);5351 5452 mod adaptive;5553 mod media_v2;5654 mod media_v3;55+mod moq;5756 mod session;5857 mod startup;5958 @@ -65,6 +64,7 @@ use media_v2::{media_frame_for_encoded, serve_media, serve_media_v2};
6564 use media_v3::{6665 MediaV3GroupCursor, forward_media_v3_control_requests, new_current_gop_frames, serve_media_v3,6766 };67+use moq::{MOQ_REJECT_CODE, serve_authorized_moq, serve_control_moq};6868 6969 pub use session::SessionRegistry;7070 use session::{@@ -94,144 +94,6 @@ enum MediaV3GroupDecision {
9494 EnterResync,9595 }9696 97-98-enum MoqGroupDecision {99- Published {100- group_id: u64,101- frame_id: u32,102- cancelled_previous: bool,103- },104- SkipUntilKeyframe,105- EnterResync,106-}107-108-/// Owns the single bounded live MoQ track. One configured H.264 GOP maps to109-/// one native MoQ group; its application frame sequence remains inside the110-/// encoded object envelope and is never reused as the transport group id.111-struct MoqGroupPublisher {112- track: TrackProducer,113- current_group: Option<GroupProducer>,114- cursor: MediaV3GroupCursor,115- object_bytes: usize,116-}117-118-impl MoqGroupPublisher {119- fn new(track: TrackProducer) -> Self {120- Self {121- track,122- current_group: None,123- cursor: MediaV3GroupCursor::default(),124- object_bytes: 0,125- }126- }127-128- fn publish(129- &mut self,130- config: &HostConfig,131- frame: &EncodedFrame,132- replay_discontinuity: bool,133- ) -> Result<MoqGroupDecision> {134- let position = match self.cursor.classify(frame) {135- MediaV3GroupDecision::Send(position) => position,136- MediaV3GroupDecision::SkipUntilKeyframe => {137- return Ok(MoqGroupDecision::SkipUntilKeyframe);138- }139- MediaV3GroupDecision::EnterResync => {140- self.abort_current();141- return Ok(MoqGroupDecision::EnterResync);142- }143- };144- let object = encode_media_frame_object(&media_frame_for_encoded(145- config,146- frame,147- replay_discontinuity || position.discontinuity,148- )?)?;149- let next_object_bytes = if position.object_id == 0 {150- Some(object.len())151- } else {152- self.object_bytes.checked_add(object.len())153- };154- let Some(next_object_bytes) =155- next_object_bytes.filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)156- else {157- self.cursor.request_keyframe();158- self.abort_current();159- return Ok(MoqGroupDecision::EnterResync);160- };161-162- if position.object_id == 0 {163- // A new independently-decodable GOP supersedes the previous one.164- // Actively aborting it cancels a slow subscriber rather than165- // retaining a playable history behind the live edge.166- let cancelled_previous = self.abort_current().is_some();167- let mut group = self168- .track169- .append_group()170- .context("creating sequential MoQ video group")?;171- let group_id = group.sequence;172- group173- .write_frame(object)174- .context("writing configured keyframe to MoQ group")?;175- self.object_bytes = next_object_bytes;176- self.current_group = Some(group);177- return Ok(MoqGroupDecision::Published {178- group_id,179- frame_id: 0,180- cancelled_previous,181- });182- }183-184- let group = self185- .current_group186- .as_mut()187- .context("MoQ delta frame has no active configured-keyframe group")?;188- let group_id = group.sequence;189- group190- .write_frame(object)191- .context("writing delta access unit to MoQ group")?;192- self.object_bytes = next_object_bytes;193- Ok(MoqGroupDecision::Published {194- group_id,195- frame_id: position.object_id,196- cancelled_previous: false,197- })198- }199-200- fn request_keyframe(&mut self) -> Option<u64> {201- self.cursor.request_keyframe();202- self.abort_current()203- }204-205- fn abort_current(&mut self) -> Option<u64> {206- self.object_bytes = 0;207- let mut group = self.current_group.take()?;208- let group_id = group.sequence;209- let _ = group.abort(MoqError::Cancel);210- Some(group_id)211- }212-213- fn abort(mut self) {214- self.abort_current();215- let _ = self.track.abort(MoqError::Cancel);216- }217-}218-219-fn apply_moq_keyframe_request(220- publisher: &mut MoqGroupPublisher,221- replay_cursor: &mut MediaReplayCursor,222- through_sequence: Option<u64>,223- reason: KeyframeRequestReasonV3,224-) -> Option<u64> {225- // The bounded current group is already the late joiner's decodable replay.226- // Aborting it on Join can strand a static source until its next natural IDR.227- if reason == KeyframeRequestReasonV3::Join {228- return None;229- }230- let cancelled_group = publisher.request_keyframe();231- replay_cursor.enter_resync_through(through_sequence);232- cancelled_group233-}234-23597 23698 struct MediaReplayCursor {23799 last_sequence: Option<u64>,@@ -500,411 +362,6 @@ impl ProtocolHandler for InputHandler {
500362 }501363 }502364 503-async fn serve_authorized_moq(504- connection: Connection,505- origin: Origin,506- attachment: ClaimedMoqAttachment,507-) -> Result<()> {508- let ClaimedMoqAttachment {509- session_id,510- broadcast_name,511- broadcast,512- attached,513- closed,514- telemetry,515- } = attachment;516- let result: Result<()> = async {517- let web_transport = web_transport_iroh::Session::raw(connection);518- let session = tokio::time::timeout(519- MOQ_ATTACHMENT_TIMEOUT,520- iroh_moq::MoqSession::session_accept(web_transport, origin),521- )522- .await523- .context("timed out completing authorized MoQ handshake")?524- .context("completing authorized MoQ handshake")?;525- let broadcast_closed = broadcast.clone();526- session.publish(&broadcast_name, broadcast);527- ensure!(528- attached.send(()).is_ok(),529- "control session ended before MoQ attachment completed"530- );531- info!(532- remote = %session.remote_id(),533- session_id,534- %broadcast_name,535- track = MOQ_VIDEO_H264_TRACK,536- "authorized MoQ media attachment accepted"537- );538- let mut telemetry_interval = tokio::time::interval(Duration::from_secs(1));539- telemetry_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);540- loop {541- tokio::select! {542- reason = session.closed() => {543- debug!(remote = %session.remote_id(), ?reason, "MoQ media session closed");544- break;545- }546- reason = broadcast_closed.closed() => {547- debug!(remote = %session.remote_id(), ?reason, "control-owned MoQ broadcast closed");548- session.close(0, b"control session ended");549- break;550- }551- _ = telemetry_interval.tick() => {552- telemetry.record_selected_path(session.conn());553- }554- }555- }556- Ok(())557- }558- .await;559- let _ = closed.send(());560- result561-}562-563-async fn serve_control_moq(564- connection: Connection,565- config: HostConfig,566- sessions: &Arc<SessionRegistry>,567- authorization: &AuthorizationPolicy,568-) -> Result<()> {569- let remote = connection.remote_id();570- let handshake_permit = sessions571- .pending_handshakes572- .try_acquire()573- .context("too many pending handshakes")?;574- let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())575- .await576- .context("timed out accepting MoQ control stream")?577- .context("accepting MoQ control stream")?;578- let hello = receive_hello(&mut recv, Capability::VideoH264).await?;579- drop(handshake_permit);580- debug!(%remote, agent = %hello.agent, "MoQ control hello received");581-582- let grants = match authorization.authorize_or_redeem(583- remote,584- hello.invitation.as_deref(),585- unix_timestamp_now()?,586- ) {587- Ok(grants) => grants,588- Err(error) => {589- send_rejection(&mut send, "Portal peer is not authorized").await?;590- return Err(error.context("authorizing MoQ control peer"));591- }592- };593- ensure!(594- grants.contains(InvitationGrants::VIEW),595- "authorized MoQ control peer lacks view permission"596- );597- let lease = match sessions.claim(remote, hello.nonce, grants) {598- Ok(lease) => lease,599- Err(error) => {600- send_rejection(&mut send, "host already has an active client").await?;601- return Err(error);602- }603- };604-605- let source = match config.source {606- VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),607- VideoSource::GamescopePipewire => {608- let primary = spawn_gamescope_pipewire_after_static_preflight(609- config.clone(),610- lease.session_clock,611- )612- .await?;613- select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await614- }615- };616- let EncodedSource {617- frames: frame_receiver,618- current_gop: mut current_gop_receiver,619- task: source_task,620- pointer_surface_dimensions,621- encoder_control,622- } = match source {623- Ok(source) => source,624- Err(error) => {625- send_rejection(&mut send, "video source is unavailable").await?;626- return Err(error);627- }628- };629- let source_task = SourceTaskGuard::new(source_task);630- sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;631-632- let mut broadcast = Broadcast::new().produce();633- let track = broadcast634- .create_track(Track {635- name: MOQ_VIDEO_H264_TRACK.to_owned(),636- priority: MOQ_VIDEO_TRACK_PRIORITY,637- })638- .context("creating static MoQ H.264 track")?;639- let catalog = publish_goq_catalog(&mut broadcast)?;640- let broadcast_name = media_moq_broadcast_name(lease.session_id)?;641- let attachment = sessions.expect_moq(642- remote,643- lease.session_id,644- broadcast_name.clone(),645- broadcast.consume(),646- )?;647-648- let mut control_hello = HostHello::accepted(649- lease.session_id,650- negotiated_capabilities(&hello, MEDIA_CAPABILITIES),651- );652- if let Some(dimensions) = pointer_surface_dimensions {653- control_hello = control_hello.with_pointer_surface_dimensions(dimensions);654- }655- write_host_hello(&mut send, &control_hello).await?;656- send.finish().context("finishing MoQ control response")?;657- drop(send);658- info!(659- %remote,660- session_id = lease.session_id,661- %broadcast_name,662- "MoQ control client accepted; awaiting authorized media attachment"663- );664-665- let MoqAttachmentWait {666- mut attached,667- closed,668- } = attachment;669- tokio::time::timeout(MOQ_ATTACHMENT_TIMEOUT, async {670- tokio::select! {671- result = &mut attached => {672- result.context("authorized MoQ handler ended before attachment")673- }674- reason = connection.closed() => {675- Err(anyhow::anyhow!("control connection closed before MoQ attachment: {reason:?}"))676- }677- }678- })679- .await680- .context("timed out waiting for authorized MoQ attachment")??;681-682- let session_result = run_control_moq_session(683- &connection,684- &config,685- &mut current_gop_receiver,686- recv,687- remote,688- closed,689- track,690- &mut broadcast,691- encoder_control,692- Arc::clone(&lease.media_v3_telemetry),693- )694- .await;695- let catalog_result = catalog.finish();696-697- drop(current_gop_receiver);698- drop(frame_receiver);699- source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;700- drop(lease);701- info!(%remote, "MoQ control client released");702- match session_result {703- Err(error) => Err(error),704- Ok(()) => catalog_result,705- }706-}707-708-709-async fn run_control_moq_session(710- connection: &Connection,711- config: &HostConfig,712- current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,713- control_recv: iroh::endpoint::RecvStream,714- remote: EndpointId,715- mut moq_closed: tokio::sync::oneshot::Receiver<()>,716- track: TrackProducer,717- broadcast: &mut BroadcastProducer,718- encoder_control: Option<EncoderControl>,719- telemetry: Arc<MediaV3Telemetry>,720-) -> Result<()> {721- let maximum_replay_age = maximum_media_replay_age(config.framerate);722- let mut replay_cursor = MediaReplayCursor::default();723- let mut publisher = MoqGroupPublisher::new(track);724- let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);725- let mut control_task = tokio::spawn(forward_media_v3_control_requests(726- control_recv,727- control_sender,728- ));729- let mut control_task_finished = false;730- let mut control_receiver_open = true;731- let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));732-733- let result = async {734- loop {735- tokio::select! {736- biased;737- control_result = &mut control_task, if !control_task_finished => {738- control_task_finished = true;739- match control_result {740- Ok(Ok(())) => {741- debug!(%remote, "MoQ keyframe-control stream closed");742- }743- Ok(Err(error)) => {744- return Err(error).context("reading MoQ keyframe-control stream");745- }746- Err(error) => {747- return Err(error).context("MoQ keyframe-control task failed");748- }749- }750- }751- changed = control_requests.changed(), if control_receiver_open => {752- if changed.is_err() {753- control_receiver_open = false;754- continue;755- }756- let Some(request) = *control_requests.borrow_and_update() else {757- continue;758- };759- let through_sequence = current_gop_receiver760- .borrow()761- .as_ref()762- .and_then(|gop| gop.frames.last())763- .map(|frame| frame.sequence);764- let cancelled_group = apply_moq_keyframe_request(765- &mut publisher,766- &mut replay_cursor,767- through_sequence,768- request.reason,769- );770- if cancelled_group.is_some() {771- telemetry772- .scheduler_cancellations773- .fetch_add(1, Ordering::Relaxed);774- }775- let forced_idr_disposition = forced_idr.request(request.reason);776- if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {777- warn!(778- %remote,779- request_id = request.request_id,780- ?request.reason,781- %error,782- "forced-IDR request failed; retaining natural-IDR fallback"783- );784- }785- debug!(786- %remote,787- request_id = request.request_id,788- ?request.reason,789- advisory_last_sequence = ?request.last_sequence,790- coalesced = cancelled_group.is_none(),791- ?cancelled_group,792- ?forced_idr_disposition,793- "accepted MoQ keyframe request"794- );795- }796- acknowledgement = forced_idr.acknowledgements.join_next(),797- if forced_idr.pending_revision.is_some() =>798- {799- forced_idr.complete(acknowledgement, remote, "iroh-moq");800- }801- reason = connection.closed() => {802- debug!(%remote, ?reason, "MoQ control connection closed");803- return Ok(());804- }805- result = &mut moq_closed => {806- debug!(%remote, ?result, "authorized MoQ media attachment closed");807- return Ok(());808- }809- changed = current_gop_receiver.changed() => {810- if let Err(error) = changed {811- return Err(error).context("encoded source stopped");812- }813- let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {814- continue;815- };816- let initial_replay_started_at =817- replay_cursor.last_sequence.is_none().then(Instant::now);818- let replay_through_sequence = current_gop819- .frames820- .last()821- .map(|frame| frame.sequence)822- .context("current GOP snapshot is empty")?;823- for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {824- let replay_discontinuity = match replay_cursor.classify(825- &frame,826- replay_through_sequence,827- initial_replay_started_at,828- Instant::now(),829- maximum_replay_age,830- ) {831- MediaReplayDecision::Send { discontinuity } => discontinuity,832- MediaReplayDecision::SkipUntilKeyframe => {833- if publisher.request_keyframe().is_some() {834- telemetry835- .scheduler_cancellations836- .fetch_add(1, Ordering::Relaxed);837- }838- replay_cursor.enter_resync_through(Some(replay_through_sequence));839- break;840- }841- MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {842- let cancelled_group = publisher.request_keyframe();843- if cancelled_group.is_some() {844- telemetry845- .scheduler_cancellations846- .fetch_add(1, Ordering::Relaxed);847- }848- debug!(849- %remote,850- through_sequence,851- ?cancelled_group,852- "cancelled stale MoQ media suffix"853- );854- break;855- }856- };857- let decision = publisher858- .publish(config, &frame, replay_discontinuity)859- .inspect_err(|_error| {860- telemetry.send_failures.fetch_add(1, Ordering::Relaxed);861- })?;862- match decision {863- MoqGroupDecision::Published {864- group_id,865- frame_id,866- cancelled_previous,867- } => {868- if cancelled_previous {869- telemetry870- .scheduler_cancellations871- .fetch_add(1, Ordering::Relaxed);872- }873- debug!(874- sequence = frame.sequence,875- group_id,876- frame_id,877- cancelled_previous,878- "published upstream MoQ video frame"879- );880- replay_cursor.commit_sent(&frame);881- }882- MoqGroupDecision::SkipUntilKeyframe => {883- replay_cursor.enter_resync_through(Some(replay_through_sequence));884- break;885- }886- MoqGroupDecision::EnterResync => {887- replay_cursor.enter_resync_through(Some(replay_through_sequence));888- break;889- }890- }891- }892- }893- }894- }895- }896- .await;897-898- forced_idr.abort_and_drain(remote, "iroh-moq").await;899- publisher.abort();900- let _ = broadcast.abort(MoqError::Cancel);901- if !control_task_finished {902- control_task.abort();903- let _ = control_task.await;904- }905- result906-}907-908365 async fn serve_input(909366 connection: Connection,910367 backend: &InputBackend,@@ -1371,180 +828,6 @@ impl Drop for DropNotify {
1371828 mod tests {1372829 use super::*;1373830 1374- 1375- async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {1376- let track_info = Track {1377- name: MOQ_VIDEO_H264_TRACK.to_owned(),1378- priority: MOQ_VIDEO_TRACK_PRIORITY,1379- };1380- let mut broadcast = Broadcast::new().produce();1381- let track = broadcast.create_track(track_info.clone()).unwrap();1382- let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1383- let mut publisher = MoqGroupPublisher::new(track);1384- let config = moq_test_config();1385-1386- assert_eq!(1387- publisher1388- .publish(&config, &media_v3_encoded_frame(100, true, true, 4), false)1389- .unwrap(),1390- MoqGroupDecision::Published {1391- group_id: 0,1392- frame_id: 0,1393- cancelled_previous: false,1394- }1395- );1396- let mut first_group = consumer.recv_group().await.unwrap().unwrap();1397- assert_eq!(first_group.sequence, 0);1398- assert!(first_group.read_frame().await.unwrap().is_some());1399-1400- assert_eq!(1401- publisher1402- .publish(1403- &config,1404- &media_v3_encoded_frame(101, false, false, 4),1405- false,1406- )1407- .unwrap(),1408- MoqGroupDecision::Published {1409- group_id: 0,1410- frame_id: 1,1411- cancelled_previous: false,1412- }1413- );1414- assert!(first_group.read_frame().await.unwrap().is_some());1415-1416- assert_eq!(1417- publisher1418- .publish(&config, &media_v3_encoded_frame(200, true, true, 4), false)1419- .unwrap(),1420- MoqGroupDecision::Published {1421- group_id: 1,1422- frame_id: 0,1423- cancelled_previous: true,1424- }1425- );1426- assert!(first_group.finished().await.is_err());1427- let mut second_group = consumer.recv_group().await.unwrap().unwrap();1428- assert_eq!(second_group.sequence, 1);1429- let object = second_group.read_frame().await.unwrap().unwrap();1430- let frame = sigil_protocol::decode_media_frame_object(&object).unwrap();1431- assert_eq!(frame.header.sequence, 200);1432- assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));1433- }1434-1435- 1436- async fn upstream_moq_late_join_preserves_active_static_group() {1437- let track_info = Track {1438- name: MOQ_VIDEO_H264_TRACK.to_owned(),1439- priority: MOQ_VIDEO_TRACK_PRIORITY,1440- };1441- let mut broadcast = Broadcast::new().produce();1442- let track = broadcast.create_track(track_info.clone()).unwrap();1443- let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1444- let mut publisher = MoqGroupPublisher::new(track);1445- let mut replay_cursor = MediaReplayCursor::default();1446- let config = moq_test_config();1447- let keyframe = media_v3_encoded_frame(10, true, true, 1);1448- let first_delta = media_v3_encoded_frame(11, false, false, 1);1449- let next_delta = media_v3_encoded_frame(12, false, false, 1);1450-1451- publisher.publish(&config, &keyframe, false).unwrap();1452- replay_cursor.commit_sent(&keyframe);1453- let mut active_group = consumer.recv_group().await.unwrap().unwrap();1454- assert!(active_group.read_frame().await.unwrap().is_some());1455-1456- publisher.publish(&config, &first_delta, false).unwrap();1457- replay_cursor.commit_sent(&first_delta);1458- assert!(active_group.read_frame().await.unwrap().is_some());1459-1460- assert_eq!(1461- apply_moq_keyframe_request(1462- &mut publisher,1463- &mut replay_cursor,1464- Some(first_delta.sequence),1465- KeyframeRequestReasonV3::Join,1466- ),1467- None1468- );1469- assert_eq!(replay_cursor.last_sequence, Some(first_delta.sequence));1470- assert!(!replay_cursor.waiting_for_keyframe);1471- assert_eq!(1472- publisher.publish(&config, &next_delta, false).unwrap(),1473- MoqGroupDecision::Published {1474- group_id: 0,1475- frame_id: 2,1476- cancelled_previous: false,1477- }1478- );1479- assert!(active_group.read_frame().await.unwrap().is_some());1480- }1481-1482- 1483- async fn upstream_moq_resync_aborts_current_group_and_waits_for_configured_idr() {1484- let track_info = Track {1485- name: MOQ_VIDEO_H264_TRACK.to_owned(),1486- priority: MOQ_VIDEO_TRACK_PRIORITY,1487- };1488- let mut broadcast = Broadcast::new().produce();1489- let track = broadcast.create_track(track_info.clone()).unwrap();1490- let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1491- let mut publisher = MoqGroupPublisher::new(track);1492- let config = moq_test_config();1493-1494- publisher1495- .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)1496- .unwrap();1497- let mut cancelled = consumer.recv_group().await.unwrap().unwrap();1498- assert_eq!(publisher.request_keyframe(), Some(0));1499- assert!(cancelled.finished().await.is_err());1500- assert_eq!(1501- publisher1502- .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)1503- .unwrap(),1504- MoqGroupDecision::SkipUntilKeyframe1505- );1506- assert_eq!(1507- publisher1508- .publish(&config, &media_v3_encoded_frame(20, true, true, 1), false)1509- .unwrap(),1510- MoqGroupDecision::Published {1511- group_id: 1,1512- frame_id: 0,1513- cancelled_previous: false,1514- }1515- );1516- }1517-1518- 1519- async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {1520- let track_info = Track {1521- name: MOQ_VIDEO_H264_TRACK.to_owned(),1522- priority: MOQ_VIDEO_TRACK_PRIORITY,1523- };1524- let mut broadcast = Broadcast::new().produce();1525- let track = broadcast.create_track(track_info.clone()).unwrap();1526- let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();1527- let mut publisher = MoqGroupPublisher::new(track);1528- let config = moq_test_config();1529- publisher1530- .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)1531- .unwrap();1532- let mut cancelled = consumer.recv_group().await.unwrap().unwrap();1533-1534- // Payload-only accounting would accept this next one-byte access unit,1535- // but its fixed application envelope would overflow moq-net's 32 MiB1536- // group cache and silently evict the keyframe.1537- publisher.object_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;1538- assert_eq!(1539- publisher1540- .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)1541- .unwrap(),1542- MoqGroupDecision::EnterResync1543- );1544- assert!(cancelled.finished().await.is_err());1545- assert_eq!(publisher.object_bytes, 0);1546- }1547-1548831 1549832 fn capability_negotiation_is_an_exact_intersection() {1550833 let hello = ClientHello::new(@@ -0,0 +1,727 @@
1+use super::*;2+3+const MOQ_ATTACHMENT_TIMEOUT: Duration = Duration::from_secs(10);4+pub(super) const MOQ_REJECT_CODE: u32 = 0x534d;5+6+7+enum MoqGroupDecision {8+ Published {9+ group_id: u64,10+ frame_id: u32,11+ cancelled_previous: bool,12+ },13+ SkipUntilKeyframe,14+ EnterResync,15+}16+17+/// Owns the single bounded live MoQ track. One configured H.264 GOP maps to18+/// one native MoQ group; its application frame sequence remains inside the19+/// encoded object envelope and is never reused as the transport group id.20+struct MoqGroupPublisher {21+ track: TrackProducer,22+ current_group: Option<GroupProducer>,23+ cursor: MediaV3GroupCursor,24+ object_bytes: usize,25+}26+27+impl MoqGroupPublisher {28+ fn new(track: TrackProducer) -> Self {29+ Self {30+ track,31+ current_group: None,32+ cursor: MediaV3GroupCursor::default(),33+ object_bytes: 0,34+ }35+ }36+37+ fn publish(38+ &mut self,39+ config: &HostConfig,40+ frame: &EncodedFrame,41+ replay_discontinuity: bool,42+ ) -> Result<MoqGroupDecision> {43+ let position = match self.cursor.classify(frame) {44+ MediaV3GroupDecision::Send(position) => position,45+ MediaV3GroupDecision::SkipUntilKeyframe => {46+ return Ok(MoqGroupDecision::SkipUntilKeyframe);47+ }48+ MediaV3GroupDecision::EnterResync => {49+ self.abort_current();50+ return Ok(MoqGroupDecision::EnterResync);51+ }52+ };53+ let object = encode_media_frame_object(&media_frame_for_encoded(54+ config,55+ frame,56+ replay_discontinuity || position.discontinuity,57+ )?)?;58+ let next_object_bytes = if position.object_id == 0 {59+ Some(object.len())60+ } else {61+ self.object_bytes.checked_add(object.len())62+ };63+ let Some(next_object_bytes) =64+ next_object_bytes.filter(|bytes| *bytes <= MAX_MEDIA_GROUP_BYTES_V3)65+ else {66+ self.cursor.request_keyframe();67+ self.abort_current();68+ return Ok(MoqGroupDecision::EnterResync);69+ };70+71+ if position.object_id == 0 {72+ // A new independently-decodable GOP supersedes the previous one.73+ // Actively aborting it cancels a slow subscriber rather than74+ // retaining a playable history behind the live edge.75+ let cancelled_previous = self.abort_current().is_some();76+ let mut group = self77+ .track78+ .append_group()79+ .context("creating sequential MoQ video group")?;80+ let group_id = group.sequence;81+ group82+ .write_frame(object)83+ .context("writing configured keyframe to MoQ group")?;84+ self.object_bytes = next_object_bytes;85+ self.current_group = Some(group);86+ return Ok(MoqGroupDecision::Published {87+ group_id,88+ frame_id: 0,89+ cancelled_previous,90+ });91+ }92+93+ let group = self94+ .current_group95+ .as_mut()96+ .context("MoQ delta frame has no active configured-keyframe group")?;97+ let group_id = group.sequence;98+ group99+ .write_frame(object)100+ .context("writing delta access unit to MoQ group")?;101+ self.object_bytes = next_object_bytes;102+ Ok(MoqGroupDecision::Published {103+ group_id,104+ frame_id: position.object_id,105+ cancelled_previous: false,106+ })107+ }108+109+ fn request_keyframe(&mut self) -> Option<u64> {110+ self.cursor.request_keyframe();111+ self.abort_current()112+ }113+114+ fn abort_current(&mut self) -> Option<u64> {115+ self.object_bytes = 0;116+ let mut group = self.current_group.take()?;117+ let group_id = group.sequence;118+ let _ = group.abort(MoqError::Cancel);119+ Some(group_id)120+ }121+122+ fn abort(mut self) {123+ self.abort_current();124+ let _ = self.track.abort(MoqError::Cancel);125+ }126+}127+128+fn apply_moq_keyframe_request(129+ publisher: &mut MoqGroupPublisher,130+ replay_cursor: &mut MediaReplayCursor,131+ through_sequence: Option<u64>,132+ reason: KeyframeRequestReasonV3,133+) -> Option<u64> {134+ // The bounded current group is already the late joiner's decodable replay.135+ // Aborting it on Join can strand a static source until its next natural IDR.136+ if reason == KeyframeRequestReasonV3::Join {137+ return None;138+ }139+ let cancelled_group = publisher.request_keyframe();140+ replay_cursor.enter_resync_through(through_sequence);141+ cancelled_group142+}143+144+pub(super) async fn serve_authorized_moq(145+ connection: Connection,146+ origin: Origin,147+ attachment: ClaimedMoqAttachment,148+) -> Result<()> {149+ let ClaimedMoqAttachment {150+ session_id,151+ broadcast_name,152+ broadcast,153+ attached,154+ closed,155+ telemetry,156+ } = attachment;157+ let result: Result<()> = async {158+ let web_transport = web_transport_iroh::Session::raw(connection);159+ let session = tokio::time::timeout(160+ MOQ_ATTACHMENT_TIMEOUT,161+ iroh_moq::MoqSession::session_accept(web_transport, origin),162+ )163+ .await164+ .context("timed out completing authorized MoQ handshake")?165+ .context("completing authorized MoQ handshake")?;166+ let broadcast_closed = broadcast.clone();167+ session.publish(&broadcast_name, broadcast);168+ ensure!(169+ attached.send(()).is_ok(),170+ "control session ended before MoQ attachment completed"171+ );172+ info!(173+ remote = %session.remote_id(),174+ session_id,175+ %broadcast_name,176+ track = MOQ_VIDEO_H264_TRACK,177+ "authorized MoQ media attachment accepted"178+ );179+ let mut telemetry_interval = tokio::time::interval(Duration::from_secs(1));180+ telemetry_interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);181+ loop {182+ tokio::select! {183+ reason = session.closed() => {184+ debug!(remote = %session.remote_id(), ?reason, "MoQ media session closed");185+ break;186+ }187+ reason = broadcast_closed.closed() => {188+ debug!(remote = %session.remote_id(), ?reason, "control-owned MoQ broadcast closed");189+ session.close(0, b"control session ended");190+ break;191+ }192+ _ = telemetry_interval.tick() => {193+ telemetry.record_selected_path(session.conn());194+ }195+ }196+ }197+ Ok(())198+ }199+ .await;200+ let _ = closed.send(());201+ result202+}203+204+pub(super) async fn serve_control_moq(205+ connection: Connection,206+ config: HostConfig,207+ sessions: &Arc<SessionRegistry>,208+ authorization: &AuthorizationPolicy,209+) -> Result<()> {210+ let remote = connection.remote_id();211+ let handshake_permit = sessions212+ .pending_handshakes213+ .try_acquire()214+ .context("too many pending handshakes")?;215+ let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())216+ .await217+ .context("timed out accepting MoQ control stream")?218+ .context("accepting MoQ control stream")?;219+ let hello = receive_hello(&mut recv, Capability::VideoH264).await?;220+ drop(handshake_permit);221+ debug!(%remote, agent = %hello.agent, "MoQ control hello received");222+223+ let grants = match authorization.authorize_or_redeem(224+ remote,225+ hello.invitation.as_deref(),226+ unix_timestamp_now()?,227+ ) {228+ Ok(grants) => grants,229+ Err(error) => {230+ send_rejection(&mut send, "Portal peer is not authorized").await?;231+ return Err(error.context("authorizing MoQ control peer"));232+ }233+ };234+ ensure!(235+ grants.contains(InvitationGrants::VIEW),236+ "authorized MoQ control peer lacks view permission"237+ );238+ let lease = match sessions.claim(remote, hello.nonce, grants) {239+ Ok(lease) => lease,240+ Err(error) => {241+ send_rejection(&mut send, "host already has an active client").await?;242+ return Err(error);243+ }244+ };245+246+ let source = match config.source {247+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),248+ VideoSource::GamescopePipewire => {249+ let primary = spawn_gamescope_pipewire_after_static_preflight(250+ config.clone(),251+ lease.session_clock,252+ )253+ .await?;254+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await255+ }256+ };257+ let EncodedSource {258+ frames: frame_receiver,259+ current_gop: mut current_gop_receiver,260+ task: source_task,261+ pointer_surface_dimensions,262+ encoder_control,263+ } = match source {264+ Ok(source) => source,265+ Err(error) => {266+ send_rejection(&mut send, "video source is unavailable").await?;267+ return Err(error);268+ }269+ };270+ let source_task = SourceTaskGuard::new(source_task);271+ sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;272+273+ let mut broadcast = Broadcast::new().produce();274+ let track = broadcast275+ .create_track(Track {276+ name: MOQ_VIDEO_H264_TRACK.to_owned(),277+ priority: MOQ_VIDEO_TRACK_PRIORITY,278+ })279+ .context("creating static MoQ H.264 track")?;280+ let catalog = publish_goq_catalog(&mut broadcast)?;281+ let broadcast_name = media_moq_broadcast_name(lease.session_id)?;282+ let attachment = sessions.expect_moq(283+ remote,284+ lease.session_id,285+ broadcast_name.clone(),286+ broadcast.consume(),287+ )?;288+289+ let mut control_hello = HostHello::accepted(290+ lease.session_id,291+ negotiated_capabilities(&hello, MEDIA_CAPABILITIES),292+ );293+ if let Some(dimensions) = pointer_surface_dimensions {294+ control_hello = control_hello.with_pointer_surface_dimensions(dimensions);295+ }296+ write_host_hello(&mut send, &control_hello).await?;297+ send.finish().context("finishing MoQ control response")?;298+ drop(send);299+ info!(300+ %remote,301+ session_id = lease.session_id,302+ %broadcast_name,303+ "MoQ control client accepted; awaiting authorized media attachment"304+ );305+306+ let MoqAttachmentWait {307+ mut attached,308+ closed,309+ } = attachment;310+ tokio::time::timeout(MOQ_ATTACHMENT_TIMEOUT, async {311+ tokio::select! {312+ result = &mut attached => {313+ result.context("authorized MoQ handler ended before attachment")314+ }315+ reason = connection.closed() => {316+ Err(anyhow::anyhow!("control connection closed before MoQ attachment: {reason:?}"))317+ }318+ }319+ })320+ .await321+ .context("timed out waiting for authorized MoQ attachment")??;322+323+ let session_result = run_control_moq_session(324+ &connection,325+ &config,326+ &mut current_gop_receiver,327+ recv,328+ remote,329+ closed,330+ track,331+ &mut broadcast,332+ encoder_control,333+ Arc::clone(&lease.media_v3_telemetry),334+ )335+ .await;336+ let catalog_result = catalog.finish();337+338+ drop(current_gop_receiver);339+ drop(frame_receiver);340+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;341+ drop(lease);342+ info!(%remote, "MoQ control client released");343+ match session_result {344+ Err(error) => Err(error),345+ Ok(()) => catalog_result,346+ }347+}348+349+350+async fn run_control_moq_session(351+ connection: &Connection,352+ config: &HostConfig,353+ current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,354+ control_recv: iroh::endpoint::RecvStream,355+ remote: EndpointId,356+ mut moq_closed: tokio::sync::oneshot::Receiver<()>,357+ track: TrackProducer,358+ broadcast: &mut BroadcastProducer,359+ encoder_control: Option<EncoderControl>,360+ telemetry: Arc<MediaV3Telemetry>,361+) -> Result<()> {362+ let maximum_replay_age = maximum_media_replay_age(config.framerate);363+ let mut replay_cursor = MediaReplayCursor::default();364+ let mut publisher = MoqGroupPublisher::new(track);365+ let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);366+ let mut control_task = tokio::spawn(forward_media_v3_control_requests(367+ control_recv,368+ control_sender,369+ ));370+ let mut control_task_finished = false;371+ let mut control_receiver_open = true;372+ let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));373+374+ let result = async {375+ loop {376+ tokio::select! {377+ biased;378+ control_result = &mut control_task, if !control_task_finished => {379+ control_task_finished = true;380+ match control_result {381+ Ok(Ok(())) => {382+ debug!(%remote, "MoQ keyframe-control stream closed");383+ }384+ Ok(Err(error)) => {385+ return Err(error).context("reading MoQ keyframe-control stream");386+ }387+ Err(error) => {388+ return Err(error).context("MoQ keyframe-control task failed");389+ }390+ }391+ }392+ changed = control_requests.changed(), if control_receiver_open => {393+ if changed.is_err() {394+ control_receiver_open = false;395+ continue;396+ }397+ let Some(request) = *control_requests.borrow_and_update() else {398+ continue;399+ };400+ let through_sequence = current_gop_receiver401+ .borrow()402+ .as_ref()403+ .and_then(|gop| gop.frames.last())404+ .map(|frame| frame.sequence);405+ let cancelled_group = apply_moq_keyframe_request(406+ &mut publisher,407+ &mut replay_cursor,408+ through_sequence,409+ request.reason,410+ );411+ if cancelled_group.is_some() {412+ telemetry413+ .scheduler_cancellations414+ .fetch_add(1, Ordering::Relaxed);415+ }416+ let forced_idr_disposition = forced_idr.request(request.reason);417+ if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {418+ warn!(419+ %remote,420+ request_id = request.request_id,421+ ?request.reason,422+ %error,423+ "forced-IDR request failed; retaining natural-IDR fallback"424+ );425+ }426+ debug!(427+ %remote,428+ request_id = request.request_id,429+ ?request.reason,430+ advisory_last_sequence = ?request.last_sequence,431+ coalesced = cancelled_group.is_none(),432+ ?cancelled_group,433+ ?forced_idr_disposition,434+ "accepted MoQ keyframe request"435+ );436+ }437+ acknowledgement = forced_idr.acknowledgements.join_next(),438+ if forced_idr.pending_revision.is_some() =>439+ {440+ forced_idr.complete(acknowledgement, remote, "iroh-moq");441+ }442+ reason = connection.closed() => {443+ debug!(%remote, ?reason, "MoQ control connection closed");444+ return Ok(());445+ }446+ result = &mut moq_closed => {447+ debug!(%remote, ?result, "authorized MoQ media attachment closed");448+ return Ok(());449+ }450+ changed = current_gop_receiver.changed() => {451+ if let Err(error) = changed {452+ return Err(error).context("encoded source stopped");453+ }454+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {455+ continue;456+ };457+ let initial_replay_started_at =458+ replay_cursor.last_sequence.is_none().then(Instant::now);459+ let replay_through_sequence = current_gop460+ .frames461+ .last()462+ .map(|frame| frame.sequence)463+ .context("current GOP snapshot is empty")?;464+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {465+ let replay_discontinuity = match replay_cursor.classify(466+ &frame,467+ replay_through_sequence,468+ initial_replay_started_at,469+ Instant::now(),470+ maximum_replay_age,471+ ) {472+ MediaReplayDecision::Send { discontinuity } => discontinuity,473+ MediaReplayDecision::SkipUntilKeyframe => {474+ if publisher.request_keyframe().is_some() {475+ telemetry476+ .scheduler_cancellations477+ .fetch_add(1, Ordering::Relaxed);478+ }479+ replay_cursor.enter_resync_through(Some(replay_through_sequence));480+ break;481+ }482+ MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {483+ let cancelled_group = publisher.request_keyframe();484+ if cancelled_group.is_some() {485+ telemetry486+ .scheduler_cancellations487+ .fetch_add(1, Ordering::Relaxed);488+ }489+ debug!(490+ %remote,491+ through_sequence,492+ ?cancelled_group,493+ "cancelled stale MoQ media suffix"494+ );495+ break;496+ }497+ };498+ let decision = publisher499+ .publish(config, &frame, replay_discontinuity)500+ .inspect_err(|_error| {501+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);502+ })?;503+ match decision {504+ MoqGroupDecision::Published {505+ group_id,506+ frame_id,507+ cancelled_previous,508+ } => {509+ if cancelled_previous {510+ telemetry511+ .scheduler_cancellations512+ .fetch_add(1, Ordering::Relaxed);513+ }514+ debug!(515+ sequence = frame.sequence,516+ group_id,517+ frame_id,518+ cancelled_previous,519+ "published upstream MoQ video frame"520+ );521+ replay_cursor.commit_sent(&frame);522+ }523+ MoqGroupDecision::SkipUntilKeyframe => {524+ replay_cursor.enter_resync_through(Some(replay_through_sequence));525+ break;526+ }527+ MoqGroupDecision::EnterResync => {528+ replay_cursor.enter_resync_through(Some(replay_through_sequence));529+ break;530+ }531+ }532+ }533+ }534+ }535+ }536+ }537+ .await;538+539+ forced_idr.abort_and_drain(remote, "iroh-moq").await;540+ publisher.abort();541+ let _ = broadcast.abort(MoqError::Cancel);542+ if !control_task_finished {543+ control_task.abort();544+ let _ = control_task.await;545+ }546+ result547+}548+549+550+mod tests {551+ use super::super::{media_v3_encoded_frame, moq_test_config};552+ use super::*;553+554+ 555+ async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {556+ let track_info = Track {557+ name: MOQ_VIDEO_H264_TRACK.to_owned(),558+ priority: MOQ_VIDEO_TRACK_PRIORITY,559+ };560+ let mut broadcast = Broadcast::new().produce();561+ let track = broadcast.create_track(track_info.clone()).unwrap();562+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();563+ let mut publisher = MoqGroupPublisher::new(track);564+ let config = moq_test_config();565+566+ assert_eq!(567+ publisher568+ .publish(&config, &media_v3_encoded_frame(100, true, true, 4), false)569+ .unwrap(),570+ MoqGroupDecision::Published {571+ group_id: 0,572+ frame_id: 0,573+ cancelled_previous: false,574+ }575+ );576+ let mut first_group = consumer.recv_group().await.unwrap().unwrap();577+ assert_eq!(first_group.sequence, 0);578+ assert!(first_group.read_frame().await.unwrap().is_some());579+580+ assert_eq!(581+ publisher582+ .publish(583+ &config,584+ &media_v3_encoded_frame(101, false, false, 4),585+ false,586+ )587+ .unwrap(),588+ MoqGroupDecision::Published {589+ group_id: 0,590+ frame_id: 1,591+ cancelled_previous: false,592+ }593+ );594+ assert!(first_group.read_frame().await.unwrap().is_some());595+596+ assert_eq!(597+ publisher598+ .publish(&config, &media_v3_encoded_frame(200, true, true, 4), false)599+ .unwrap(),600+ MoqGroupDecision::Published {601+ group_id: 1,602+ frame_id: 0,603+ cancelled_previous: true,604+ }605+ );606+ assert!(first_group.finished().await.is_err());607+ let mut second_group = consumer.recv_group().await.unwrap().unwrap();608+ assert_eq!(second_group.sequence, 1);609+ let object = second_group.read_frame().await.unwrap().unwrap();610+ let frame = sigil_protocol::decode_media_frame_object(&object).unwrap();611+ assert_eq!(frame.header.sequence, 200);612+ assert!(frame.header.flags.contains(FrameFlags::DISCONTINUITY));613+ }614+615+ 616+ async fn upstream_moq_late_join_preserves_active_static_group() {617+ let track_info = Track {618+ name: MOQ_VIDEO_H264_TRACK.to_owned(),619+ priority: MOQ_VIDEO_TRACK_PRIORITY,620+ };621+ let mut broadcast = Broadcast::new().produce();622+ let track = broadcast.create_track(track_info.clone()).unwrap();623+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();624+ let mut publisher = MoqGroupPublisher::new(track);625+ let mut replay_cursor = MediaReplayCursor::default();626+ let config = moq_test_config();627+ let keyframe = media_v3_encoded_frame(10, true, true, 1);628+ let first_delta = media_v3_encoded_frame(11, false, false, 1);629+ let next_delta = media_v3_encoded_frame(12, false, false, 1);630+631+ publisher.publish(&config, &keyframe, false).unwrap();632+ replay_cursor.commit_sent(&keyframe);633+ let mut active_group = consumer.recv_group().await.unwrap().unwrap();634+ assert!(active_group.read_frame().await.unwrap().is_some());635+636+ publisher.publish(&config, &first_delta, false).unwrap();637+ replay_cursor.commit_sent(&first_delta);638+ assert!(active_group.read_frame().await.unwrap().is_some());639+640+ assert_eq!(641+ apply_moq_keyframe_request(642+ &mut publisher,643+ &mut replay_cursor,644+ Some(first_delta.sequence),645+ KeyframeRequestReasonV3::Join,646+ ),647+ None648+ );649+ assert_eq!(replay_cursor.last_sequence, Some(first_delta.sequence));650+ assert!(!replay_cursor.waiting_for_keyframe);651+ assert_eq!(652+ publisher.publish(&config, &next_delta, false).unwrap(),653+ MoqGroupDecision::Published {654+ group_id: 0,655+ frame_id: 2,656+ cancelled_previous: false,657+ }658+ );659+ assert!(active_group.read_frame().await.unwrap().is_some());660+ }661+662+ 663+ async fn upstream_moq_resync_aborts_current_group_and_waits_for_configured_idr() {664+ let track_info = Track {665+ name: MOQ_VIDEO_H264_TRACK.to_owned(),666+ priority: MOQ_VIDEO_TRACK_PRIORITY,667+ };668+ let mut broadcast = Broadcast::new().produce();669+ let track = broadcast.create_track(track_info.clone()).unwrap();670+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();671+ let mut publisher = MoqGroupPublisher::new(track);672+ let config = moq_test_config();673+674+ publisher675+ .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)676+ .unwrap();677+ let mut cancelled = consumer.recv_group().await.unwrap().unwrap();678+ assert_eq!(publisher.request_keyframe(), Some(0));679+ assert!(cancelled.finished().await.is_err());680+ assert_eq!(681+ publisher682+ .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)683+ .unwrap(),684+ MoqGroupDecision::SkipUntilKeyframe685+ );686+ assert_eq!(687+ publisher688+ .publish(&config, &media_v3_encoded_frame(20, true, true, 1), false)689+ .unwrap(),690+ MoqGroupDecision::Published {691+ group_id: 1,692+ frame_id: 0,693+ cancelled_previous: false,694+ }695+ );696+ }697+698+ 699+ async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {700+ let track_info = Track {701+ name: MOQ_VIDEO_H264_TRACK.to_owned(),702+ priority: MOQ_VIDEO_TRACK_PRIORITY,703+ };704+ let mut broadcast = Broadcast::new().produce();705+ let track = broadcast.create_track(track_info.clone()).unwrap();706+ let mut consumer = broadcast.consume().subscribe_track(&track_info).unwrap();707+ let mut publisher = MoqGroupPublisher::new(track);708+ let config = moq_test_config();709+ publisher710+ .publish(&config, &media_v3_encoded_frame(10, true, true, 1), false)711+ .unwrap();712+ let mut cancelled = consumer.recv_group().await.unwrap().unwrap();713+714+ // Payload-only accounting would accept this next one-byte access unit,715+ // but its fixed application envelope would overflow moq-net's 32 MiB716+ // group cache and silently evict the keyframe.717+ publisher.object_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;718+ assert_eq!(719+ publisher720+ .publish(&config, &media_v3_encoded_frame(11, false, false, 1), false,)721+ .unwrap(),722+ MoqGroupDecision::EnterResync723+ );724+ assert!(cancelled.finished().await.is_err());725+ assert_eq!(publisher.object_bytes, 0);726+ }727+}