- commit
- 5f6d0fddb14aaaf361514e86b3d7bfc46b11f411
- parent
- 8b97c3bf7681f3c31a9693b03912e8de024a0145
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 1443 | +----------------------------- |
| A | crates/ | 1356 | ++++++++++++++++++++++++++++ |
@@ -40,17 +40,6 @@ use crate::source::{
4040 const MEDIA_CAPABILITIES: &[Capability] = &[Capability::VideoH264];4141 const AUDIO_CAPABILITIES: &[Capability] = &[Capability::AudioOpus];4242 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);43-const MEDIA_V3_IN_FLIGHT_CAPACITY: usize = 4;44-const MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY: u8 = 0;45-const MEDIA_V3_DELTA_PUBLISHER_PRIORITY: u8 = 128;46-const MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY: i32 = 10;47-const MEDIA_V3_DELTA_TRANSPORT_PRIORITY: i32 = 0;48-// Wire-visible application reset code. Keep stable across compatible v3 releases.49-const MEDIA_V3_RESET_CODE: u32 = 0x5357;50-const MEDIA_V3_DELIVERY_FRAME_PERIODS: u64 = 4;51-const MEDIA_V3_MAX_CONTROL_REQUESTS_PER_SECOND: u32 = 10;52-const MEDIA_V3_CONTROL_REQUEST_INTERVAL: Duration =53- Duration::from_millis(1_000 / MEDIA_V3_MAX_CONTROL_REQUESTS_PER_SECOND as u64);5443 const INPUT_ACK_TIMEOUT: Duration = Duration::from_secs(1);5544 const REJECTION_DRAIN_TIMEOUT: Duration = Duration::from_secs(1);5645 const MAX_PENDING_HANDSHAKES: usize = 4;@@ -64,6 +53,7 @@ const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);
6453 6554 mod adaptive;6655 mod media_v2;56+mod media_v3;6757 mod session;6858 mod startup;6959 @@ -72,6 +62,9 @@ pub(crate) use adaptive::MotionResolutionPolicy;
7262 pub(crate) use adaptive::VideoDimensions;7363 use adaptive::serve_media_feedback;7464 use media_v2::{media_frame_for_encoded, serve_media, serve_media_v2};65+use media_v3::{66+ MediaV3GroupCursor, forward_media_v3_control_requests, new_current_gop_frames, serve_media_v3,67+};7568 7669 pub use session::SessionRegistry;7770 use session::{@@ -101,104 +94,6 @@ enum MediaV3GroupDecision {
10194 EnterResync,10295 }10396 104-105-struct MediaV3GroupCursor {106- group_id: Option<u64>,107- last_sequence: Option<u64>,108- last_object_id: Option<u32>,109- payload_bytes: usize,110- waiting_for_keyframe: bool,111- discontinuity_pending: bool,112-}113-114-impl Default for MediaV3GroupCursor {115- fn default() -> Self {116- Self {117- group_id: None,118- last_sequence: None,119- last_object_id: None,120- payload_bytes: 0,121- waiting_for_keyframe: true,122- discontinuity_pending: false,123- }124- }125-}126-127-impl MediaV3GroupCursor {128- fn classify(&mut self, frame: &EncodedFrame) -> MediaV3GroupDecision {129- let independently_decodable = frame.keyframe && frame.codec_config;130- if independently_decodable {131- if frame.data.len() > MAX_MEDIA_GROUP_BYTES_V3 {132- self.enter_resync();133- return MediaV3GroupDecision::EnterResync;134- }135- let sequence_discontinuity = self136- .last_sequence137- .is_some_and(|last| last.checked_add(1) != Some(frame.sequence));138- let position = MediaV3ObjectPosition {139- group_id: frame.sequence,140- object_id: 0,141- discontinuity: self.discontinuity_pending || sequence_discontinuity,142- };143- self.group_id = Some(frame.sequence);144- self.last_sequence = Some(frame.sequence);145- self.last_object_id = Some(0);146- self.payload_bytes = frame.data.len();147- self.waiting_for_keyframe = false;148- self.discontinuity_pending = false;149- return MediaV3GroupDecision::Send(position);150- }151-152- if self.waiting_for_keyframe {153- return MediaV3GroupDecision::SkipUntilKeyframe;154- }155- if frame.keyframe || frame.codec_config {156- self.enter_resync();157- return MediaV3GroupDecision::EnterResync;158- }159- let Some(group_id) = self.group_id else {160- self.enter_resync();161- return MediaV3GroupDecision::EnterResync;162- };163- let contiguous =164- self.last_sequence.and_then(|last| last.checked_add(1)) == Some(frame.sequence);165- let object_id = self.last_object_id.and_then(|last| last.checked_add(1));166- let payload_bytes = self.payload_bytes.checked_add(frame.data.len());167- let (Some(object_id), Some(payload_bytes)) = (object_id, payload_bytes) else {168- self.enter_resync();169- return MediaV3GroupDecision::EnterResync;170- };171- if !contiguous172- || object_id > MAX_MEDIA_OBJECT_ID_V3173- || payload_bytes > MAX_MEDIA_GROUP_BYTES_V3174- {175- self.enter_resync();176- return MediaV3GroupDecision::EnterResync;177- }178-179- self.last_sequence = Some(frame.sequence);180- self.last_object_id = Some(object_id);181- self.payload_bytes = payload_bytes;182- MediaV3GroupDecision::Send(MediaV3ObjectPosition {183- group_id,184- object_id,185- discontinuity: false,186- })187- }188-189- fn request_keyframe(&mut self) {190- self.enter_resync();191- }192-193- fn enter_resync(&mut self) {194- self.group_id = None;195- self.last_object_id = None;196- self.payload_bytes = 0;197- self.waiting_for_keyframe = true;198- self.discontinuity_pending = true;199- }200-}201-20297 20398 enum MoqGroupDecision {20499 Published {@@ -321,135 +216,6 @@ impl MoqGroupPublisher {
321216 }322217 }323218 324-325-enum MediaV3ScheduleDecision {326- Send {327- discontinuity: bool,328- cancel_sequences: Vec<u64>,329- },330- SkipUntilKeyframe,331- EnterResync {332- cancel_sequences: Vec<u64>,333- },334-}335-336-337-struct MediaV3Scheduler {338- in_flight: Vec<u64>,339- last_scheduled_sequence: Option<u64>,340- waiting_for_keyframe: bool,341- discontinuity_pending: bool,342-}343-344-impl Default for MediaV3Scheduler {345- fn default() -> Self {346- Self {347- in_flight: Vec::with_capacity(MEDIA_V3_IN_FLIGHT_CAPACITY),348- last_scheduled_sequence: None,349- waiting_for_keyframe: true,350- discontinuity_pending: false,351- }352- }353-}354-355-impl MediaV3Scheduler {356- fn schedule(357- &mut self,358- sequence: u64,359- independently_decodable: bool,360- ) -> MediaV3ScheduleDecision {361- if self362- .last_scheduled_sequence363- .is_some_and(|last| sequence <= last)364- {365- return MediaV3ScheduleDecision::SkipUntilKeyframe;366- }367-368- let sequence_discontinuity = self369- .last_scheduled_sequence370- .is_some_and(|last| last.checked_add(1) != Some(sequence));371- if independently_decodable {372- let cancel_sequences = std::mem::take(&mut self.in_flight);373- let discontinuity = self.discontinuity_pending374- || sequence_discontinuity375- || !cancel_sequences.is_empty();376- self.in_flight.push(sequence);377- self.last_scheduled_sequence = Some(sequence);378- self.waiting_for_keyframe = false;379- self.discontinuity_pending = false;380- return MediaV3ScheduleDecision::Send {381- discontinuity,382- cancel_sequences,383- };384- }385-386- if self.waiting_for_keyframe {387- return MediaV3ScheduleDecision::SkipUntilKeyframe;388- }389- if sequence_discontinuity || self.in_flight.len() == MEDIA_V3_IN_FLIGHT_CAPACITY {390- return MediaV3ScheduleDecision::EnterResync {391- cancel_sequences: self.enter_resync(),392- };393- }394-395- self.in_flight.push(sequence);396- self.last_scheduled_sequence = Some(sequence);397- MediaV3ScheduleDecision::Send {398- discontinuity: false,399- cancel_sequences: Vec::new(),400- }401- }402-403- fn complete(&mut self, sequence: u64) {404- if let Some(index) = self.in_flight.iter().position(|value| *value == sequence) {405- self.in_flight.swap_remove(index);406- }407- }408-409- fn fail(&mut self, sequence: u64) -> Option<Vec<u64>> {410- let index = self.in_flight.iter().position(|value| *value == sequence)?;411- self.in_flight.swap_remove(index);412- Some(self.enter_resync())413- }414-415- fn fail_all(&mut self) -> Vec<u64> {416- self.enter_resync()417- }418-419- fn request_keyframe(&mut self) -> Vec<u64> {420- if self.waiting_for_keyframe {421- self.discontinuity_pending = true;422- return Vec::new();423- }424- self.enter_resync()425- }426-427- fn enter_resync(&mut self) -> Vec<u64> {428- self.waiting_for_keyframe = true;429- self.discontinuity_pending = true;430- std::mem::take(&mut self.in_flight)431- }432-}433-434-fn apply_media_v3_keyframe_request(435- scheduler: &mut MediaV3Scheduler,436- group_cursor: &mut MediaV3GroupCursor,437- replay_cursor: &mut MediaReplayCursor,438- through_sequence: Option<u64>,439- reason: KeyframeRequestReasonV3,440-) -> (bool, Vec<u64>) {441- // Every v3 session already begins by replaying the bounded current GOP442- // from object zero. A Join arriving after that replay was scheduled must443- // not cancel the only decodable image on a damage-driven static source.444- if reason == KeyframeRequestReasonV3::Join || scheduler.waiting_for_keyframe {445- return (false, Vec::new());446- }447- let cancel_sequences = scheduler.request_keyframe();448- group_cursor.request_keyframe();449- replay_cursor.enter_resync_through(through_sequence);450- (true, cancel_sequences)451-}452-453219 fn apply_moq_keyframe_request(454220 publisher: &mut MoqGroupPublisher,455221 replay_cursor: &mut MediaReplayCursor,@@ -466,116 +232,6 @@ fn apply_moq_keyframe_request(
466232 cancelled_group467233 }468234 469-fn apply_media_v3_send_failure(470- scheduler: &mut MediaV3Scheduler,471- group_cursor: &mut MediaV3GroupCursor,472- replay_cursor: &mut MediaReplayCursor,473- sequence: u64,474- through_sequence: Option<u64>,475-) -> Option<Vec<u64>> {476- let cancel_sequences = scheduler.fail(sequence)?;477- group_cursor.request_keyframe();478- replay_cursor.enter_resync_through(through_sequence);479- Some(cancel_sequences)480-}481-482-483-enum MediaV3ControlDecision {484- Accept,485- Pace(Duration),486-}487-488-489-struct MediaV3ControlGate {490- last_request_id: Option<u64>,491- last_accepted_at: Option<Instant>,492-}493-494-impl MediaV3ControlGate {495- fn accept(496- &mut self,497- request: MediaControlRequestV3,498- now: Instant,499- ) -> Result<MediaV3ControlDecision> {500- ensure!(501- self.last_request_id502- .is_none_or(|last| request.request_id > last),503- "v3 media control request IDs must be strictly increasing"504- );505- self.last_request_id = Some(request.request_id);506- if let Some(last) = self.last_accepted_at {507- let elapsed = now.saturating_duration_since(last);508- if elapsed < MEDIA_V3_CONTROL_REQUEST_INTERVAL {509- return Ok(MediaV3ControlDecision::Pace(510- MEDIA_V3_CONTROL_REQUEST_INTERVAL - elapsed,511- ));512- }513- }514- self.last_accepted_at = Some(now);515- Ok(MediaV3ControlDecision::Accept)516- }517-}518-519-async fn forward_media_v3_control_requests<R>(520- mut reader: R,521- sender: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,522-) -> Result<()>523-where524- R: tokio::io::AsyncRead + Unpin,525-{526- let mut gate = MediaV3ControlGate::default();527- while let Some(request) = read_media_control_request_v3(&mut reader).await? {528- match gate.accept(request, Instant::now())? {529- MediaV3ControlDecision::Accept => {}530- MediaV3ControlDecision::Pace(retry_after) => {531- // Stop reading during the rejection interval so QUIC flow532- // control, rather than this task, absorbs an abusive burst.533- // This also bounds rejection logging to the configured rate.534- debug!(535- request_id = request.request_id,536- retry_after_ms = retry_after.as_millis(),537- "paced rate-limited v3 keyframe request"538- );539- tokio::time::sleep(retry_after).await;540- continue;541- }542- }543- sender.send_replace(Some(request));544- if sender.is_closed() {545- return Ok(());546- }547- }548- Ok(())549-}550-551-struct ResetOnDropSendStreamV3(Option<SendStream>);552-553-impl ResetOnDropSendStreamV3 {554- fn new(stream: SendStream) -> Self {555- Self(Some(stream))556- }557-558- fn stream(&self) -> &SendStream {559- self.0.as_ref().expect("send stream guard is armed")560- }561-562- fn stream_mut(&mut self) -> &mut SendStream {563- self.0.as_mut().expect("send stream guard is armed")564- }565-566- fn disarm(&mut self) {567- self.0.take();568- }569-}570-571-impl Drop for ResetOnDropSendStreamV3 {572- fn drop(&mut self) {573- if let Some(stream) = self.0.as_mut() {574- let _ = stream.reset(MEDIA_V3_RESET_CODE.into());575- }576- }577-}578-579235 580236 struct MediaReplayCursor {581237 last_sequence: Option<u64>,@@ -1249,11 +905,11 @@ async fn run_control_moq_session(
1249905 result1250906 }1251907 1252-async fn serve_media_v3(908+async fn serve_input(1253909 connection: Connection,1254- config: HostConfig,910+ backend: &InputBackend,911+ pointer_positions: Option<&PointerPositionTracker>,1255912 sessions: &Arc<SessionRegistry>,1256- authorization: &AuthorizationPolicy,1257913 ) -> Result<()> {1258914 let remote = connection.remote_id();1259915 let handshake_permit = sessions@@ -1262,585 +918,38 @@ async fn serve_media_v3(
1262918 .context("too many pending handshakes")?;1263919 let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1264920 .await1265- .context("timed out accepting media v3 handshake stream")?1266- .context("accepting media v3 handshake stream")?;1267- let hello = receive_hello(&mut recv, Capability::VideoH264).await?;921+ .context("timed out accepting input stream")?922+ .context("accepting input stream")?;923+ let hello = receive_hello_unconstrained(&mut recv).await?;1268924 drop(handshake_permit);1269- debug!(%remote, agent = %hello.agent, "media v3 hello received");1270-1271- let grants = match authorization.authorize_or_redeem(1272- remote,1273- hello.invitation.as_deref(),1274- unix_timestamp_now()?,1275- ) {1276- Ok(grants) => grants,1277- Err(error) => {1278- send_rejection(&mut send, "Portal peer is not authorized").await?;1279- return Err(error.context("authorizing media v3 peer"));1280- }1281- };925+ debug!(%remote, agent = %hello.agent, "input hello received");1282926 ensure!(1283- grants.contains(InvitationGrants::VIEW),1284- "authorized media peer lacks view permission"927+ hello.invitation.is_none(),928+ "invitations are accepted only on the first media handshake"1285929 );1286-1287- let lease = match sessions.claim(remote, hello.nonce, grants) {930+ let lease = match sessions.claim_input(remote, hello.nonce) {1288931 Ok(lease) => lease,1289932 Err(error) => {1290- send_rejection(&mut send, "host already has an active client").await?;1291- return Err(error);1292- }1293- };1294-1295- let source = match config.source {1296- VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),1297- VideoSource::GamescopePipewire => {1298- let primary = spawn_gamescope_pipewire_after_static_preflight(1299- config.clone(),1300- lease.session_clock,1301- )1302- .await?;1303- select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await1304- }1305- };1306- let EncodedSource {1307- frames: frame_receiver,1308- current_gop: mut current_gop_receiver,1309- task: source_task,1310- pointer_surface_dimensions,1311- encoder_control,1312- } = match source {1313- Ok(source) => source,1314- Err(error) => {1315- send_rejection(&mut send, "video source is unavailable").await?;933+ send_rejection(&mut send, error.to_string()).await?;1316934 return Err(error);1317935 }1318936 };1319- let source_task = SourceTaskGuard::new(source_task);1320- sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;1321-1322- let mut media_hello = HostHello::accepted(1323- lease.session_id,1324- negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1325- );1326- if let Some(dimensions) = pointer_surface_dimensions {1327- media_hello = media_hello.with_pointer_surface_dimensions(dimensions);1328- }1329- write_host_hello(&mut send, &media_hello).await?;1330- send.finish()1331- .context("finishing media v3 handshake response")?;1332- drop(send);1333- info!(%remote, session_id = lease.session_id, "media v3 client accepted");937+ let supported =938+ supported_input_capabilities(backend.capabilities(), pointer_positions.is_some());939+ let negotiated = negotiated_input_capabilities(&hello, &supported, lease.grants);1334940 1335- let session_result = run_media_v3_session(1336- &connection,1337- &config,1338- &mut current_gop_receiver,1339- recv,1340- remote,1341- encoder_control,1342- Arc::clone(&lease.media_v3_telemetry),941+ let ack_enabled = negotiated.contains(&Capability::InputAck);942+ let feedback_enabled = negotiated.contains(&Capability::PointerPositionFeedback);943+ let visibility_feedback_enabled = negotiated.contains(&Capability::PointerVisibilityFeedback);944+ let mut pointer_positions = pointer_positions945+ .filter(|_| feedback_enabled)946+ .map(PointerPositionTracker::subscribe);947+ write_host_hello(948+ &mut send,949+ &HostHello::accepted(lease.session_id, negotiated.clone()),1343950 )1344- .await;1345-1346- drop(current_gop_receiver);1347- drop(frame_receiver);1348- source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;1349- drop(lease);1350- info!(%remote, "media v3 client released");1351- session_result1352-}1353-1354-async fn run_media_v3_session(1355- connection: &Connection,1356- config: &HostConfig,1357- current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,1358- control_recv: iroh::endpoint::RecvStream,1359- remote: EndpointId,1360- encoder_control: Option<EncoderControl>,1361- telemetry: Arc<MediaV3Telemetry>,1362-) -> Result<()> {1363- let maximum_replay_age = maximum_media_replay_age(config.framerate);1364- let delivery_timeout_ms = media_v3_delivery_timeout_ms(config.framerate);1365- let mut replay_cursor = MediaReplayCursor::default();1366- let mut group_cursor = MediaV3GroupCursor::default();1367- let mut scheduler = MediaV3Scheduler::default();1368- let mut send_tasks = tokio::task::JoinSet::new();1369- let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);1370- let mut control_task = tokio::spawn(forward_media_v3_control_requests(1371- control_recv,1372- control_sender,1373- ));1374- let mut control_task_finished = false;1375- let mut control_receiver_open = true;1376- let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));1377-1378- let result = loop {1379- telemetry.record_selected_path(connection);1380- tokio::select! {1381- biased;1382- control_result = &mut control_task, if !control_task_finished => {1383- control_task_finished = true;1384- match control_result {1385- Ok(Ok(())) => {1386- // Keep polling the watch receiver after clean EOF so a1387- // final request published immediately before the1388- // sender closed cannot be lost.1389- debug!(%remote, "media v3 control stream closed");1390- }1391- Ok(Err(error)) => {1392- break Err(error).context("reading media v3 control stream");1393- }1394- Err(error) => {1395- break Err(error).context("media v3 control task failed");1396- }1397- }1398- }1399- changed = control_requests.changed(), if control_receiver_open => {1400- if changed.is_err() {1401- control_receiver_open = false;1402- continue;1403- }1404- let request = *control_requests.borrow_and_update();1405- let Some(request) = request else {1406- continue;1407- };1408- let through_sequence = current_gop_receiver1409- .borrow()1410- .as_ref()1411- .and_then(|gop| gop.frames.last())1412- .map(|frame| frame.sequence);1413- let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(1414- &mut scheduler,1415- &mut group_cursor,1416- &mut replay_cursor,1417- through_sequence,1418- request.reason,1419- );1420- if !cancel_sequences.is_empty() {1421- telemetry.scheduler_cancellations.fetch_add(1422- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1423- Ordering::Relaxed,1424- );1425- send_tasks.abort_all();1426- }1427- let forced_idr_disposition = forced_idr.request(request.reason);1428- if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {1429- warn!(1430- %remote,1431- request_id = request.request_id,1432- ?request.reason,1433- %error,1434- "forced-IDR request failed; retaining natural-IDR fallback"1435- );1436- }1437- debug!(1438- %remote,1439- request_id = request.request_id,1440- ?request.reason,1441- advisory_last_sequence = ?request.last_sequence,1442- coalesced = !transitioned,1443- ?cancel_sequences,1444- ?forced_idr_disposition,1445- "accepted media v3 keyframe request"1446- );1447- }1448- acknowledgement = forced_idr.acknowledgements.join_next(),1449- if forced_idr.pending_revision.is_some() =>1450- {1451- forced_idr.complete(acknowledgement, remote, "grouped-v3");1452- }1453- closed = connection.closed() => {1454- debug!(%remote, ?closed, "media v3 connection closed");1455- break Ok(());1456- }1457- task = send_tasks.join_next(), if !send_tasks.is_empty() => {1458- match task.expect("guarded by non-empty send task set") {1459- Ok((sequence, Ok(()))) => scheduler.complete(sequence),1460- Ok((sequence, Err(error))) => {1461- let through_sequence = current_gop_receiver1462- .borrow()1463- .as_ref()1464- .and_then(|gop| gop.frames.last())1465- .map(|frame| frame.sequence);1466- if let Some(cancel_sequences) = apply_media_v3_send_failure(1467- &mut scheduler,1468- &mut group_cursor,1469- &mut replay_cursor,1470- sequence,1471- through_sequence,1472- ) {1473- telemetry.send_failures.fetch_add(1, Ordering::Relaxed);1474- telemetry.scheduler_cancellations.fetch_add(1475- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1476- Ordering::Relaxed,1477- );1478- warn!(1479- sequence,1480- %error,1481- "media v3 object send failed; waiting for keyframe"1482- );1483- if !cancel_sequences.is_empty() {1484- send_tasks.abort_all();1485- }1486- } else {1487- debug!(1488- sequence,1489- %error,1490- "ignored stale media v3 object failure from a superseded group"1491- );1492- }1493- }1494- Err(error) if error.is_cancelled() => {}1495- Err(error) => {1496- warn!(%error, "media v3 object task failed; waiting for keyframe");1497- let cancel_sequences = scheduler.fail_all();1498- telemetry.send_failures.fetch_add(1, Ordering::Relaxed);1499- telemetry.scheduler_cancellations.fetch_add(1500- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1501- Ordering::Relaxed,1502- );1503- group_cursor.request_keyframe();1504- let through_sequence = current_gop_receiver1505- .borrow()1506- .as_ref()1507- .and_then(|gop| gop.frames.last())1508- .map(|frame| frame.sequence);1509- replay_cursor.enter_resync_through(through_sequence);1510- if !cancel_sequences.is_empty() {1511- send_tasks.abort_all();1512- }1513- }1514- }1515- }1516- changed = current_gop_receiver.changed() => {1517- if let Err(error) = changed {1518- break Err(error).context("encoded source stopped");1519- }1520- let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {1521- continue;1522- };1523- let initial_replay_started_at =1524- replay_cursor.last_sequence.is_none().then(Instant::now);1525- let replay_through_sequence = current_gop1526- .frames1527- .last()1528- .map(|frame| frame.sequence)1529- .context("current GOP snapshot is empty")?;1530-1531- for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {1532- let replay_discontinuity = match replay_cursor.classify(1533- &frame,1534- replay_through_sequence,1535- initial_replay_started_at,1536- Instant::now(),1537- maximum_replay_age,1538- ) {1539- MediaReplayDecision::Send { discontinuity } => discontinuity,1540- MediaReplayDecision::SkipUntilKeyframe => {1541- replay_cursor.enter_resync_through(Some(replay_through_sequence));1542- break;1543- }1544- MediaReplayDecision::DiscardStaleSuffix { .. } => {1545- group_cursor.request_keyframe();1546- let cancel_sequences = scheduler.fail_all();1547- telemetry.scheduler_cancellations.fetch_add(1548- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1549- Ordering::Relaxed,1550- );1551- if !cancel_sequences.is_empty() {1552- send_tasks.abort_all();1553- }1554- break;1555- }1556- };1557- let position = match group_cursor.classify(&frame) {1558- MediaV3GroupDecision::Send(position) => position,1559- MediaV3GroupDecision::SkipUntilKeyframe => {1560- replay_cursor.enter_resync_through(Some(replay_through_sequence));1561- break;1562- }1563- MediaV3GroupDecision::EnterResync => {1564- replay_cursor.enter_resync_through(Some(replay_through_sequence));1565- let cancel_sequences = scheduler.fail_all();1566- telemetry.scheduler_cancellations.fetch_add(1567- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1568- Ordering::Relaxed,1569- );1570- if !cancel_sequences.is_empty() {1571- send_tasks.abort_all();1572- }1573- break;1574- }1575- };1576- let independently_decodable = position.object_id == 0;1577- let (scheduler_discontinuity, cancel_sequences) =1578- match scheduler.schedule(frame.sequence, independently_decodable) {1579- MediaV3ScheduleDecision::Send {1580- discontinuity,1581- cancel_sequences,1582- } => (discontinuity, cancel_sequences),1583- MediaV3ScheduleDecision::SkipUntilKeyframe => continue,1584- MediaV3ScheduleDecision::EnterResync { cancel_sequences } => {1585- group_cursor.request_keyframe();1586- replay_cursor.enter_resync_through(Some(replay_through_sequence));1587- telemetry.scheduler_cancellations.fetch_add(1588- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1589- Ordering::Relaxed,1590- );1591- if !cancel_sequences.is_empty() {1592- send_tasks.abort_all();1593- }1594- break;1595- }1596- };1597- if !cancel_sequences.is_empty() {1598- telemetry.scheduler_cancellations.fetch_add(1599- u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),1600- Ordering::Relaxed,1601- );1602- debug!(1603- sequence = frame.sequence,1604- group_id = position.group_id,1605- ?cancel_sequences,1606- "configured keyframe superseding media v3 objects"1607- );1608- send_tasks.abort_all();1609- }1610-1611- let media_object = media_v3_object_for_encoded(1612- config,1613- &frame,1614- position,1615- replay_discontinuity1616- || position.discontinuity1617- || scheduler_discontinuity,1618- delivery_timeout_ms,1619- )?;1620- let sequence = frame.sequence;1621- let stream = match tokio::time::timeout(1622- Duration::from_millis(u64::from(delivery_timeout_ms)),1623- connection.open_uni(),1624- )1625- .await1626- {1627- Ok(Ok(stream)) => stream,1628- Ok(Err(error)) => {1629- telemetry.send_failures.fetch_add(1, Ordering::Relaxed);1630- warn!(sequence, %error, "opening media v3 object stream failed");1631- if let Some(cancel_sequences) = apply_media_v3_send_failure(1632- &mut scheduler,1633- &mut group_cursor,1634- &mut replay_cursor,1635- sequence,1636- Some(replay_through_sequence),1637- ) && !cancel_sequences.is_empty()1638- {1639- send_tasks.abort_all();1640- }1641- break;1642- }1643- Err(_) => {1644- telemetry.send_failures.fetch_add(1, Ordering::Relaxed);1645- warn!(sequence, "opening media v3 object stream timed out");1646- if let Some(cancel_sequences) = apply_media_v3_send_failure(1647- &mut scheduler,1648- &mut group_cursor,1649- &mut replay_cursor,1650- sequence,1651- Some(replay_through_sequence),1652- ) && !cancel_sequences.is_empty()1653- {1654- send_tasks.abort_all();1655- }1656- break;1657- }1658- };1659- let stream = ResetOnDropSendStreamV3::new(stream);1660- let transport_priority =1661- media_v3_transport_priority(media_object.header.publisher_priority)?;1662- if let Err(error) = stream.stream().set_priority(transport_priority) {1663- telemetry.send_failures.fetch_add(1, Ordering::Relaxed);1664- warn!(sequence, %error, "setting media v3 object priority failed");1665- if let Some(cancel_sequences) = apply_media_v3_send_failure(1666- &mut scheduler,1667- &mut group_cursor,1668- &mut replay_cursor,1669- sequence,1670- Some(replay_through_sequence),1671- ) && !cancel_sequences.is_empty()1672- {1673- send_tasks.abort_all();1674- }1675- break;1676- }1677- send_tasks.spawn(async move {1678- (1679- sequence,1680- send_media_v3_object(stream, media_object).await,1681- )1682- });1683- replay_cursor.commit_sent(&frame);1684- }1685- }1686- }1687- };1688-1689- forced_idr.abort_and_drain(remote, "grouped-v3").await;1690- send_tasks.abort_all();1691- while send_tasks.join_next().await.is_some() {}1692- if !control_task_finished {1693- control_task.abort();1694- let _ = control_task.await;1695- }1696- result1697-}1698-1699-fn media_v3_delivery_timeout_ms(framerate: u32) -> u32 {1700- debug_assert!(framerate > 0);1701- let timeout_ms = 1_000_u641702- .saturating_mul(MEDIA_V3_DELIVERY_FRAME_PERIODS)1703- .div_ceil(u64::from(framerate.max(1)));1704- u32::try_from(timeout_ms).unwrap_or(u32::MAX).clamp(1705- MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,1706- MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,1707- )1708-}1709-1710-fn media_v3_object_for_encoded(1711- _config: &HostConfig,1712- frame: &EncodedFrame,1713- position: MediaV3ObjectPosition,1714- discontinuity: bool,1715- delivery_timeout_ms: u32,1716-) -> Result<MediaObjectV3> {1717- let mut flags = FrameFlags::NONE;1718- if frame.keyframe {1719- flags = flags.union(FrameFlags::KEYFRAME);1720- }1721- if frame.codec_config {1722- flags = flags.union(FrameFlags::CODEC_CONFIG);1723- }1724- if discontinuity || frame.discontinuity {1725- flags = flags.union(FrameFlags::DISCONTINUITY);1726- }1727- let publisher_priority = if position.object_id == 0 {1728- MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY1729- } else {1730- MEDIA_V3_DELTA_PUBLISHER_PRIORITY1731- };1732- let header = MediaObjectHeaderV3::h264(1733- frame.width,1734- frame.height,1735- frame.data.len(),1736- publisher_priority,1737- flags,1738- position.object_id,1739- position.group_id,1740- frame.sequence,1741- frame.capture_timestamp_micros,1742- frame.presentation_timestamp_micros,1743- delivery_timeout_ms,1744- )?;1745- MediaObjectV3::new(header, frame.data.as_ref().to_vec()).map_err(Into::into)1746-}1747-1748-async fn send_media_v3_object(1749- mut stream: ResetOnDropSendStreamV3,1750- object: MediaObjectV3,1751-) -> Result<()> {1752- let delivery_timeout = Duration::from_millis(u64::from(object.header.delivery_timeout_ms));1753- tokio::time::timeout(delivery_timeout, async {1754- write_media_object_v3(stream.stream_mut(), &object)1755- .await1756- .context("writing media v3 object")?;1757- stream1758- .stream_mut()1759- .finish()1760- .context("finishing media v3 object stream")?;1761- match stream1762- .stream()1763- .stopped()1764- .await1765- .context("waiting for media v3 object acknowledgement")?1766- {1767- None => {1768- stream.disarm();1769- Ok(())1770- }1771- Some(code) => bail!("peer stopped media v3 object stream with code {code}"),1772- }1773- })1774- .await1775- .context("media v3 object exceeded its delivery timeout")?1776-}1777-1778-fn media_v3_transport_priority(publisher_priority: u8) -> Result<i32> {1779- // V3 follows MoQ's lower-is-higher publisher priority, while Iroh/QUIC1780- // uses larger integers for more important streams. Keep the inversion1781- // explicit rather than leaking either convention across the boundary.1782- match publisher_priority {1783- MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY => Ok(MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY),1784- MEDIA_V3_DELTA_PUBLISHER_PRIORITY => Ok(MEDIA_V3_DELTA_TRANSPORT_PRIORITY),1785- _ => bail!("unsupported media v3 publisher priority {publisher_priority}"),1786- }1787-}1788-1789-fn new_current_gop_frames(1790- current_gop: EncodedGop,1791- last_sequence: Option<u64>,1792-) -> impl Iterator<Item = EncodedFrame> {1793- current_gop1794- .frames1795- .into_iter()1796- .skip_while(move |frame| last_sequence.is_some_and(|last| frame.sequence <= last))1797-}1798-1799-async fn serve_input(1800- connection: Connection,1801- backend: &InputBackend,1802- pointer_positions: Option<&PointerPositionTracker>,1803- sessions: &Arc<SessionRegistry>,1804-) -> Result<()> {1805- let remote = connection.remote_id();1806- let handshake_permit = sessions1807- .pending_handshakes1808- .try_acquire()1809- .context("too many pending handshakes")?;1810- let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1811- .await1812- .context("timed out accepting input stream")?1813- .context("accepting input stream")?;1814- let hello = receive_hello_unconstrained(&mut recv).await?;1815- drop(handshake_permit);1816- debug!(%remote, agent = %hello.agent, "input hello received");1817- ensure!(1818- hello.invitation.is_none(),1819- "invitations are accepted only on the first media handshake"1820- );1821- let lease = match sessions.claim_input(remote, hello.nonce) {1822- Ok(lease) => lease,1823- Err(error) => {1824- send_rejection(&mut send, error.to_string()).await?;1825- return Err(error);1826- }1827- };1828- let supported =1829- supported_input_capabilities(backend.capabilities(), pointer_positions.is_some());1830- let negotiated = negotiated_input_capabilities(&hello, &supported, lease.grants);1831-1832- let ack_enabled = negotiated.contains(&Capability::InputAck);1833- let feedback_enabled = negotiated.contains(&Capability::PointerPositionFeedback);1834- let visibility_feedback_enabled = negotiated.contains(&Capability::PointerVisibilityFeedback);1835- let mut pointer_positions = pointer_positions1836- .filter(|_| feedback_enabled)1837- .map(PointerPositionTracker::subscribe);1838- write_host_hello(1839- &mut send,1840- &HostHello::accepted(lease.session_id, negotiated.clone()),1841- )1842- .await?;1843- info!(%remote, session_id = lease.session_id, "input client accepted");951+ .await?;952+ info!(%remote, session_id = lease.session_id, "input client accepted");1844953 1845954 let session_result: Result<()> = async {1846955 let mut received_events = 0_u64;@@ -2225,6 +1334,27 @@ fn moq_test_config() -> HostConfig {
22251334 }22261335 }22271336 1337+1338+fn media_v3_encoded_frame(1339+ sequence: u64,1340+ keyframe: bool,1341+ codec_config: bool,1342+ payload_len: usize,1343+) -> EncodedFrame {1344+ EncodedFrame {1345+ sequence,1346+ width: 1_280,1347+ height: 800,1348+ capture_timestamp_micros: sequence,1349+ presentation_timestamp_micros: sequence as i64,1350+ observed_at: Instant::now(),1351+ keyframe,1352+ codec_config,1353+ discontinuity: false,1354+ data: Arc::from(vec![sequence as u8; payload_len]),1355+ }1356+}1357+22281358 22291359 struct DropNotify(Option<tokio::sync::oneshot::Sender<()>>);22301360 @@ -2241,61 +1371,6 @@ impl Drop for DropNotify {
22411371 mod tests {22421372 use super::*;22431373 2244- fn media_v3_encoded_frame(2245- sequence: u64,2246- keyframe: bool,2247- codec_config: bool,2248- payload_len: usize,2249- ) -> EncodedFrame {2250- EncodedFrame {2251- sequence,2252- width: 1_280,2253- height: 800,2254- capture_timestamp_micros: sequence,2255- presentation_timestamp_micros: sequence as i64,2256- observed_at: Instant::now(),2257- keyframe,2258- codec_config,2259- discontinuity: false,2260- data: Arc::from(vec![sequence as u8; payload_len]),2261- }2262- }2263-2264- 2265- fn media_v3_groups_begin_at_configured_idr_and_assign_contiguous_objects() {2266- let mut cursor = MediaV3GroupCursor::default();2267- assert_eq!(2268- cursor.classify(&media_v3_encoded_frame(10, true, true, 4)),2269- MediaV3GroupDecision::Send(MediaV3ObjectPosition {2270- group_id: 10,2271- object_id: 0,2272- discontinuity: false,2273- })2274- );2275- assert_eq!(2276- cursor.classify(&media_v3_encoded_frame(11, false, false, 3)),2277- MediaV3GroupDecision::Send(MediaV3ObjectPosition {2278- group_id: 10,2279- object_id: 1,2280- discontinuity: false,2281- })2282- );2283-2284- cursor.request_keyframe();2285- assert_eq!(2286- cursor.classify(&media_v3_encoded_frame(12, false, false, 3)),2287- MediaV3GroupDecision::SkipUntilKeyframe2288- );2289- assert_eq!(2290- cursor.classify(&media_v3_encoded_frame(20, true, true, 4)),2291- MediaV3GroupDecision::Send(MediaV3ObjectPosition {2292- group_id: 20,2293- object_id: 0,2294- discontinuity: true,2295- })2296- );2297- }2298-22991374 23001375 async fn upstream_moq_groups_are_sequential_and_cancel_the_superseded_gop() {23011376 let track_info = Track {@@ -2470,378 +1545,6 @@ mod tests {
24701545 assert_eq!(publisher.object_bytes, 0);24711546 }24721547 2473- 2474- fn media_v3_group_rejects_noncontiguous_or_unconfigured_frames() {2475- let mut cursor = MediaV3GroupCursor::default();2476- assert_eq!(2477- cursor.classify(&media_v3_encoded_frame(10, true, false, 1)),2478- MediaV3GroupDecision::SkipUntilKeyframe2479- );2480- assert!(matches!(2481- cursor.classify(&media_v3_encoded_frame(20, true, true, 1)),2482- MediaV3GroupDecision::Send(_)2483- ));2484- assert_eq!(2485- cursor.classify(&media_v3_encoded_frame(22, false, false, 1)),2486- MediaV3GroupDecision::EnterResync2487- );2488- assert!(cursor.waiting_for_keyframe);2489- }2490-2491- 2492- fn media_v3_group_accepts_exact_limits_and_rejects_overflow() {2493- let mut object_cursor = MediaV3GroupCursor::default();2494- assert!(matches!(2495- object_cursor.classify(&media_v3_encoded_frame(0, true, true, 1)),2496- MediaV3GroupDecision::Send(_)2497- ));2498- for sequence in 1..=u64::from(MAX_MEDIA_OBJECT_ID_V3) {2499- assert!(matches!(2500- object_cursor.classify(&media_v3_encoded_frame(sequence, false, false, 1)),2501- MediaV3GroupDecision::Send(MediaV3ObjectPosition { object_id, .. })2502- if u64::from(object_id) == sequence2503- ));2504- }2505- assert_eq!(2506- object_cursor.classify(&media_v3_encoded_frame(2507- u64::from(MAX_MEDIA_OBJECT_ID_V3) + 1,2508- false,2509- false,2510- 1,2511- )),2512- MediaV3GroupDecision::EnterResync2513- );2514-2515- let mut byte_cursor = MediaV3GroupCursor::default();2516- assert!(matches!(2517- byte_cursor.classify(&media_v3_encoded_frame(10, true, true, 1)),2518- MediaV3GroupDecision::Send(_)2519- ));2520- byte_cursor.payload_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;2521- assert!(matches!(2522- byte_cursor.classify(&media_v3_encoded_frame(11, false, false, 1)),2523- MediaV3GroupDecision::Send(_)2524- ));2525- assert_eq!(byte_cursor.payload_bytes, MAX_MEDIA_GROUP_BYTES_V3);2526- assert_eq!(2527- byte_cursor.classify(&media_v3_encoded_frame(12, false, false, 1)),2528- MediaV3GroupDecision::EnterResync2529- );2530- }2531-2532- 2533- fn media_v3_keyframe_requests_cancel_once_and_recover_discontinuously() {2534- let mut scheduler = MediaV3Scheduler::default();2535- assert!(matches!(2536- scheduler.schedule(10, true),2537- MediaV3ScheduleDecision::Send { .. }2538- ));2539- assert!(matches!(2540- scheduler.schedule(11, false),2541- MediaV3ScheduleDecision::Send { .. }2542- ));2543- assert_eq!(scheduler.request_keyframe(), vec![10, 11]);2544- assert!(scheduler.request_keyframe().is_empty());2545- assert_eq!(2546- scheduler.schedule(12, false),2547- MediaV3ScheduleDecision::SkipUntilKeyframe2548- );2549- assert_eq!(2550- scheduler.schedule(20, true),2551- MediaV3ScheduleDecision::Send {2552- discontinuity: true,2553- cancel_sequences: vec![],2554- }2555- );2556- }2557-2558- 2559- fn media_v3_join_request_preserves_initial_current_gop_replay() {2560- let mut scheduler = MediaV3Scheduler::default();2561- let mut group_cursor = MediaV3GroupCursor::default();2562- let mut replay_cursor = MediaReplayCursor::default();2563-2564- let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(2565- &mut scheduler,2566- &mut group_cursor,2567- &mut replay_cursor,2568- Some(42),2569- KeyframeRequestReasonV3::Join,2570- );2571-2572- assert!(!transitioned);2573- assert!(cancel_sequences.is_empty());2574- assert_eq!(replay_cursor.last_sequence, None);2575- assert!(replay_cursor.waiting_for_keyframe);2576- assert!(!replay_cursor.discontinuity_pending);2577- assert!(group_cursor.waiting_for_keyframe);2578- assert!(!group_cursor.discontinuity_pending);2579- assert!(!scheduler.discontinuity_pending);2580-2581- assert!(matches!(2582- group_cursor.classify(&media_v3_encoded_frame(10, true, true, 1)),2583- MediaV3GroupDecision::Send(MediaV3ObjectPosition {2584- discontinuity: false,2585- ..2586- })2587- ));2588- assert!(matches!(2589- scheduler.schedule(10, true),2590- MediaV3ScheduleDecision::Send {2591- discontinuity: false,2592- ..2593- }2594- ));2595- }2596-2597- 2598- fn media_v3_late_join_preserves_active_static_current_gop() {2599- let mut scheduler = MediaV3Scheduler::default();2600- let mut group_cursor = MediaV3GroupCursor::default();2601- let mut replay_cursor = MediaReplayCursor::default();2602- let keyframe = media_v3_encoded_frame(10, true, true, 1);2603- let delta = media_v3_encoded_frame(11, false, false, 1);2604-2605- assert!(matches!(2606- group_cursor.classify(&keyframe),2607- MediaV3GroupDecision::Send(_)2608- ));2609- assert!(matches!(2610- scheduler.schedule(keyframe.sequence, true),2611- MediaV3ScheduleDecision::Send { .. }2612- ));2613- replay_cursor.commit_sent(&keyframe);2614- assert!(matches!(2615- group_cursor.classify(&delta),2616- MediaV3GroupDecision::Send(_)2617- ));2618- assert!(matches!(2619- scheduler.schedule(delta.sequence, false),2620- MediaV3ScheduleDecision::Send { .. }2621- ));2622- replay_cursor.commit_sent(&delta);2623-2624- let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(2625- &mut scheduler,2626- &mut group_cursor,2627- &mut replay_cursor,2628- Some(delta.sequence),2629- KeyframeRequestReasonV3::Join,2630- );2631-2632- assert!(!transitioned);2633- assert!(cancel_sequences.is_empty());2634- assert_eq!(scheduler.in_flight, vec![10, 11]);2635- assert!(!scheduler.waiting_for_keyframe);2636- assert!(!scheduler.discontinuity_pending);2637- assert_eq!(group_cursor.group_id, Some(10));2638- assert_eq!(group_cursor.last_sequence, Some(11));2639- assert!(!group_cursor.waiting_for_keyframe);2640- assert_eq!(replay_cursor.last_sequence, Some(11));2641- assert!(!replay_cursor.waiting_for_keyframe);2642- }2643-2644- 2645- fn media_v3_stale_old_group_failure_preserves_replacement_group() {2646- let mut scheduler = MediaV3Scheduler::default();2647- let mut group_cursor = MediaV3GroupCursor::default();2648- let mut replay_cursor = MediaReplayCursor::default();2649- let old_keyframe = media_v3_encoded_frame(60, true, true, 1);2650- let old_delta = media_v3_encoded_frame(61, false, false, 1);2651- let replacement = media_v3_encoded_frame(70, true, true, 1);2652-2653- for frame in [&old_keyframe, &old_delta] {2654- assert!(matches!(2655- group_cursor.classify(frame),2656- MediaV3GroupDecision::Send(_)2657- ));2658- assert!(matches!(2659- scheduler.schedule(frame.sequence, frame.keyframe && frame.codec_config),2660- MediaV3ScheduleDecision::Send { .. }2661- ));2662- replay_cursor.commit_sent(frame);2663- }2664- assert!(matches!(2665- group_cursor.classify(&replacement),2666- MediaV3GroupDecision::Send(_)2667- ));2668- assert_eq!(2669- scheduler.schedule(replacement.sequence, true),2670- MediaV3ScheduleDecision::Send {2671- discontinuity: true,2672- cancel_sequences: vec![60, 61],2673- }2674- );2675- replay_cursor.commit_sent(&replacement);2676-2677- assert_eq!(2678- apply_media_v3_send_failure(2679- &mut scheduler,2680- &mut group_cursor,2681- &mut replay_cursor,2682- old_keyframe.sequence,2683- Some(replacement.sequence),2684- ),2685- None2686- );2687- assert_eq!(scheduler.in_flight, vec![70]);2688- assert!(!scheduler.waiting_for_keyframe);2689- assert_eq!(group_cursor.group_id, Some(70));2690- assert!(!group_cursor.waiting_for_keyframe);2691- assert_eq!(replay_cursor.last_sequence, Some(70));2692- assert!(!replay_cursor.waiting_for_keyframe);2693-2694- let next_delta = media_v3_encoded_frame(71, false, false, 1);2695- assert!(matches!(2696- group_cursor.classify(&next_delta),2697- MediaV3GroupDecision::Send(MediaV3ObjectPosition {2698- group_id: 70,2699- object_id: 1,2700- ..2701- })2702- ));2703- assert!(matches!(2704- scheduler.schedule(next_delta.sequence, false),2705- MediaV3ScheduleDecision::Send {2706- discontinuity: false,2707- ..2708- }2709- ));2710- }2711-2712- 2713- fn media_v3_control_gate_is_monotonic_and_accepts_at_most_ten_per_second() {2714- let started = Instant::now();2715- let request = |request_id| {2716- MediaControlRequestV3::request_keyframe(2717- request_id,2718- None,2719- sigil_protocol::KeyframeRequestReasonV3::TransportGap,2720- )2721- };2722- let mut gate = MediaV3ControlGate::default();2723- assert_eq!(2724- gate.accept(request(1), started).unwrap(),2725- MediaV3ControlDecision::Accept2726- );2727- assert_eq!(2728- gate.accept(2729- request(2),2730- started + MEDIA_V3_CONTROL_REQUEST_INTERVAL - Duration::from_nanos(1),2731- )2732- .unwrap(),2733- MediaV3ControlDecision::Pace(Duration::from_nanos(1))2734- );2735- assert_eq!(2736- gate.accept(request(3), started + MEDIA_V3_CONTROL_REQUEST_INTERVAL)2737- .unwrap(),2738- MediaV3ControlDecision::Accept2739- );2740- assert!(2741- gate.accept(request(3), started + Duration::from_secs(1))2742- .is_err()2743- );2744- }2745-2746- 2747- async fn media_v3_rejected_control_requests_apply_read_side_pacing() {2748- use tokio::io::AsyncWriteExt as _;2749-2750- let requests = [1, 2, 3].map(|request_id| {2751- MediaControlRequestV3::request_keyframe(2752- request_id,2753- None,2754- KeyframeRequestReasonV3::TransportGap,2755- )2756- });2757- let (mut writer, reader) = tokio::io::duplex(128);2758- for request in &requests {2759- sigil_protocol::write_media_control_request_v3(&mut writer, request)2760- .await2761- .unwrap();2762- }2763- writer.shutdown().await.unwrap();2764- let (sender, mut receiver) = tokio::sync::watch::channel(None);2765-2766- let started = Instant::now();2767- forward_media_v3_control_requests(reader, sender)2768- .await2769- .unwrap();2770-2771- assert!(started.elapsed() >= MEDIA_V3_CONTROL_REQUEST_INTERVAL);2772- receiver.changed().await.unwrap();2773- assert_eq!(*receiver.borrow_and_update(), Some(requests[2]));2774- }2775-2776- 2777- async fn media_v3_control_eof_is_clean_and_malformed_input_is_terminal() {2778- use tokio::io::AsyncWriteExt as _;2779-2780- let (writer, reader) = tokio::io::duplex(64);2781- let (sender, mut receiver) = tokio::sync::watch::channel(None);2782- drop(writer);2783- forward_media_v3_control_requests(reader, sender)2784- .await2785- .unwrap();2786- assert!(receiver.changed().await.is_err());2787-2788- let final_request = MediaControlRequestV3::request_keyframe(2789- 1,2790- Some(9),2791- sigil_protocol::KeyframeRequestReasonV3::DecoderReset,2792- );2793- let (mut writer, reader) = tokio::io::duplex(64);2794- let (sender, mut receiver) = tokio::sync::watch::channel(None);2795- sigil_protocol::write_media_control_request_v3(&mut writer, &final_request)2796- .await2797- .unwrap();2798- writer.shutdown().await.unwrap();2799- forward_media_v3_control_requests(reader, sender)2800- .await2801- .unwrap();2802- receiver.changed().await.unwrap();2803- assert_eq!(*receiver.borrow_and_update(), Some(final_request));2804-2805- let (mut writer, reader) = tokio::io::duplex(64);2806- let (sender, _receiver) = tokio::sync::watch::channel(None);2807- writer2808- .write_all(&[0; sigil_protocol::MEDIA_CONTROL_REQUEST_V3_LEN])2809- .await2810- .unwrap();2811- writer.shutdown().await.unwrap();2812- assert!(2813- forward_media_v3_control_requests(reader, sender)2814- .await2815- .is_err()2816- );2817- }2818-2819- 2820- fn media_v3_deadline_ceil_clamps_four_frame_periods() {2821- assert_eq!(media_v3_delivery_timeout_ms(60), 67);2822- assert_eq!(media_v3_delivery_timeout_ms(240), 17);2823- assert_eq!(2824- media_v3_delivery_timeout_ms(1_000),2825- MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS2826- );2827- assert_eq!(2828- media_v3_delivery_timeout_ms(1),2829- MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS2830- );2831- }2832-2833- 2834- fn media_v3_publisher_priority_maps_to_inverse_transport_priority() {2835- let keyframe_transport =2836- media_v3_transport_priority(MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY).unwrap();2837- let delta_transport =2838- media_v3_transport_priority(MEDIA_V3_DELTA_PUBLISHER_PRIORITY).unwrap();2839- assert_eq!(keyframe_transport, MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY);2840- assert_eq!(delta_transport, MEDIA_V3_DELTA_TRANSPORT_PRIORITY);2841- assert!(keyframe_transport > delta_transport);2842- assert!(media_v3_transport_priority(1).is_err());2843- }2844-28451548 28461549 fn capability_negotiation_is_an_exact_intersection() {28471550 let hello = ClientHello::new(@@ -3050,52 +1753,6 @@ mod tests {
30501753 );30511754 }30521755 3053- 3054- fn current_gop_replay_is_complete_and_skips_already_sent_frames() {3055- let frame = |sequence, keyframe| EncodedFrame {3056- sequence,3057- width: 1_280,3058- height: 800,3059- capture_timestamp_micros: sequence,3060- presentation_timestamp_micros: sequence as i64,3061- observed_at: std::time::Instant::now(),3062- keyframe,3063- codec_config: keyframe,3064- discontinuity: false,3065- data: Arc::from([sequence as u8]),3066- };3067-3068- let gop = || EncodedGop {3069- frames: vec![frame(10, true), frame(11, false), frame(12, false)],3070- payload_bytes: 3,3071- };3072-3073- let initial = new_current_gop_frames(gop(), None).collect::<Vec<_>>();3074- assert_eq!(3075- initial3076- .iter()3077- .map(|frame| frame.sequence)3078- .collect::<Vec<_>>(),3079- vec![10, 11, 12]3080- );3081- assert!(initial[0].keyframe && initial[0].codec_config);3082-3083- let resumed = new_current_gop_frames(gop(), Some(10)).collect::<Vec<_>>();3084- assert_eq!(3085- resumed3086- .iter()3087- .map(|frame| frame.sequence)3088- .collect::<Vec<_>>(),3089- vec![11, 12]3090- );3091-3092- let behind_current_gop = new_current_gop_frames(gop(), Some(8)).collect::<Vec<_>>();3093- assert_eq!(behind_current_gop[0].sequence, 10);3094- assert!(behind_current_gop[0].keyframe && behind_current_gop[0].codec_config);3095-3096- assert!(new_current_gop_frames(gop(), Some(12)).next().is_none());3097- }3098-30991756 31001757 fn initial_current_gop_replay_preserves_complete_startup_snapshot() {31011758 let observed_now = Instant::now();@@ -0,0 +1,1356 @@
1+use super::*;2+3+const MEDIA_V3_IN_FLIGHT_CAPACITY: usize = 4;4+const MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY: u8 = 0;5+const MEDIA_V3_DELTA_PUBLISHER_PRIORITY: u8 = 128;6+const MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY: i32 = 10;7+const MEDIA_V3_DELTA_TRANSPORT_PRIORITY: i32 = 0;8+// Wire-visible application reset code. Keep stable across compatible v3 releases.9+const MEDIA_V3_RESET_CODE: u32 = 0x5357;10+const MEDIA_V3_DELIVERY_FRAME_PERIODS: u64 = 4;11+const MEDIA_V3_MAX_CONTROL_REQUESTS_PER_SECOND: u32 = 10;12+const MEDIA_V3_CONTROL_REQUEST_INTERVAL: Duration =13+ Duration::from_millis(1_000 / MEDIA_V3_MAX_CONTROL_REQUESTS_PER_SECOND as u64);14+15+16+pub(super) struct MediaV3GroupCursor {17+ group_id: Option<u64>,18+ last_sequence: Option<u64>,19+ last_object_id: Option<u32>,20+ payload_bytes: usize,21+ waiting_for_keyframe: bool,22+ discontinuity_pending: bool,23+}24+25+impl Default for MediaV3GroupCursor {26+ fn default() -> Self {27+ Self {28+ group_id: None,29+ last_sequence: None,30+ last_object_id: None,31+ payload_bytes: 0,32+ waiting_for_keyframe: true,33+ discontinuity_pending: false,34+ }35+ }36+}37+38+impl MediaV3GroupCursor {39+ pub(super) fn classify(&mut self, frame: &EncodedFrame) -> MediaV3GroupDecision {40+ let independently_decodable = frame.keyframe && frame.codec_config;41+ if independently_decodable {42+ if frame.data.len() > MAX_MEDIA_GROUP_BYTES_V3 {43+ self.enter_resync();44+ return MediaV3GroupDecision::EnterResync;45+ }46+ let sequence_discontinuity = self47+ .last_sequence48+ .is_some_and(|last| last.checked_add(1) != Some(frame.sequence));49+ let position = MediaV3ObjectPosition {50+ group_id: frame.sequence,51+ object_id: 0,52+ discontinuity: self.discontinuity_pending || sequence_discontinuity,53+ };54+ self.group_id = Some(frame.sequence);55+ self.last_sequence = Some(frame.sequence);56+ self.last_object_id = Some(0);57+ self.payload_bytes = frame.data.len();58+ self.waiting_for_keyframe = false;59+ self.discontinuity_pending = false;60+ return MediaV3GroupDecision::Send(position);61+ }62+63+ if self.waiting_for_keyframe {64+ return MediaV3GroupDecision::SkipUntilKeyframe;65+ }66+ if frame.keyframe || frame.codec_config {67+ self.enter_resync();68+ return MediaV3GroupDecision::EnterResync;69+ }70+ let Some(group_id) = self.group_id else {71+ self.enter_resync();72+ return MediaV3GroupDecision::EnterResync;73+ };74+ let contiguous =75+ self.last_sequence.and_then(|last| last.checked_add(1)) == Some(frame.sequence);76+ let object_id = self.last_object_id.and_then(|last| last.checked_add(1));77+ let payload_bytes = self.payload_bytes.checked_add(frame.data.len());78+ let (Some(object_id), Some(payload_bytes)) = (object_id, payload_bytes) else {79+ self.enter_resync();80+ return MediaV3GroupDecision::EnterResync;81+ };82+ if !contiguous83+ || object_id > MAX_MEDIA_OBJECT_ID_V384+ || payload_bytes > MAX_MEDIA_GROUP_BYTES_V385+ {86+ self.enter_resync();87+ return MediaV3GroupDecision::EnterResync;88+ }89+90+ self.last_sequence = Some(frame.sequence);91+ self.last_object_id = Some(object_id);92+ self.payload_bytes = payload_bytes;93+ MediaV3GroupDecision::Send(MediaV3ObjectPosition {94+ group_id,95+ object_id,96+ discontinuity: false,97+ })98+ }99+100+ pub(super) fn request_keyframe(&mut self) {101+ self.enter_resync();102+ }103+104+ fn enter_resync(&mut self) {105+ self.group_id = None;106+ self.last_object_id = None;107+ self.payload_bytes = 0;108+ self.waiting_for_keyframe = true;109+ self.discontinuity_pending = true;110+ }111+}112+113+114+enum MediaV3ScheduleDecision {115+ Send {116+ discontinuity: bool,117+ cancel_sequences: Vec<u64>,118+ },119+ SkipUntilKeyframe,120+ EnterResync {121+ cancel_sequences: Vec<u64>,122+ },123+}124+125+126+struct MediaV3Scheduler {127+ in_flight: Vec<u64>,128+ last_scheduled_sequence: Option<u64>,129+ waiting_for_keyframe: bool,130+ discontinuity_pending: bool,131+}132+133+impl Default for MediaV3Scheduler {134+ fn default() -> Self {135+ Self {136+ in_flight: Vec::with_capacity(MEDIA_V3_IN_FLIGHT_CAPACITY),137+ last_scheduled_sequence: None,138+ waiting_for_keyframe: true,139+ discontinuity_pending: false,140+ }141+ }142+}143+144+impl MediaV3Scheduler {145+ fn schedule(146+ &mut self,147+ sequence: u64,148+ independently_decodable: bool,149+ ) -> MediaV3ScheduleDecision {150+ if self151+ .last_scheduled_sequence152+ .is_some_and(|last| sequence <= last)153+ {154+ return MediaV3ScheduleDecision::SkipUntilKeyframe;155+ }156+157+ let sequence_discontinuity = self158+ .last_scheduled_sequence159+ .is_some_and(|last| last.checked_add(1) != Some(sequence));160+ if independently_decodable {161+ let cancel_sequences = std::mem::take(&mut self.in_flight);162+ let discontinuity = self.discontinuity_pending163+ || sequence_discontinuity164+ || !cancel_sequences.is_empty();165+ self.in_flight.push(sequence);166+ self.last_scheduled_sequence = Some(sequence);167+ self.waiting_for_keyframe = false;168+ self.discontinuity_pending = false;169+ return MediaV3ScheduleDecision::Send {170+ discontinuity,171+ cancel_sequences,172+ };173+ }174+175+ if self.waiting_for_keyframe {176+ return MediaV3ScheduleDecision::SkipUntilKeyframe;177+ }178+ if sequence_discontinuity || self.in_flight.len() == MEDIA_V3_IN_FLIGHT_CAPACITY {179+ return MediaV3ScheduleDecision::EnterResync {180+ cancel_sequences: self.enter_resync(),181+ };182+ }183+184+ self.in_flight.push(sequence);185+ self.last_scheduled_sequence = Some(sequence);186+ MediaV3ScheduleDecision::Send {187+ discontinuity: false,188+ cancel_sequences: Vec::new(),189+ }190+ }191+192+ fn complete(&mut self, sequence: u64) {193+ if let Some(index) = self.in_flight.iter().position(|value| *value == sequence) {194+ self.in_flight.swap_remove(index);195+ }196+ }197+198+ fn fail(&mut self, sequence: u64) -> Option<Vec<u64>> {199+ let index = self.in_flight.iter().position(|value| *value == sequence)?;200+ self.in_flight.swap_remove(index);201+ Some(self.enter_resync())202+ }203+204+ fn fail_all(&mut self) -> Vec<u64> {205+ self.enter_resync()206+ }207+208+ fn request_keyframe(&mut self) -> Vec<u64> {209+ if self.waiting_for_keyframe {210+ self.discontinuity_pending = true;211+ return Vec::new();212+ }213+ self.enter_resync()214+ }215+216+ fn enter_resync(&mut self) -> Vec<u64> {217+ self.waiting_for_keyframe = true;218+ self.discontinuity_pending = true;219+ std::mem::take(&mut self.in_flight)220+ }221+}222+223+fn apply_media_v3_keyframe_request(224+ scheduler: &mut MediaV3Scheduler,225+ group_cursor: &mut MediaV3GroupCursor,226+ replay_cursor: &mut MediaReplayCursor,227+ through_sequence: Option<u64>,228+ reason: KeyframeRequestReasonV3,229+) -> (bool, Vec<u64>) {230+ // Every v3 session already begins by replaying the bounded current GOP231+ // from object zero. A Join arriving after that replay was scheduled must232+ // not cancel the only decodable image on a damage-driven static source.233+ if reason == KeyframeRequestReasonV3::Join || scheduler.waiting_for_keyframe {234+ return (false, Vec::new());235+ }236+ let cancel_sequences = scheduler.request_keyframe();237+ group_cursor.request_keyframe();238+ replay_cursor.enter_resync_through(through_sequence);239+ (true, cancel_sequences)240+}241+242+fn apply_media_v3_send_failure(243+ scheduler: &mut MediaV3Scheduler,244+ group_cursor: &mut MediaV3GroupCursor,245+ replay_cursor: &mut MediaReplayCursor,246+ sequence: u64,247+ through_sequence: Option<u64>,248+) -> Option<Vec<u64>> {249+ let cancel_sequences = scheduler.fail(sequence)?;250+ group_cursor.request_keyframe();251+ replay_cursor.enter_resync_through(through_sequence);252+ Some(cancel_sequences)253+}254+255+256+enum MediaV3ControlDecision {257+ Accept,258+ Pace(Duration),259+}260+261+262+struct MediaV3ControlGate {263+ last_request_id: Option<u64>,264+ last_accepted_at: Option<Instant>,265+}266+267+impl MediaV3ControlGate {268+ fn accept(269+ &mut self,270+ request: MediaControlRequestV3,271+ now: Instant,272+ ) -> Result<MediaV3ControlDecision> {273+ ensure!(274+ self.last_request_id275+ .is_none_or(|last| request.request_id > last),276+ "v3 media control request IDs must be strictly increasing"277+ );278+ self.last_request_id = Some(request.request_id);279+ if let Some(last) = self.last_accepted_at {280+ let elapsed = now.saturating_duration_since(last);281+ if elapsed < MEDIA_V3_CONTROL_REQUEST_INTERVAL {282+ return Ok(MediaV3ControlDecision::Pace(283+ MEDIA_V3_CONTROL_REQUEST_INTERVAL - elapsed,284+ ));285+ }286+ }287+ self.last_accepted_at = Some(now);288+ Ok(MediaV3ControlDecision::Accept)289+ }290+}291+292+pub(super) async fn forward_media_v3_control_requests<R>(293+ mut reader: R,294+ sender: tokio::sync::watch::Sender<Option<MediaControlRequestV3>>,295+) -> Result<()>296+where297+ R: tokio::io::AsyncRead + Unpin,298+{299+ let mut gate = MediaV3ControlGate::default();300+ while let Some(request) = read_media_control_request_v3(&mut reader).await? {301+ match gate.accept(request, Instant::now())? {302+ MediaV3ControlDecision::Accept => {}303+ MediaV3ControlDecision::Pace(retry_after) => {304+ // Stop reading during the rejection interval so QUIC flow305+ // control, rather than this task, absorbs an abusive burst.306+ // This also bounds rejection logging to the configured rate.307+ debug!(308+ request_id = request.request_id,309+ retry_after_ms = retry_after.as_millis(),310+ "paced rate-limited v3 keyframe request"311+ );312+ tokio::time::sleep(retry_after).await;313+ continue;314+ }315+ }316+ sender.send_replace(Some(request));317+ if sender.is_closed() {318+ return Ok(());319+ }320+ }321+ Ok(())322+}323+324+struct ResetOnDropSendStreamV3(Option<SendStream>);325+326+impl ResetOnDropSendStreamV3 {327+ fn new(stream: SendStream) -> Self {328+ Self(Some(stream))329+ }330+331+ fn stream(&self) -> &SendStream {332+ self.0.as_ref().expect("send stream guard is armed")333+ }334+335+ fn stream_mut(&mut self) -> &mut SendStream {336+ self.0.as_mut().expect("send stream guard is armed")337+ }338+339+ fn disarm(&mut self) {340+ self.0.take();341+ }342+}343+344+impl Drop for ResetOnDropSendStreamV3 {345+ fn drop(&mut self) {346+ if let Some(stream) = self.0.as_mut() {347+ let _ = stream.reset(MEDIA_V3_RESET_CODE.into());348+ }349+ }350+}351+352+pub(super) async fn serve_media_v3(353+ connection: Connection,354+ config: HostConfig,355+ sessions: &Arc<SessionRegistry>,356+ authorization: &AuthorizationPolicy,357+) -> Result<()> {358+ let remote = connection.remote_id();359+ let handshake_permit = sessions360+ .pending_handshakes361+ .try_acquire()362+ .context("too many pending handshakes")?;363+ let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())364+ .await365+ .context("timed out accepting media v3 handshake stream")?366+ .context("accepting media v3 handshake stream")?;367+ let hello = receive_hello(&mut recv, Capability::VideoH264).await?;368+ drop(handshake_permit);369+ debug!(%remote, agent = %hello.agent, "media v3 hello received");370+371+ let grants = match authorization.authorize_or_redeem(372+ remote,373+ hello.invitation.as_deref(),374+ unix_timestamp_now()?,375+ ) {376+ Ok(grants) => grants,377+ Err(error) => {378+ send_rejection(&mut send, "Portal peer is not authorized").await?;379+ return Err(error.context("authorizing media v3 peer"));380+ }381+ };382+ ensure!(383+ grants.contains(InvitationGrants::VIEW),384+ "authorized media peer lacks view permission"385+ );386+387+ let lease = match sessions.claim(remote, hello.nonce, grants) {388+ Ok(lease) => lease,389+ Err(error) => {390+ send_rejection(&mut send, "host already has an active client").await?;391+ return Err(error);392+ }393+ };394+395+ let source = match config.source {396+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),397+ VideoSource::GamescopePipewire => {398+ let primary = spawn_gamescope_pipewire_after_static_preflight(399+ config.clone(),400+ lease.session_clock,401+ )402+ .await?;403+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await404+ }405+ };406+ let EncodedSource {407+ frames: frame_receiver,408+ current_gop: mut current_gop_receiver,409+ task: source_task,410+ pointer_surface_dimensions,411+ encoder_control,412+ } = match source {413+ Ok(source) => source,414+ Err(error) => {415+ send_rejection(&mut send, "video source is unavailable").await?;416+ return Err(error);417+ }418+ };419+ let source_task = SourceTaskGuard::new(source_task);420+ sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;421+422+ let mut media_hello = HostHello::accepted(423+ lease.session_id,424+ negotiated_capabilities(&hello, MEDIA_CAPABILITIES),425+ );426+ if let Some(dimensions) = pointer_surface_dimensions {427+ media_hello = media_hello.with_pointer_surface_dimensions(dimensions);428+ }429+ write_host_hello(&mut send, &media_hello).await?;430+ send.finish()431+ .context("finishing media v3 handshake response")?;432+ drop(send);433+ info!(%remote, session_id = lease.session_id, "media v3 client accepted");434+435+ let session_result = run_media_v3_session(436+ &connection,437+ &config,438+ &mut current_gop_receiver,439+ recv,440+ remote,441+ encoder_control,442+ Arc::clone(&lease.media_v3_telemetry),443+ )444+ .await;445+446+ drop(current_gop_receiver);447+ drop(frame_receiver);448+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;449+ drop(lease);450+ info!(%remote, "media v3 client released");451+ session_result452+}453+454+async fn run_media_v3_session(455+ connection: &Connection,456+ config: &HostConfig,457+ current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,458+ control_recv: iroh::endpoint::RecvStream,459+ remote: EndpointId,460+ encoder_control: Option<EncoderControl>,461+ telemetry: Arc<MediaV3Telemetry>,462+) -> Result<()> {463+ let maximum_replay_age = maximum_media_replay_age(config.framerate);464+ let delivery_timeout_ms = media_v3_delivery_timeout_ms(config.framerate);465+ let mut replay_cursor = MediaReplayCursor::default();466+ let mut group_cursor = MediaV3GroupCursor::default();467+ let mut scheduler = MediaV3Scheduler::default();468+ let mut send_tasks = tokio::task::JoinSet::new();469+ let (control_sender, mut control_requests) = tokio::sync::watch::channel(None);470+ let mut control_task = tokio::spawn(forward_media_v3_control_requests(471+ control_recv,472+ control_sender,473+ ));474+ let mut control_task_finished = false;475+ let mut control_receiver_open = true;476+ let mut forced_idr = ForcedIdrCoordinator::new(encoder_control, Arc::clone(&telemetry));477+478+ let result = loop {479+ telemetry.record_selected_path(connection);480+ tokio::select! {481+ biased;482+ control_result = &mut control_task, if !control_task_finished => {483+ control_task_finished = true;484+ match control_result {485+ Ok(Ok(())) => {486+ // Keep polling the watch receiver after clean EOF so a487+ // final request published immediately before the488+ // sender closed cannot be lost.489+ debug!(%remote, "media v3 control stream closed");490+ }491+ Ok(Err(error)) => {492+ break Err(error).context("reading media v3 control stream");493+ }494+ Err(error) => {495+ break Err(error).context("media v3 control task failed");496+ }497+ }498+ }499+ changed = control_requests.changed(), if control_receiver_open => {500+ if changed.is_err() {501+ control_receiver_open = false;502+ continue;503+ }504+ let request = *control_requests.borrow_and_update();505+ let Some(request) = request else {506+ continue;507+ };508+ let through_sequence = current_gop_receiver509+ .borrow()510+ .as_ref()511+ .and_then(|gop| gop.frames.last())512+ .map(|frame| frame.sequence);513+ let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(514+ &mut scheduler,515+ &mut group_cursor,516+ &mut replay_cursor,517+ through_sequence,518+ request.reason,519+ );520+ if !cancel_sequences.is_empty() {521+ telemetry.scheduler_cancellations.fetch_add(522+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),523+ Ordering::Relaxed,524+ );525+ send_tasks.abort_all();526+ }527+ let forced_idr_disposition = forced_idr.request(request.reason);528+ if let ForcedIdrDisposition::Failed { error } = &forced_idr_disposition {529+ warn!(530+ %remote,531+ request_id = request.request_id,532+ ?request.reason,533+ %error,534+ "forced-IDR request failed; retaining natural-IDR fallback"535+ );536+ }537+ debug!(538+ %remote,539+ request_id = request.request_id,540+ ?request.reason,541+ advisory_last_sequence = ?request.last_sequence,542+ coalesced = !transitioned,543+ ?cancel_sequences,544+ ?forced_idr_disposition,545+ "accepted media v3 keyframe request"546+ );547+ }548+ acknowledgement = forced_idr.acknowledgements.join_next(),549+ if forced_idr.pending_revision.is_some() =>550+ {551+ forced_idr.complete(acknowledgement, remote, "grouped-v3");552+ }553+ closed = connection.closed() => {554+ debug!(%remote, ?closed, "media v3 connection closed");555+ break Ok(());556+ }557+ task = send_tasks.join_next(), if !send_tasks.is_empty() => {558+ match task.expect("guarded by non-empty send task set") {559+ Ok((sequence, Ok(()))) => scheduler.complete(sequence),560+ Ok((sequence, Err(error))) => {561+ let through_sequence = current_gop_receiver562+ .borrow()563+ .as_ref()564+ .and_then(|gop| gop.frames.last())565+ .map(|frame| frame.sequence);566+ if let Some(cancel_sequences) = apply_media_v3_send_failure(567+ &mut scheduler,568+ &mut group_cursor,569+ &mut replay_cursor,570+ sequence,571+ through_sequence,572+ ) {573+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);574+ telemetry.scheduler_cancellations.fetch_add(575+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),576+ Ordering::Relaxed,577+ );578+ warn!(579+ sequence,580+ %error,581+ "media v3 object send failed; waiting for keyframe"582+ );583+ if !cancel_sequences.is_empty() {584+ send_tasks.abort_all();585+ }586+ } else {587+ debug!(588+ sequence,589+ %error,590+ "ignored stale media v3 object failure from a superseded group"591+ );592+ }593+ }594+ Err(error) if error.is_cancelled() => {}595+ Err(error) => {596+ warn!(%error, "media v3 object task failed; waiting for keyframe");597+ let cancel_sequences = scheduler.fail_all();598+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);599+ telemetry.scheduler_cancellations.fetch_add(600+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),601+ Ordering::Relaxed,602+ );603+ group_cursor.request_keyframe();604+ let through_sequence = current_gop_receiver605+ .borrow()606+ .as_ref()607+ .and_then(|gop| gop.frames.last())608+ .map(|frame| frame.sequence);609+ replay_cursor.enter_resync_through(through_sequence);610+ if !cancel_sequences.is_empty() {611+ send_tasks.abort_all();612+ }613+ }614+ }615+ }616+ changed = current_gop_receiver.changed() => {617+ if let Err(error) = changed {618+ break Err(error).context("encoded source stopped");619+ }620+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {621+ continue;622+ };623+ let initial_replay_started_at =624+ replay_cursor.last_sequence.is_none().then(Instant::now);625+ let replay_through_sequence = current_gop626+ .frames627+ .last()628+ .map(|frame| frame.sequence)629+ .context("current GOP snapshot is empty")?;630+631+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {632+ let replay_discontinuity = match replay_cursor.classify(633+ &frame,634+ replay_through_sequence,635+ initial_replay_started_at,636+ Instant::now(),637+ maximum_replay_age,638+ ) {639+ MediaReplayDecision::Send { discontinuity } => discontinuity,640+ MediaReplayDecision::SkipUntilKeyframe => {641+ replay_cursor.enter_resync_through(Some(replay_through_sequence));642+ break;643+ }644+ MediaReplayDecision::DiscardStaleSuffix { .. } => {645+ group_cursor.request_keyframe();646+ let cancel_sequences = scheduler.fail_all();647+ telemetry.scheduler_cancellations.fetch_add(648+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),649+ Ordering::Relaxed,650+ );651+ if !cancel_sequences.is_empty() {652+ send_tasks.abort_all();653+ }654+ break;655+ }656+ };657+ let position = match group_cursor.classify(&frame) {658+ MediaV3GroupDecision::Send(position) => position,659+ MediaV3GroupDecision::SkipUntilKeyframe => {660+ replay_cursor.enter_resync_through(Some(replay_through_sequence));661+ break;662+ }663+ MediaV3GroupDecision::EnterResync => {664+ replay_cursor.enter_resync_through(Some(replay_through_sequence));665+ let cancel_sequences = scheduler.fail_all();666+ telemetry.scheduler_cancellations.fetch_add(667+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),668+ Ordering::Relaxed,669+ );670+ if !cancel_sequences.is_empty() {671+ send_tasks.abort_all();672+ }673+ break;674+ }675+ };676+ let independently_decodable = position.object_id == 0;677+ let (scheduler_discontinuity, cancel_sequences) =678+ match scheduler.schedule(frame.sequence, independently_decodable) {679+ MediaV3ScheduleDecision::Send {680+ discontinuity,681+ cancel_sequences,682+ } => (discontinuity, cancel_sequences),683+ MediaV3ScheduleDecision::SkipUntilKeyframe => continue,684+ MediaV3ScheduleDecision::EnterResync { cancel_sequences } => {685+ group_cursor.request_keyframe();686+ replay_cursor.enter_resync_through(Some(replay_through_sequence));687+ telemetry.scheduler_cancellations.fetch_add(688+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),689+ Ordering::Relaxed,690+ );691+ if !cancel_sequences.is_empty() {692+ send_tasks.abort_all();693+ }694+ break;695+ }696+ };697+ if !cancel_sequences.is_empty() {698+ telemetry.scheduler_cancellations.fetch_add(699+ u64::try_from(cancel_sequences.len()).unwrap_or(u64::MAX),700+ Ordering::Relaxed,701+ );702+ debug!(703+ sequence = frame.sequence,704+ group_id = position.group_id,705+ ?cancel_sequences,706+ "configured keyframe superseding media v3 objects"707+ );708+ send_tasks.abort_all();709+ }710+711+ let media_object = media_v3_object_for_encoded(712+ config,713+ &frame,714+ position,715+ replay_discontinuity716+ || position.discontinuity717+ || scheduler_discontinuity,718+ delivery_timeout_ms,719+ )?;720+ let sequence = frame.sequence;721+ let stream = match tokio::time::timeout(722+ Duration::from_millis(u64::from(delivery_timeout_ms)),723+ connection.open_uni(),724+ )725+ .await726+ {727+ Ok(Ok(stream)) => stream,728+ Ok(Err(error)) => {729+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);730+ warn!(sequence, %error, "opening media v3 object stream failed");731+ if let Some(cancel_sequences) = apply_media_v3_send_failure(732+ &mut scheduler,733+ &mut group_cursor,734+ &mut replay_cursor,735+ sequence,736+ Some(replay_through_sequence),737+ ) && !cancel_sequences.is_empty()738+ {739+ send_tasks.abort_all();740+ }741+ break;742+ }743+ Err(_) => {744+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);745+ warn!(sequence, "opening media v3 object stream timed out");746+ if let Some(cancel_sequences) = apply_media_v3_send_failure(747+ &mut scheduler,748+ &mut group_cursor,749+ &mut replay_cursor,750+ sequence,751+ Some(replay_through_sequence),752+ ) && !cancel_sequences.is_empty()753+ {754+ send_tasks.abort_all();755+ }756+ break;757+ }758+ };759+ let stream = ResetOnDropSendStreamV3::new(stream);760+ let transport_priority =761+ media_v3_transport_priority(media_object.header.publisher_priority)?;762+ if let Err(error) = stream.stream().set_priority(transport_priority) {763+ telemetry.send_failures.fetch_add(1, Ordering::Relaxed);764+ warn!(sequence, %error, "setting media v3 object priority failed");765+ if let Some(cancel_sequences) = apply_media_v3_send_failure(766+ &mut scheduler,767+ &mut group_cursor,768+ &mut replay_cursor,769+ sequence,770+ Some(replay_through_sequence),771+ ) && !cancel_sequences.is_empty()772+ {773+ send_tasks.abort_all();774+ }775+ break;776+ }777+ send_tasks.spawn(async move {778+ (779+ sequence,780+ send_media_v3_object(stream, media_object).await,781+ )782+ });783+ replay_cursor.commit_sent(&frame);784+ }785+ }786+ }787+ };788+789+ forced_idr.abort_and_drain(remote, "grouped-v3").await;790+ send_tasks.abort_all();791+ while send_tasks.join_next().await.is_some() {}792+ if !control_task_finished {793+ control_task.abort();794+ let _ = control_task.await;795+ }796+ result797+}798+799+fn media_v3_delivery_timeout_ms(framerate: u32) -> u32 {800+ debug_assert!(framerate > 0);801+ let timeout_ms = 1_000_u64802+ .saturating_mul(MEDIA_V3_DELIVERY_FRAME_PERIODS)803+ .div_ceil(u64::from(framerate.max(1)));804+ u32::try_from(timeout_ms).unwrap_or(u32::MAX).clamp(805+ MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,806+ MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS,807+ )808+}809+810+fn media_v3_object_for_encoded(811+ _config: &HostConfig,812+ frame: &EncodedFrame,813+ position: MediaV3ObjectPosition,814+ discontinuity: bool,815+ delivery_timeout_ms: u32,816+) -> Result<MediaObjectV3> {817+ let mut flags = FrameFlags::NONE;818+ if frame.keyframe {819+ flags = flags.union(FrameFlags::KEYFRAME);820+ }821+ if frame.codec_config {822+ flags = flags.union(FrameFlags::CODEC_CONFIG);823+ }824+ if discontinuity || frame.discontinuity {825+ flags = flags.union(FrameFlags::DISCONTINUITY);826+ }827+ let publisher_priority = if position.object_id == 0 {828+ MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY829+ } else {830+ MEDIA_V3_DELTA_PUBLISHER_PRIORITY831+ };832+ let header = MediaObjectHeaderV3::h264(833+ frame.width,834+ frame.height,835+ frame.data.len(),836+ publisher_priority,837+ flags,838+ position.object_id,839+ position.group_id,840+ frame.sequence,841+ frame.capture_timestamp_micros,842+ frame.presentation_timestamp_micros,843+ delivery_timeout_ms,844+ )?;845+ MediaObjectV3::new(header, frame.data.as_ref().to_vec()).map_err(Into::into)846+}847+848+async fn send_media_v3_object(849+ mut stream: ResetOnDropSendStreamV3,850+ object: MediaObjectV3,851+) -> Result<()> {852+ let delivery_timeout = Duration::from_millis(u64::from(object.header.delivery_timeout_ms));853+ tokio::time::timeout(delivery_timeout, async {854+ write_media_object_v3(stream.stream_mut(), &object)855+ .await856+ .context("writing media v3 object")?;857+ stream858+ .stream_mut()859+ .finish()860+ .context("finishing media v3 object stream")?;861+ match stream862+ .stream()863+ .stopped()864+ .await865+ .context("waiting for media v3 object acknowledgement")?866+ {867+ None => {868+ stream.disarm();869+ Ok(())870+ }871+ Some(code) => bail!("peer stopped media v3 object stream with code {code}"),872+ }873+ })874+ .await875+ .context("media v3 object exceeded its delivery timeout")?876+}877+878+fn media_v3_transport_priority(publisher_priority: u8) -> Result<i32> {879+ // V3 follows MoQ's lower-is-higher publisher priority, while Iroh/QUIC880+ // uses larger integers for more important streams. Keep the inversion881+ // explicit rather than leaking either convention across the boundary.882+ match publisher_priority {883+ MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY => Ok(MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY),884+ MEDIA_V3_DELTA_PUBLISHER_PRIORITY => Ok(MEDIA_V3_DELTA_TRANSPORT_PRIORITY),885+ _ => bail!("unsupported media v3 publisher priority {publisher_priority}"),886+ }887+}888+889+pub(super) fn new_current_gop_frames(890+ current_gop: EncodedGop,891+ last_sequence: Option<u64>,892+) -> impl Iterator<Item = EncodedFrame> {893+ current_gop894+ .frames895+ .into_iter()896+ .skip_while(move |frame| last_sequence.is_some_and(|last| frame.sequence <= last))897+}898+899+900+mod tests {901+ use super::super::media_v3_encoded_frame;902+ use super::*;903+904+ 905+ fn media_v3_groups_begin_at_configured_idr_and_assign_contiguous_objects() {906+ let mut cursor = MediaV3GroupCursor::default();907+ assert_eq!(908+ cursor.classify(&media_v3_encoded_frame(10, true, true, 4)),909+ MediaV3GroupDecision::Send(MediaV3ObjectPosition {910+ group_id: 10,911+ object_id: 0,912+ discontinuity: false,913+ })914+ );915+ assert_eq!(916+ cursor.classify(&media_v3_encoded_frame(11, false, false, 3)),917+ MediaV3GroupDecision::Send(MediaV3ObjectPosition {918+ group_id: 10,919+ object_id: 1,920+ discontinuity: false,921+ })922+ );923+924+ cursor.request_keyframe();925+ assert_eq!(926+ cursor.classify(&media_v3_encoded_frame(12, false, false, 3)),927+ MediaV3GroupDecision::SkipUntilKeyframe928+ );929+ assert_eq!(930+ cursor.classify(&media_v3_encoded_frame(20, true, true, 4)),931+ MediaV3GroupDecision::Send(MediaV3ObjectPosition {932+ group_id: 20,933+ object_id: 0,934+ discontinuity: true,935+ })936+ );937+ }938+939+ 940+ fn media_v3_group_rejects_noncontiguous_or_unconfigured_frames() {941+ let mut cursor = MediaV3GroupCursor::default();942+ assert_eq!(943+ cursor.classify(&media_v3_encoded_frame(10, true, false, 1)),944+ MediaV3GroupDecision::SkipUntilKeyframe945+ );946+ assert!(matches!(947+ cursor.classify(&media_v3_encoded_frame(20, true, true, 1)),948+ MediaV3GroupDecision::Send(_)949+ ));950+ assert_eq!(951+ cursor.classify(&media_v3_encoded_frame(22, false, false, 1)),952+ MediaV3GroupDecision::EnterResync953+ );954+ assert!(cursor.waiting_for_keyframe);955+ }956+957+ 958+ fn media_v3_group_accepts_exact_limits_and_rejects_overflow() {959+ let mut object_cursor = MediaV3GroupCursor::default();960+ assert!(matches!(961+ object_cursor.classify(&media_v3_encoded_frame(0, true, true, 1)),962+ MediaV3GroupDecision::Send(_)963+ ));964+ for sequence in 1..=u64::from(MAX_MEDIA_OBJECT_ID_V3) {965+ assert!(matches!(966+ object_cursor.classify(&media_v3_encoded_frame(sequence, false, false, 1)),967+ MediaV3GroupDecision::Send(MediaV3ObjectPosition { object_id, .. })968+ if u64::from(object_id) == sequence969+ ));970+ }971+ assert_eq!(972+ object_cursor.classify(&media_v3_encoded_frame(973+ u64::from(MAX_MEDIA_OBJECT_ID_V3) + 1,974+ false,975+ false,976+ 1,977+ )),978+ MediaV3GroupDecision::EnterResync979+ );980+981+ let mut byte_cursor = MediaV3GroupCursor::default();982+ assert!(matches!(983+ byte_cursor.classify(&media_v3_encoded_frame(10, true, true, 1)),984+ MediaV3GroupDecision::Send(_)985+ ));986+ byte_cursor.payload_bytes = MAX_MEDIA_GROUP_BYTES_V3 - 1;987+ assert!(matches!(988+ byte_cursor.classify(&media_v3_encoded_frame(11, false, false, 1)),989+ MediaV3GroupDecision::Send(_)990+ ));991+ assert_eq!(byte_cursor.payload_bytes, MAX_MEDIA_GROUP_BYTES_V3);992+ assert_eq!(993+ byte_cursor.classify(&media_v3_encoded_frame(12, false, false, 1)),994+ MediaV3GroupDecision::EnterResync995+ );996+ }997+998+ 999+ fn media_v3_keyframe_requests_cancel_once_and_recover_discontinuously() {1000+ let mut scheduler = MediaV3Scheduler::default();1001+ assert!(matches!(1002+ scheduler.schedule(10, true),1003+ MediaV3ScheduleDecision::Send { .. }1004+ ));1005+ assert!(matches!(1006+ scheduler.schedule(11, false),1007+ MediaV3ScheduleDecision::Send { .. }1008+ ));1009+ assert_eq!(scheduler.request_keyframe(), vec![10, 11]);1010+ assert!(scheduler.request_keyframe().is_empty());1011+ assert_eq!(1012+ scheduler.schedule(12, false),1013+ MediaV3ScheduleDecision::SkipUntilKeyframe1014+ );1015+ assert_eq!(1016+ scheduler.schedule(20, true),1017+ MediaV3ScheduleDecision::Send {1018+ discontinuity: true,1019+ cancel_sequences: vec![],1020+ }1021+ );1022+ }1023+1024+ 1025+ fn media_v3_join_request_preserves_initial_current_gop_replay() {1026+ let mut scheduler = MediaV3Scheduler::default();1027+ let mut group_cursor = MediaV3GroupCursor::default();1028+ let mut replay_cursor = MediaReplayCursor::default();1029+1030+ let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(1031+ &mut scheduler,1032+ &mut group_cursor,1033+ &mut replay_cursor,1034+ Some(42),1035+ KeyframeRequestReasonV3::Join,1036+ );1037+1038+ assert!(!transitioned);1039+ assert!(cancel_sequences.is_empty());1040+ assert_eq!(replay_cursor.last_sequence, None);1041+ assert!(replay_cursor.waiting_for_keyframe);1042+ assert!(!replay_cursor.discontinuity_pending);1043+ assert!(group_cursor.waiting_for_keyframe);1044+ assert!(!group_cursor.discontinuity_pending);1045+ assert!(!scheduler.discontinuity_pending);1046+1047+ assert!(matches!(1048+ group_cursor.classify(&media_v3_encoded_frame(10, true, true, 1)),1049+ MediaV3GroupDecision::Send(MediaV3ObjectPosition {1050+ discontinuity: false,1051+ ..1052+ })1053+ ));1054+ assert!(matches!(1055+ scheduler.schedule(10, true),1056+ MediaV3ScheduleDecision::Send {1057+ discontinuity: false,1058+ ..1059+ }1060+ ));1061+ }1062+1063+ 1064+ fn media_v3_late_join_preserves_active_static_current_gop() {1065+ let mut scheduler = MediaV3Scheduler::default();1066+ let mut group_cursor = MediaV3GroupCursor::default();1067+ let mut replay_cursor = MediaReplayCursor::default();1068+ let keyframe = media_v3_encoded_frame(10, true, true, 1);1069+ let delta = media_v3_encoded_frame(11, false, false, 1);1070+1071+ assert!(matches!(1072+ group_cursor.classify(&keyframe),1073+ MediaV3GroupDecision::Send(_)1074+ ));1075+ assert!(matches!(1076+ scheduler.schedule(keyframe.sequence, true),1077+ MediaV3ScheduleDecision::Send { .. }1078+ ));1079+ replay_cursor.commit_sent(&keyframe);1080+ assert!(matches!(1081+ group_cursor.classify(&delta),1082+ MediaV3GroupDecision::Send(_)1083+ ));1084+ assert!(matches!(1085+ scheduler.schedule(delta.sequence, false),1086+ MediaV3ScheduleDecision::Send { .. }1087+ ));1088+ replay_cursor.commit_sent(&delta);1089+1090+ let (transitioned, cancel_sequences) = apply_media_v3_keyframe_request(1091+ &mut scheduler,1092+ &mut group_cursor,1093+ &mut replay_cursor,1094+ Some(delta.sequence),1095+ KeyframeRequestReasonV3::Join,1096+ );1097+1098+ assert!(!transitioned);1099+ assert!(cancel_sequences.is_empty());1100+ assert_eq!(scheduler.in_flight, vec![10, 11]);1101+ assert!(!scheduler.waiting_for_keyframe);1102+ assert!(!scheduler.discontinuity_pending);1103+ assert_eq!(group_cursor.group_id, Some(10));1104+ assert_eq!(group_cursor.last_sequence, Some(11));1105+ assert!(!group_cursor.waiting_for_keyframe);1106+ assert_eq!(replay_cursor.last_sequence, Some(11));1107+ assert!(!replay_cursor.waiting_for_keyframe);1108+ }1109+1110+ 1111+ fn media_v3_stale_old_group_failure_preserves_replacement_group() {1112+ let mut scheduler = MediaV3Scheduler::default();1113+ let mut group_cursor = MediaV3GroupCursor::default();1114+ let mut replay_cursor = MediaReplayCursor::default();1115+ let old_keyframe = media_v3_encoded_frame(60, true, true, 1);1116+ let old_delta = media_v3_encoded_frame(61, false, false, 1);1117+ let replacement = media_v3_encoded_frame(70, true, true, 1);1118+1119+ for frame in [&old_keyframe, &old_delta] {1120+ assert!(matches!(1121+ group_cursor.classify(frame),1122+ MediaV3GroupDecision::Send(_)1123+ ));1124+ assert!(matches!(1125+ scheduler.schedule(frame.sequence, frame.keyframe && frame.codec_config),1126+ MediaV3ScheduleDecision::Send { .. }1127+ ));1128+ replay_cursor.commit_sent(frame);1129+ }1130+ assert!(matches!(1131+ group_cursor.classify(&replacement),1132+ MediaV3GroupDecision::Send(_)1133+ ));1134+ assert_eq!(1135+ scheduler.schedule(replacement.sequence, true),1136+ MediaV3ScheduleDecision::Send {1137+ discontinuity: true,1138+ cancel_sequences: vec![60, 61],1139+ }1140+ );1141+ replay_cursor.commit_sent(&replacement);1142+1143+ assert_eq!(1144+ apply_media_v3_send_failure(1145+ &mut scheduler,1146+ &mut group_cursor,1147+ &mut replay_cursor,1148+ old_keyframe.sequence,1149+ Some(replacement.sequence),1150+ ),1151+ None1152+ );1153+ assert_eq!(scheduler.in_flight, vec![70]);1154+ assert!(!scheduler.waiting_for_keyframe);1155+ assert_eq!(group_cursor.group_id, Some(70));1156+ assert!(!group_cursor.waiting_for_keyframe);1157+ assert_eq!(replay_cursor.last_sequence, Some(70));1158+ assert!(!replay_cursor.waiting_for_keyframe);1159+1160+ let next_delta = media_v3_encoded_frame(71, false, false, 1);1161+ assert!(matches!(1162+ group_cursor.classify(&next_delta),1163+ MediaV3GroupDecision::Send(MediaV3ObjectPosition {1164+ group_id: 70,1165+ object_id: 1,1166+ ..1167+ })1168+ ));1169+ assert!(matches!(1170+ scheduler.schedule(next_delta.sequence, false),1171+ MediaV3ScheduleDecision::Send {1172+ discontinuity: false,1173+ ..1174+ }1175+ ));1176+ }1177+1178+ 1179+ fn media_v3_control_gate_is_monotonic_and_accepts_at_most_ten_per_second() {1180+ let started = Instant::now();1181+ let request = |request_id| {1182+ MediaControlRequestV3::request_keyframe(1183+ request_id,1184+ None,1185+ sigil_protocol::KeyframeRequestReasonV3::TransportGap,1186+ )1187+ };1188+ let mut gate = MediaV3ControlGate::default();1189+ assert_eq!(1190+ gate.accept(request(1), started).unwrap(),1191+ MediaV3ControlDecision::Accept1192+ );1193+ assert_eq!(1194+ gate.accept(1195+ request(2),1196+ started + MEDIA_V3_CONTROL_REQUEST_INTERVAL - Duration::from_nanos(1),1197+ )1198+ .unwrap(),1199+ MediaV3ControlDecision::Pace(Duration::from_nanos(1))1200+ );1201+ assert_eq!(1202+ gate.accept(request(3), started + MEDIA_V3_CONTROL_REQUEST_INTERVAL)1203+ .unwrap(),1204+ MediaV3ControlDecision::Accept1205+ );1206+ assert!(1207+ gate.accept(request(3), started + Duration::from_secs(1))1208+ .is_err()1209+ );1210+ }1211+1212+ 1213+ async fn media_v3_rejected_control_requests_apply_read_side_pacing() {1214+ use tokio::io::AsyncWriteExt as _;1215+1216+ let requests = [1, 2, 3].map(|request_id| {1217+ MediaControlRequestV3::request_keyframe(1218+ request_id,1219+ None,1220+ KeyframeRequestReasonV3::TransportGap,1221+ )1222+ });1223+ let (mut writer, reader) = tokio::io::duplex(128);1224+ for request in &requests {1225+ sigil_protocol::write_media_control_request_v3(&mut writer, request)1226+ .await1227+ .unwrap();1228+ }1229+ writer.shutdown().await.unwrap();1230+ let (sender, mut receiver) = tokio::sync::watch::channel(None);1231+1232+ let started = Instant::now();1233+ forward_media_v3_control_requests(reader, sender)1234+ .await1235+ .unwrap();1236+1237+ assert!(started.elapsed() >= MEDIA_V3_CONTROL_REQUEST_INTERVAL);1238+ receiver.changed().await.unwrap();1239+ assert_eq!(*receiver.borrow_and_update(), Some(requests[2]));1240+ }1241+1242+ 1243+ async fn media_v3_control_eof_is_clean_and_malformed_input_is_terminal() {1244+ use tokio::io::AsyncWriteExt as _;1245+1246+ let (writer, reader) = tokio::io::duplex(64);1247+ let (sender, mut receiver) = tokio::sync::watch::channel(None);1248+ drop(writer);1249+ forward_media_v3_control_requests(reader, sender)1250+ .await1251+ .unwrap();1252+ assert!(receiver.changed().await.is_err());1253+1254+ let final_request = MediaControlRequestV3::request_keyframe(1255+ 1,1256+ Some(9),1257+ sigil_protocol::KeyframeRequestReasonV3::DecoderReset,1258+ );1259+ let (mut writer, reader) = tokio::io::duplex(64);1260+ let (sender, mut receiver) = tokio::sync::watch::channel(None);1261+ sigil_protocol::write_media_control_request_v3(&mut writer, &final_request)1262+ .await1263+ .unwrap();1264+ writer.shutdown().await.unwrap();1265+ forward_media_v3_control_requests(reader, sender)1266+ .await1267+ .unwrap();1268+ receiver.changed().await.unwrap();1269+ assert_eq!(*receiver.borrow_and_update(), Some(final_request));1270+1271+ let (mut writer, reader) = tokio::io::duplex(64);1272+ let (sender, _receiver) = tokio::sync::watch::channel(None);1273+ writer1274+ .write_all(&[0; sigil_protocol::MEDIA_CONTROL_REQUEST_V3_LEN])1275+ .await1276+ .unwrap();1277+ writer.shutdown().await.unwrap();1278+ assert!(1279+ forward_media_v3_control_requests(reader, sender)1280+ .await1281+ .is_err()1282+ );1283+ }1284+1285+ 1286+ fn media_v3_deadline_ceil_clamps_four_frame_periods() {1287+ assert_eq!(media_v3_delivery_timeout_ms(60), 67);1288+ assert_eq!(media_v3_delivery_timeout_ms(240), 17);1289+ assert_eq!(1290+ media_v3_delivery_timeout_ms(1_000),1291+ MIN_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS1292+ );1293+ assert_eq!(1294+ media_v3_delivery_timeout_ms(1),1295+ MAX_MEDIA_OBJECT_DELIVERY_TIMEOUT_MS1296+ );1297+ }1298+1299+ 1300+ fn media_v3_publisher_priority_maps_to_inverse_transport_priority() {1301+ let keyframe_transport =1302+ media_v3_transport_priority(MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY).unwrap();1303+ let delta_transport =1304+ media_v3_transport_priority(MEDIA_V3_DELTA_PUBLISHER_PRIORITY).unwrap();1305+ assert_eq!(keyframe_transport, MEDIA_V3_KEYFRAME_TRANSPORT_PRIORITY);1306+ assert_eq!(delta_transport, MEDIA_V3_DELTA_TRANSPORT_PRIORITY);1307+ assert!(keyframe_transport > delta_transport);1308+ assert!(media_v3_transport_priority(1).is_err());1309+ }1310+1311+ 1312+ fn current_gop_replay_is_complete_and_skips_already_sent_frames() {1313+ let frame = |sequence, keyframe| EncodedFrame {1314+ sequence,1315+ width: 1_280,1316+ height: 800,1317+ capture_timestamp_micros: sequence,1318+ presentation_timestamp_micros: sequence as i64,1319+ observed_at: std::time::Instant::now(),1320+ keyframe,1321+ codec_config: keyframe,1322+ discontinuity: false,1323+ data: Arc::from([sequence as u8]),1324+ };1325+1326+ let gop = || EncodedGop {1327+ frames: vec![frame(10, true), frame(11, false), frame(12, false)],1328+ payload_bytes: 3,1329+ };1330+1331+ let initial = new_current_gop_frames(gop(), None).collect::<Vec<_>>();1332+ assert_eq!(1333+ initial1334+ .iter()1335+ .map(|frame| frame.sequence)1336+ .collect::<Vec<_>>(),1337+ vec![10, 11, 12]1338+ );1339+ assert!(initial[0].keyframe && initial[0].codec_config);1340+1341+ let resumed = new_current_gop_frames(gop(), Some(10)).collect::<Vec<_>>();1342+ assert_eq!(1343+ resumed1344+ .iter()1345+ .map(|frame| frame.sequence)1346+ .collect::<Vec<_>>(),1347+ vec![11, 12]1348+ );1349+1350+ let behind_current_gop = new_current_gop_frames(gop(), Some(8)).collect::<Vec<_>>();1351+ assert_eq!(behind_current_gop[0].sequence, 10);1352+ assert!(behind_current_gop[0].keyframe && behind_current_gop[0].codec_config);1353+1354+ assert!(new_current_gop_frames(gop(), Some(12)).next().is_none());1355+ }1356+}