- commit
- 8b97c3bf7681f3c31a9693b03912e8de024a0145
- parent
- a0e7153ef0506b893e6eed5f8f6638444e19679a
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 839 | +----------------------------- |
| A | crates/ | 821 | +++++++++++++++++++++++++++++ |
@@ -40,12 +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_WRITE_TIMEOUT: Duration = Duration::from_secs(5);44-const MEDIA_V2_PEER_ACK_TIMEOUT: Duration = Duration::from_secs(2);45-const MEDIA_V2_IN_FLIGHT_CAPACITY: usize = 4;46-const MEDIA_V2_KEYFRAME_PRIORITY: i32 = 10;47-const MEDIA_V2_DELTA_PRIORITY: i32 = 0;48-const MEDIA_V2_RESET_CODE: u32 = 0x5356;4943 const MEDIA_V3_IN_FLIGHT_CAPACITY: usize = 4;5044 const MEDIA_V3_KEYFRAME_PUBLISHER_PRIORITY: u8 = 0;5145 const MEDIA_V3_DELTA_PUBLISHER_PRIORITY: u8 = 128;@@ -69,6 +63,7 @@ const MOQ_REJECT_CODE: u32 = 0x534d;
6963 const ENCODER_CONTROL_COMMIT_TIMEOUT: Duration = Duration::from_secs(2);7064 7165 mod adaptive;66+mod media_v2;7267 mod session;7368 mod startup;7469 @@ -76,6 +71,7 @@ mod startup;
7671 pub(crate) use adaptive::MotionResolutionPolicy;7772 pub(crate) use adaptive::VideoDimensions;7873 use adaptive::serve_media_feedback;74+use media_v2::{media_frame_for_encoded, serve_media, serve_media_v2};7975 8076 pub use session::SessionRegistry;8177 use session::{@@ -84,12 +80,6 @@ use session::{
8480 };8581 use startup::select_gamescope_startup_source;8682 87-88-enum MediaActivity {89- SourceChanged,90- PeerDisconnected,91-}92-9383 9484 enum MediaReplayDecision {9585 Send { discontinuity: bool },@@ -97,112 +87,6 @@ enum MediaReplayDecision {
9787 DiscardStaleSuffix { through_sequence: u64 },9888 }9989 100-101-enum MediaV2ScheduleDecision {102- Send {103- discontinuity: bool,104- cancel_sequences: Vec<u64>,105- },106- SkipUntilKeyframe,107- EnterResync {108- cancel_sequences: Vec<u64>,109- },110-}111-112-113-struct MediaV2Scheduler {114- in_flight: Vec<u64>,115- last_scheduled_sequence: Option<u64>,116- waiting_for_keyframe: bool,117- discontinuity_pending: bool,118-}119-120-impl Default for MediaV2Scheduler {121- fn default() -> Self {122- Self {123- in_flight: Vec::with_capacity(MEDIA_V2_IN_FLIGHT_CAPACITY),124- last_scheduled_sequence: None,125- waiting_for_keyframe: true,126- discontinuity_pending: false,127- }128- }129-}130-131-impl MediaV2Scheduler {132- fn schedule(133- &mut self,134- sequence: u64,135- independently_decodable: bool,136- ) -> MediaV2ScheduleDecision {137- if self138- .last_scheduled_sequence139- .is_some_and(|last| sequence <= last)140- {141- return MediaV2ScheduleDecision::SkipUntilKeyframe;142- }143-144- let sequence_discontinuity = self145- .last_scheduled_sequence146- .is_some_and(|last| last.checked_add(1) != Some(sequence));147- if independently_decodable {148- let cancel_sequences = std::mem::take(&mut self.in_flight);149- let discontinuity = self.discontinuity_pending150- || sequence_discontinuity151- || !cancel_sequences.is_empty();152- self.in_flight.push(sequence);153- self.last_scheduled_sequence = Some(sequence);154- self.waiting_for_keyframe = false;155- self.discontinuity_pending = false;156- return MediaV2ScheduleDecision::Send {157- discontinuity,158- cancel_sequences,159- };160- }161-162- if self.waiting_for_keyframe {163- return MediaV2ScheduleDecision::SkipUntilKeyframe;164- }165- if sequence_discontinuity || self.in_flight.len() == MEDIA_V2_IN_FLIGHT_CAPACITY {166- return MediaV2ScheduleDecision::EnterResync {167- cancel_sequences: self.enter_resync(),168- };169- }170-171- self.in_flight.push(sequence);172- self.last_scheduled_sequence = Some(sequence);173- MediaV2ScheduleDecision::Send {174- discontinuity: false,175- cancel_sequences: Vec::new(),176- }177- }178-179- fn complete(&mut self, sequence: u64) {180- if let Some(index) = self.in_flight.iter().position(|value| *value == sequence) {181- self.in_flight.swap_remove(index);182- }183- }184-185- fn fail(&mut self, sequence: u64) -> Vec<u64> {186- let Some(index) = self.in_flight.iter().position(|value| *value == sequence) else {187- // A completion from an already-cancelled GOP must not poison the188- // newer GOP which replaced it.189- return Vec::new();190- };191- self.in_flight.swap_remove(index);192- self.enter_resync()193- }194-195- fn fail_all(&mut self) -> Vec<u64> {196- self.enter_resync()197- }198-199- fn enter_resync(&mut self) -> Vec<u64> {200- self.waiting_for_keyframe = true;201- self.discontinuity_pending = true;202- std::mem::take(&mut self.in_flight)203- }204-}205-20690 20791 struct MediaV3ObjectPosition {20892 group_id: u64,@@ -692,34 +576,6 @@ impl Drop for ResetOnDropSendStreamV3 {
692576 }693577 }694578 695-struct ResetOnDropSendStream(Option<SendStream>);696-697-impl ResetOnDropSendStream {698- fn new(stream: SendStream) -> Self {699- Self(Some(stream))700- }701-702- fn stream(&self) -> &SendStream {703- self.0.as_ref().expect("send stream guard is armed")704- }705-706- fn stream_mut(&mut self) -> &mut SendStream {707- self.0.as_mut().expect("send stream guard is armed")708- }709-710- fn disarm(&mut self) {711- self.0.take();712- }713-}714-715-impl Drop for ResetOnDropSendStream {716- fn drop(&mut self) {717- if let Some(stream) = self.0.as_mut() {718- let _ = stream.reset(MEDIA_V2_RESET_CODE.into());719- }720- }721-}722-723579 724580 struct MediaReplayCursor {725581 last_sequence: Option<u64>,@@ -798,23 +654,6 @@ fn maximum_media_replay_age(framerate: u32) -> Duration {
798654 Duration::from_nanos(frame_period_nanos.saturating_mul(MAX_MEDIA_REPLAY_AGE_FRAME_PERIODS))799655 }800656 801-async fn wait_for_media_activity<T, F>(802- receiver: &mut tokio::sync::watch::Receiver<T>,803- peer_disconnected: Pin<&mut F>,804-) -> Result<MediaActivity>805-where806- T: Clone + Send + Sync,807- F: Future<Output = ()>,808-{809- tokio::select! {810- changed = receiver.changed() => {811- changed.context("encoded source stopped")?;812- Ok(MediaActivity::SourceChanged)813- }814- () = peer_disconnected => Ok(MediaActivity::PeerDisconnected),815- }816-}817-818657 819658 pub struct MediaHandler {820659 pub config: HostConfig,@@ -1395,273 +1234,19 @@ async fn run_control_moq_session(
13951234 }13961235 }13971236 }1398- }1399- }1400- }1401- .await;1402-1403- forced_idr.abort_and_drain(remote, "iroh-moq").await;1404- publisher.abort();1405- let _ = broadcast.abort(MoqError::Cancel);1406- if !control_task_finished {1407- control_task.abort();1408- let _ = control_task.await;1409- }1410- result1411-}1412-1413-async fn serve_media(1414- connection: Connection,1415- config: HostConfig,1416- sessions: &Arc<SessionRegistry>,1417- authorization: &AuthorizationPolicy,1418-) -> Result<()> {1419- let remote = connection.remote_id();1420- let handshake_permit = sessions1421- .pending_handshakes1422- .try_acquire()1423- .context("too many pending handshakes")?;1424- let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1425- .await1426- .context("timed out accepting media stream")?1427- .context("accepting media stream")?;1428- let hello = receive_hello(&mut recv, Capability::VideoH264).await?;1429- drop(handshake_permit);1430- debug!(%remote, agent = %hello.agent, "media hello received");1431-1432- let grants = match authorization.authorize_or_redeem(1433- remote,1434- hello.invitation.as_deref(),1435- unix_timestamp_now()?,1436- ) {1437- Ok(grants) => grants,1438- Err(error) => {1439- send_rejection(&mut send, "Portal peer is not authorized").await?;1440- return Err(error.context("authorizing media peer"));1441- }1442- };1443- ensure!(1444- grants.contains(InvitationGrants::VIEW),1445- "authorized media peer lacks view permission"1446- );1447-1448- let lease = match sessions.claim(remote, hello.nonce, grants) {1449- Ok(lease) => lease,1450- Err(error) => {1451- send_rejection(&mut send, "host already has an active client").await?;1452- return Err(error);1453- }1454- };1455-1456- // `serve` has already completed the static executable and encoder1457- // preflight. Resolve the live PipeWire node and create the bounded source1458- // before accepting the session so plugin discovery can never sit behind an1459- // accepted HostHello and the client's media-idle timeout.1460- let source = match config.source {1461- VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),1462- VideoSource::GamescopePipewire => {1463- let primary = spawn_gamescope_pipewire_after_static_preflight(1464- config.clone(),1465- lease.session_clock,1466- )1467- .await?;1468- select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await1469- }1470- };1471- let EncodedSource {1472- frames: frame_receiver,1473- current_gop: mut current_gop_receiver,1474- task: source_task,1475- pointer_surface_dimensions,1476- encoder_control,1477- } = match source {1478- Ok(source) => source,1479- Err(error) => {1480- send_rejection(&mut send, "video source is unavailable").await?;1481- return Err(error);1482- }1483- };1484- let source_task = SourceTaskGuard::new(source_task);1485- sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;1486- let _encoder_control = encoder_control;1487-1488- let mut media_hello = HostHello::accepted(1489- lease.session_id,1490- negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1491- );1492- if let Some(dimensions) = pointer_surface_dimensions {1493- media_hello = media_hello.with_pointer_surface_dimensions(dimensions);1494- }1495- write_host_hello(&mut send, &media_hello).await?;1496- info!(%remote, session_id = lease.session_id, "media client accepted");1497-1498- let session_result: Result<()> = async {1499- let maximum_replay_age = maximum_media_replay_age(config.framerate);1500- let mut replay_cursor = MediaReplayCursor::default();1501- let peer_disconnected = async {1502- let result = connection.closed().await;1503- debug!(%remote, ?result, "media connection closed");1504- };1505- tokio::pin!(peer_disconnected);1506- loop {1507- match wait_for_media_activity(&mut current_gop_receiver, peer_disconnected.as_mut())1508- .await?1509- {1510- MediaActivity::PeerDisconnected => return Ok(()),1511- MediaActivity::SourceChanged => {}1512- }1513- let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {1514- continue;1515- };1516- let initial_replay_started_at =1517- replay_cursor.last_sequence.is_none().then(Instant::now);1518- let replay_through_sequence = current_gop1519- .frames1520- .last()1521- .map(|frame| frame.sequence)1522- .context("current GOP snapshot is empty")?;1523- for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {1524- let discontinuity = match replay_cursor.classify(1525- &frame,1526- replay_through_sequence,1527- initial_replay_started_at,1528- Instant::now(),1529- maximum_replay_age,1530- ) {1531- MediaReplayDecision::Send { discontinuity } => discontinuity,1532- MediaReplayDecision::SkipUntilKeyframe => {1533- debug!(1534- sequence = frame.sequence,1535- "waiting for keyframe with codec configuration"1536- );1537- continue;1538- }1539- MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {1540- debug!(1541- sequence = frame.sequence,1542- through_sequence,1543- replay_age_micros = frame.observed_at.elapsed().as_micros(),1544- maximum_replay_age_micros = maximum_replay_age.as_micros(),1545- "discarding stale media suffix and waiting for keyframe"1546- );1547- break;1548- }1549- };1550- let media_frame = media_frame_for_encoded(&config, &frame, discontinuity)?;1551- tokio::time::timeout(1552- MEDIA_WRITE_TIMEOUT,1553- write_media_frame(&mut send, &media_frame),1554- )1555- .await1556- .context("timed out writing media frame")??;1557- replay_cursor.commit_sent(&frame);1558- }1559- }1560- }1561- .await;1562-1563- drop(current_gop_receiver);1564- drop(frame_receiver);1565- source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;1566- drop(lease);1567- info!(%remote, "media client released");1568- session_result1569-}1570-1571-async fn serve_media_v2(1572- connection: Connection,1573- config: HostConfig,1574- sessions: &Arc<SessionRegistry>,1575- authorization: &AuthorizationPolicy,1576-) -> Result<()> {1577- let remote = connection.remote_id();1578- let handshake_permit = sessions1579- .pending_handshakes1580- .try_acquire()1581- .context("too many pending handshakes")?;1582- let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())1583- .await1584- .context("timed out accepting media v2 handshake stream")?1585- .context("accepting media v2 handshake stream")?;1586- let hello = receive_hello(&mut recv, Capability::VideoH264).await?;1587- drop(handshake_permit);1588- debug!(%remote, agent = %hello.agent, "media v2 hello received");1589-1590- let grants = match authorization.authorize_or_redeem(1591- remote,1592- hello.invitation.as_deref(),1593- unix_timestamp_now()?,1594- ) {1595- Ok(grants) => grants,1596- Err(error) => {1597- send_rejection(&mut send, "Portal peer is not authorized").await?;1598- return Err(error.context("authorizing media v2 peer"));1599- }1600- };1601- ensure!(1602- grants.contains(InvitationGrants::VIEW),1603- "authorized media peer lacks view permission"1604- );1605-1606- let lease = match sessions.claim(remote, hello.nonce, grants) {1607- Ok(lease) => lease,1608- Err(error) => {1609- send_rejection(&mut send, "host already has an active client").await?;1610- return Err(error);1611- }1612- };1613-1614- let source = match config.source {1615- VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),1616- VideoSource::GamescopePipewire => {1617- let primary = spawn_gamescope_pipewire_after_static_preflight(1618- config.clone(),1619- lease.session_clock,1620- )1621- .await?;1622- select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await1623- }1624- };1625- let EncodedSource {1626- frames: frame_receiver,1627- current_gop: mut current_gop_receiver,1628- task: source_task,1629- pointer_surface_dimensions,1630- encoder_control,1631- } = match source {1632- Ok(source) => source,1633- Err(error) => {1634- send_rejection(&mut send, "video source is unavailable").await?;1635- return Err(error);1636- }1637- };1638- let source_task = SourceTaskGuard::new(source_task);1639- sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;1640- let _encoder_control = encoder_control;1641-1642- let mut media_hello = HostHello::accepted(1643- lease.session_id,1644- negotiated_capabilities(&hello, MEDIA_CAPABILITIES),1645- );1646- if let Some(dimensions) = pointer_surface_dimensions {1647- media_hello = media_hello.with_pointer_surface_dimensions(dimensions);1648- }1649- write_host_hello(&mut send, &media_hello).await?;1650- send.finish()1651- .context("finishing media v2 handshake response")?;1652- drop(send);1653- drop(recv);1654- info!(%remote, session_id = lease.session_id, "media v2 client accepted");1655-1656- let session_result =1657- run_media_v2_session(&connection, &config, &mut current_gop_receiver, remote).await;1237+ }1238+ }1239+ }1240+ .await;16581241 1659- drop(current_gop_receiver);1660- drop(frame_receiver);1661- source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;1662- drop(lease);1663- info!(%remote, "media v2 client released");1664- session_result1242+ forced_idr.abort_and_drain(remote, "iroh-moq").await;1243+ publisher.abort();1244+ let _ = broadcast.abort(MoqError::Cancel);1245+ if !control_task_finished {1246+ control_task.abort();1247+ let _ = control_task.await;1248+ }1249+ result16651250 }16661251 16671252 async fn serve_media_v3(@@ -1766,212 +1351,6 @@ async fn serve_media_v3(
17661351 session_result17671352 }17681353 1769-async fn run_media_v2_session(1770- connection: &Connection,1771- config: &HostConfig,1772- current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,1773- remote: EndpointId,1774-) -> Result<()> {1775- let maximum_replay_age = maximum_media_replay_age(config.framerate);1776- let mut replay_cursor = MediaReplayCursor::default();1777- let mut scheduler = MediaV2Scheduler::default();1778- let mut send_tasks = tokio::task::JoinSet::new();1779-1780- let result = loop {1781- tokio::select! {1782- closed = connection.closed() => {1783- debug!(%remote, ?closed, "media v2 connection closed");1784- break Ok(());1785- }1786- task = send_tasks.join_next(), if !send_tasks.is_empty() => {1787- match task.expect("guarded by non-empty send task set") {1788- Ok((sequence, Ok(()))) => scheduler.complete(sequence),1789- Ok((sequence, Err(error))) => {1790- warn!(sequence, %error, "media v2 object send failed; waiting for keyframe");1791- if !scheduler.fail(sequence).is_empty() {1792- send_tasks.abort_all();1793- }1794- }1795- Err(error) if error.is_cancelled() => {}1796- Err(error) => {1797- warn!(%error, "media v2 object task failed; waiting for keyframe");1798- if !scheduler.fail_all().is_empty() {1799- send_tasks.abort_all();1800- }1801- }1802- }1803- }1804- changed = current_gop_receiver.changed() => {1805- if let Err(error) = changed {1806- break Err(error).context("encoded source stopped");1807- }1808- let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {1809- continue;1810- };1811- let initial_replay_started_at =1812- replay_cursor.last_sequence.is_none().then(Instant::now);1813- let replay_through_sequence = current_gop1814- .frames1815- .last()1816- .map(|frame| frame.sequence)1817- .context("current GOP snapshot is empty")?;1818-1819- for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {1820- let replay_discontinuity = match replay_cursor.classify(1821- &frame,1822- replay_through_sequence,1823- initial_replay_started_at,1824- Instant::now(),1825- maximum_replay_age,1826- ) {1827- MediaReplayDecision::Send { discontinuity } => discontinuity,1828- MediaReplayDecision::SkipUntilKeyframe => continue,1829- MediaReplayDecision::DiscardStaleSuffix { .. } => break,1830- };1831- let independently_decodable = frame.keyframe && frame.codec_config;1832- let (scheduler_discontinuity, cancel_sequences) =1833- match scheduler.schedule(frame.sequence, independently_decodable) {1834- MediaV2ScheduleDecision::Send {1835- discontinuity,1836- cancel_sequences,1837- } => (discontinuity, cancel_sequences),1838- MediaV2ScheduleDecision::SkipUntilKeyframe => continue,1839- MediaV2ScheduleDecision::EnterResync { cancel_sequences } => {1840- if !cancel_sequences.is_empty() {1841- send_tasks.abort_all();1842- }1843- continue;1844- }1845- };1846- if !cancel_sequences.is_empty() {1847- debug!(1848- sequence = frame.sequence,1849- ?cancel_sequences,1850- "keyframe superseding media v2 objects"1851- );1852- send_tasks.abort_all();1853- }1854-1855- let media_frame = media_frame_for_encoded(1856- config,1857- &frame,1858- replay_discontinuity || scheduler_discontinuity,1859- )?;1860- let sequence = frame.sequence;1861- let keyframe = independently_decodable;1862- // Reserve stream IDs in encoded-frame order. Opening them1863- // inside the concurrent writer tasks would let task1864- // scheduling reorder the QUIC object sequence even though1865- // each object carries a monotonic media sequence number.1866- let stream = match tokio::time::timeout(1867- MEDIA_WRITE_TIMEOUT,1868- connection.open_uni(),1869- )1870- .await1871- {1872- Ok(Ok(stream)) => stream,1873- Ok(Err(error)) => {1874- warn!(sequence, %error, "opening media v2 object stream failed");1875- if !scheduler.fail(sequence).is_empty() {1876- send_tasks.abort_all();1877- }1878- continue;1879- }1880- Err(_) => {1881- warn!(sequence, "opening media v2 object stream timed out");1882- if !scheduler.fail(sequence).is_empty() {1883- send_tasks.abort_all();1884- }1885- continue;1886- }1887- };1888- let stream = ResetOnDropSendStream::new(stream);1889- if let Err(error) = stream1890- .stream()1891- .set_priority(media_v2_priority(keyframe))1892- {1893- warn!(sequence, %error, "setting media v2 object priority failed");1894- if !scheduler.fail(sequence).is_empty() {1895- send_tasks.abort_all();1896- }1897- continue;1898- }1899- send_tasks.spawn(async move {1900- (1901- sequence,1902- send_media_v2_object(stream, media_frame).await,1903- )1904- });1905- replay_cursor.commit_sent(&frame);1906- }1907- }1908- }1909- };1910-1911- send_tasks.abort_all();1912- while send_tasks.join_next().await.is_some() {}1913- result1914-}1915-1916-fn media_frame_for_encoded(1917- _config: &HostConfig,1918- frame: &EncodedFrame,1919- discontinuity: bool,1920-) -> Result<MediaFrame> {1921- let mut flags = FrameFlags::NONE;1922- if frame.keyframe {1923- flags = flags.union(FrameFlags::KEYFRAME);1924- }1925- if frame.codec_config {1926- flags = flags.union(FrameFlags::CODEC_CONFIG);1927- }1928- if discontinuity || frame.discontinuity {1929- flags = flags.union(FrameFlags::DISCONTINUITY);1930- }1931- let header = MediaFrameHeader::h264(1932- frame.width,1933- frame.height,1934- frame.data.len(),1935- frame.sequence,1936- frame.capture_timestamp_micros,1937- frame.presentation_timestamp_micros,1938- flags,1939- )?;1940- MediaFrame::new(header, frame.data.as_ref().to_vec()).map_err(Into::into)1941-}1942-1943-async fn send_media_v2_object(mut stream: ResetOnDropSendStream, frame: MediaFrame) -> Result<()> {1944- tokio::time::timeout(1945- MEDIA_WRITE_TIMEOUT,1946- write_media_frame(stream.stream_mut(), &frame),1947- )1948- .await1949- .context("timed out writing media v2 object")??;1950- stream1951- .stream_mut()1952- .finish()1953- .context("finishing media v2 object stream")?;1954- match tokio::time::timeout(MEDIA_V2_PEER_ACK_TIMEOUT, stream.stream().stopped())1955- .await1956- .context("timed out waiting for media v2 object acknowledgement")?1957- .context("waiting for media v2 object acknowledgement")?1958- {1959- None => {1960- stream.disarm();1961- Ok(())1962- }1963- Some(code) => bail!("peer stopped media v2 object stream with code {code}"),1964- }1965-}1966-1967-fn media_v2_priority(keyframe: bool) -> i32 {1968- if keyframe {1969- MEDIA_V2_KEYFRAME_PRIORITY1970- } else {1971- MEDIA_V2_DELTA_PRIORITY1972- }1973-}1974-19751354 async fn run_media_v3_session(19761355 connection: &Connection,19771356 config: &HostConfig,@@ -2862,127 +2241,6 @@ impl Drop for DropNotify {
28622241 mod tests {28632242 use super::*;28642243 2865- 2866- fn media_v2_scheduler_is_bounded_and_drops_a_saturated_delta_suffix() {2867- let mut scheduler = MediaV2Scheduler::default();2868- assert_eq!(2869- scheduler.schedule(10, true),2870- MediaV2ScheduleDecision::Send {2871- discontinuity: false,2872- cancel_sequences: vec![],2873- }2874- );2875- for sequence in 11..14 {2876- assert!(matches!(2877- scheduler.schedule(sequence, false),2878- MediaV2ScheduleDecision::Send { .. }2879- ));2880- }2881- assert_eq!(scheduler.in_flight.len(), MEDIA_V2_IN_FLIGHT_CAPACITY);2882-2883- assert_eq!(2884- scheduler.schedule(14, false),2885- MediaV2ScheduleDecision::EnterResync {2886- cancel_sequences: vec![10, 11, 12, 13],2887- }2888- );2889- assert!(scheduler.in_flight.is_empty());2890- assert_eq!(2891- scheduler.schedule(15, false),2892- MediaV2ScheduleDecision::SkipUntilKeyframe2893- );2894- }2895-2896- 2897- fn blocked_old_streams_cannot_prevent_a_keyframe_decision() {2898- let mut scheduler = MediaV2Scheduler::default();2899- for sequence in 20..24 {2900- assert!(matches!(2901- scheduler.schedule(sequence, sequence == 20),2902- MediaV2ScheduleDecision::Send { .. }2903- ));2904- }2905- assert_eq!(scheduler.in_flight.len(), MEDIA_V2_IN_FLIGHT_CAPACITY);2906-2907- assert_eq!(2908- scheduler.schedule(30, true),2909- MediaV2ScheduleDecision::Send {2910- discontinuity: true,2911- cancel_sequences: vec![20, 21, 22, 23],2912- }2913- );2914- assert_eq!(scheduler.in_flight, vec![30]);2915- }2916-2917- 2918- fn media_v2_send_failure_cancels_dependents_and_resyncs_on_keyframe() {2919- let mut scheduler = MediaV2Scheduler::default();2920- assert!(matches!(2921- scheduler.schedule(40, true),2922- MediaV2ScheduleDecision::Send { .. }2923- ));2924- assert!(matches!(2925- scheduler.schedule(41, false),2926- MediaV2ScheduleDecision::Send { .. }2927- ));2928-2929- assert_eq!(scheduler.fail(40), vec![41]);2930- assert_eq!(2931- scheduler.schedule(42, false),2932- MediaV2ScheduleDecision::SkipUntilKeyframe2933- );2934- assert_eq!(2935- scheduler.schedule(50, true),2936- MediaV2ScheduleDecision::Send {2937- discontinuity: true,2938- cancel_sequences: vec![],2939- }2940- );2941- scheduler.complete(50);2942- assert!(scheduler.in_flight.is_empty());2943- assert!(matches!(2944- scheduler.schedule(51, false),2945- MediaV2ScheduleDecision::Send {2946- discontinuity: false,2947- ..2948- }2949- ));2950- }2951-2952- 2953- fn cancelled_old_completion_cannot_poison_the_replacement_gop() {2954- let mut scheduler = MediaV2Scheduler::default();2955- assert!(matches!(2956- scheduler.schedule(60, true),2957- MediaV2ScheduleDecision::Send { .. }2958- ));2959- assert!(matches!(2960- scheduler.schedule(61, false),2961- MediaV2ScheduleDecision::Send { .. }2962- ));2963- assert!(matches!(2964- scheduler.schedule(70, true),2965- MediaV2ScheduleDecision::Send {2966- discontinuity: true,2967- ..2968- }2969- ));2970-2971- assert!(scheduler.fail(60).is_empty());2972- assert!(matches!(2973- scheduler.schedule(71, false),2974- MediaV2ScheduleDecision::Send {2975- discontinuity: false,2976- ..2977- }2978- ));2979- }2980-2981- 2982- fn media_v2_keyframes_have_strictly_higher_transport_priority() {2983- assert!(media_v2_priority(true) > media_v2_priority(false));2984- }2985-29862244 fn media_v3_encoded_frame(29872245 sequence: u64,29882246 keyframe: bool,@@ -4055,73 +3313,4 @@ mod tests {
40553313 }40563314 );40573315 }4058-4059- 4060- async fn no_frame_peer_drop_reaps_source_and_allows_reconnect() {4061- let sessions = Arc::new(SessionRegistry::default());4062- let remote = endpoint(1);4063- let nonce = [7; 16];4064- let media = sessions4065- .claim(remote, nonce, InvitationGrants::ALL)4066- .unwrap();4067- let input = sessions.claim_input(remote, nonce).unwrap();4068- let session_id = media.session_id;4069-4070- let input_sessions = Arc::clone(&sessions);4071- let input_shutdown = tokio::spawn(async move {4072- loop {4073- let notified = input_sessions.session_changed.notified();4074- if !input_sessions.is_active(remote, session_id) {4075- break;4076- }4077- notified.await;4078- }4079- drop(input);4080- });4081-4082- let (_frame_sender, mut frame_receiver) =4083- tokio::sync::watch::channel(Option::<EncodedFrame>::None);4084- let (source_started_tx, source_started_rx) = tokio::sync::oneshot::channel();4085- let (source_reaped_tx, source_reaped_rx) = tokio::sync::oneshot::channel();4086- let source_task = tokio::spawn(async move {4087- let _notify = DropNotify(Some(source_reaped_tx));4088- let _ = source_started_tx.send(());4089- std::future::pending::<Result<()>>().await4090- });4091- let source_task = SourceTaskGuard::new(source_task);4092- source_started_rx.await.unwrap();4093-4094- let (peer_alive, peer_disconnected) = tokio::sync::oneshot::channel::<()>();4095- let disconnected = async move {4096- let _ = peer_disconnected.await;4097- };4098- tokio::pin!(disconnected);4099- drop(peer_alive);4100-4101- let activity = tokio::time::timeout(4102- Duration::from_millis(100),4103- wait_for_media_activity(&mut frame_receiver, disconnected.as_mut()),4104- )4105- .await4106- .expect("no-frame media loop ignored peer disconnect")4107- .unwrap();4108- assert_eq!(activity, MediaActivity::PeerDisconnected);4109-4110- source_task.abort_and_wait().await;4111- tokio::time::timeout(Duration::from_millis(100), source_reaped_rx)4112- .await4113- .expect("source task was not reaped after peer disconnect")4114- .unwrap();4115- drop(media);4116- tokio::time::timeout(Duration::from_millis(100), input_shutdown)4117- .await4118- .expect("input lease did not observe media shutdown")4119- .unwrap();4120-4121- assert!(4122- sessions4123- .claim(endpoint(2), [8; 16], InvitationGrants::ALL)4124- .is_ok()4125- );4126- }41273316 }@@ -0,0 +1,821 @@
1+use super::*;2+3+const MEDIA_WRITE_TIMEOUT: Duration = Duration::from_secs(5);4+const MEDIA_V2_PEER_ACK_TIMEOUT: Duration = Duration::from_secs(2);5+const MEDIA_V2_IN_FLIGHT_CAPACITY: usize = 4;6+const MEDIA_V2_KEYFRAME_PRIORITY: i32 = 10;7+const MEDIA_V2_DELTA_PRIORITY: i32 = 0;8+const MEDIA_V2_RESET_CODE: u32 = 0x5356;9+10+11+enum MediaActivity {12+ SourceChanged,13+ PeerDisconnected,14+}15+16+17+enum MediaV2ScheduleDecision {18+ Send {19+ discontinuity: bool,20+ cancel_sequences: Vec<u64>,21+ },22+ SkipUntilKeyframe,23+ EnterResync {24+ cancel_sequences: Vec<u64>,25+ },26+}27+28+29+struct MediaV2Scheduler {30+ in_flight: Vec<u64>,31+ last_scheduled_sequence: Option<u64>,32+ waiting_for_keyframe: bool,33+ discontinuity_pending: bool,34+}35+36+impl Default for MediaV2Scheduler {37+ fn default() -> Self {38+ Self {39+ in_flight: Vec::with_capacity(MEDIA_V2_IN_FLIGHT_CAPACITY),40+ last_scheduled_sequence: None,41+ waiting_for_keyframe: true,42+ discontinuity_pending: false,43+ }44+ }45+}46+47+impl MediaV2Scheduler {48+ fn schedule(49+ &mut self,50+ sequence: u64,51+ independently_decodable: bool,52+ ) -> MediaV2ScheduleDecision {53+ if self54+ .last_scheduled_sequence55+ .is_some_and(|last| sequence <= last)56+ {57+ return MediaV2ScheduleDecision::SkipUntilKeyframe;58+ }59+60+ let sequence_discontinuity = self61+ .last_scheduled_sequence62+ .is_some_and(|last| last.checked_add(1) != Some(sequence));63+ if independently_decodable {64+ let cancel_sequences = std::mem::take(&mut self.in_flight);65+ let discontinuity = self.discontinuity_pending66+ || sequence_discontinuity67+ || !cancel_sequences.is_empty();68+ self.in_flight.push(sequence);69+ self.last_scheduled_sequence = Some(sequence);70+ self.waiting_for_keyframe = false;71+ self.discontinuity_pending = false;72+ return MediaV2ScheduleDecision::Send {73+ discontinuity,74+ cancel_sequences,75+ };76+ }77+78+ if self.waiting_for_keyframe {79+ return MediaV2ScheduleDecision::SkipUntilKeyframe;80+ }81+ if sequence_discontinuity || self.in_flight.len() == MEDIA_V2_IN_FLIGHT_CAPACITY {82+ return MediaV2ScheduleDecision::EnterResync {83+ cancel_sequences: self.enter_resync(),84+ };85+ }86+87+ self.in_flight.push(sequence);88+ self.last_scheduled_sequence = Some(sequence);89+ MediaV2ScheduleDecision::Send {90+ discontinuity: false,91+ cancel_sequences: Vec::new(),92+ }93+ }94+95+ fn complete(&mut self, sequence: u64) {96+ if let Some(index) = self.in_flight.iter().position(|value| *value == sequence) {97+ self.in_flight.swap_remove(index);98+ }99+ }100+101+ fn fail(&mut self, sequence: u64) -> Vec<u64> {102+ let Some(index) = self.in_flight.iter().position(|value| *value == sequence) else {103+ // A completion from an already-cancelled GOP must not poison the104+ // newer GOP which replaced it.105+ return Vec::new();106+ };107+ self.in_flight.swap_remove(index);108+ self.enter_resync()109+ }110+111+ fn fail_all(&mut self) -> Vec<u64> {112+ self.enter_resync()113+ }114+115+ fn enter_resync(&mut self) -> Vec<u64> {116+ self.waiting_for_keyframe = true;117+ self.discontinuity_pending = true;118+ std::mem::take(&mut self.in_flight)119+ }120+}121+122+struct ResetOnDropSendStream(Option<SendStream>);123+124+impl ResetOnDropSendStream {125+ fn new(stream: SendStream) -> Self {126+ Self(Some(stream))127+ }128+129+ fn stream(&self) -> &SendStream {130+ self.0.as_ref().expect("send stream guard is armed")131+ }132+133+ fn stream_mut(&mut self) -> &mut SendStream {134+ self.0.as_mut().expect("send stream guard is armed")135+ }136+137+ fn disarm(&mut self) {138+ self.0.take();139+ }140+}141+142+impl Drop for ResetOnDropSendStream {143+ fn drop(&mut self) {144+ if let Some(stream) = self.0.as_mut() {145+ let _ = stream.reset(MEDIA_V2_RESET_CODE.into());146+ }147+ }148+}149+150+async fn wait_for_media_activity<T, F>(151+ receiver: &mut tokio::sync::watch::Receiver<T>,152+ peer_disconnected: Pin<&mut F>,153+) -> Result<MediaActivity>154+where155+ T: Clone + Send + Sync,156+ F: Future<Output = ()>,157+{158+ tokio::select! {159+ changed = receiver.changed() => {160+ changed.context("encoded source stopped")?;161+ Ok(MediaActivity::SourceChanged)162+ }163+ () = peer_disconnected => Ok(MediaActivity::PeerDisconnected),164+ }165+}166+167+pub(super) async fn serve_media(168+ connection: Connection,169+ config: HostConfig,170+ sessions: &Arc<SessionRegistry>,171+ authorization: &AuthorizationPolicy,172+) -> Result<()> {173+ let remote = connection.remote_id();174+ let handshake_permit = sessions175+ .pending_handshakes176+ .try_acquire()177+ .context("too many pending handshakes")?;178+ let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())179+ .await180+ .context("timed out accepting media stream")?181+ .context("accepting media stream")?;182+ let hello = receive_hello(&mut recv, Capability::VideoH264).await?;183+ drop(handshake_permit);184+ debug!(%remote, agent = %hello.agent, "media hello received");185+186+ let grants = match authorization.authorize_or_redeem(187+ remote,188+ hello.invitation.as_deref(),189+ unix_timestamp_now()?,190+ ) {191+ Ok(grants) => grants,192+ Err(error) => {193+ send_rejection(&mut send, "Portal peer is not authorized").await?;194+ return Err(error.context("authorizing media peer"));195+ }196+ };197+ ensure!(198+ grants.contains(InvitationGrants::VIEW),199+ "authorized media peer lacks view permission"200+ );201+202+ let lease = match sessions.claim(remote, hello.nonce, grants) {203+ Ok(lease) => lease,204+ Err(error) => {205+ send_rejection(&mut send, "host already has an active client").await?;206+ return Err(error);207+ }208+ };209+210+ // `serve` has already completed the static executable and encoder211+ // preflight. Resolve the live PipeWire node and create the bounded source212+ // before accepting the session so plugin discovery can never sit behind an213+ // accepted HostHello and the client's media-idle timeout.214+ let source = match config.source {215+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),216+ VideoSource::GamescopePipewire => {217+ let primary = spawn_gamescope_pipewire_after_static_preflight(218+ config.clone(),219+ lease.session_clock,220+ )221+ .await?;222+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await223+ }224+ };225+ let EncodedSource {226+ frames: frame_receiver,227+ current_gop: mut current_gop_receiver,228+ task: source_task,229+ pointer_surface_dimensions,230+ encoder_control,231+ } = match source {232+ Ok(source) => source,233+ Err(error) => {234+ send_rejection(&mut send, "video source is unavailable").await?;235+ return Err(error);236+ }237+ };238+ let source_task = SourceTaskGuard::new(source_task);239+ sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;240+ let _encoder_control = encoder_control;241+242+ let mut media_hello = HostHello::accepted(243+ lease.session_id,244+ negotiated_capabilities(&hello, MEDIA_CAPABILITIES),245+ );246+ if let Some(dimensions) = pointer_surface_dimensions {247+ media_hello = media_hello.with_pointer_surface_dimensions(dimensions);248+ }249+ write_host_hello(&mut send, &media_hello).await?;250+ info!(%remote, session_id = lease.session_id, "media client accepted");251+252+ let session_result: Result<()> = async {253+ let maximum_replay_age = maximum_media_replay_age(config.framerate);254+ let mut replay_cursor = MediaReplayCursor::default();255+ let peer_disconnected = async {256+ let result = connection.closed().await;257+ debug!(%remote, ?result, "media connection closed");258+ };259+ tokio::pin!(peer_disconnected);260+ loop {261+ match wait_for_media_activity(&mut current_gop_receiver, peer_disconnected.as_mut())262+ .await?263+ {264+ MediaActivity::PeerDisconnected => return Ok(()),265+ MediaActivity::SourceChanged => {}266+ }267+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {268+ continue;269+ };270+ let initial_replay_started_at =271+ replay_cursor.last_sequence.is_none().then(Instant::now);272+ let replay_through_sequence = current_gop273+ .frames274+ .last()275+ .map(|frame| frame.sequence)276+ .context("current GOP snapshot is empty")?;277+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {278+ let discontinuity = match replay_cursor.classify(279+ &frame,280+ replay_through_sequence,281+ initial_replay_started_at,282+ Instant::now(),283+ maximum_replay_age,284+ ) {285+ MediaReplayDecision::Send { discontinuity } => discontinuity,286+ MediaReplayDecision::SkipUntilKeyframe => {287+ debug!(288+ sequence = frame.sequence,289+ "waiting for keyframe with codec configuration"290+ );291+ continue;292+ }293+ MediaReplayDecision::DiscardStaleSuffix { through_sequence } => {294+ debug!(295+ sequence = frame.sequence,296+ through_sequence,297+ replay_age_micros = frame.observed_at.elapsed().as_micros(),298+ maximum_replay_age_micros = maximum_replay_age.as_micros(),299+ "discarding stale media suffix and waiting for keyframe"300+ );301+ break;302+ }303+ };304+ let media_frame = media_frame_for_encoded(&config, &frame, discontinuity)?;305+ tokio::time::timeout(306+ MEDIA_WRITE_TIMEOUT,307+ write_media_frame(&mut send, &media_frame),308+ )309+ .await310+ .context("timed out writing media frame")??;311+ replay_cursor.commit_sent(&frame);312+ }313+ }314+ }315+ .await;316+317+ drop(current_gop_receiver);318+ drop(frame_receiver);319+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;320+ drop(lease);321+ info!(%remote, "media client released");322+ session_result323+}324+325+pub(super) async fn serve_media_v2(326+ connection: Connection,327+ config: HostConfig,328+ sessions: &Arc<SessionRegistry>,329+ authorization: &AuthorizationPolicy,330+) -> Result<()> {331+ let remote = connection.remote_id();332+ let handshake_permit = sessions333+ .pending_handshakes334+ .try_acquire()335+ .context("too many pending handshakes")?;336+ let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())337+ .await338+ .context("timed out accepting media v2 handshake stream")?339+ .context("accepting media v2 handshake stream")?;340+ let hello = receive_hello(&mut recv, Capability::VideoH264).await?;341+ drop(handshake_permit);342+ debug!(%remote, agent = %hello.agent, "media v2 hello received");343+344+ let grants = match authorization.authorize_or_redeem(345+ remote,346+ hello.invitation.as_deref(),347+ unix_timestamp_now()?,348+ ) {349+ Ok(grants) => grants,350+ Err(error) => {351+ send_rejection(&mut send, "Portal peer is not authorized").await?;352+ return Err(error.context("authorizing media v2 peer"));353+ }354+ };355+ ensure!(356+ grants.contains(InvitationGrants::VIEW),357+ "authorized media peer lacks view permission"358+ );359+360+ let lease = match sessions.claim(remote, hello.nonce, grants) {361+ Ok(lease) => lease,362+ Err(error) => {363+ send_rejection(&mut send, "host already has an active client").await?;364+ return Err(error);365+ }366+ };367+368+ let source = match config.source {369+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),370+ VideoSource::GamescopePipewire => {371+ let primary = spawn_gamescope_pipewire_after_static_preflight(372+ config.clone(),373+ lease.session_clock,374+ )375+ .await?;376+ select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await377+ }378+ };379+ let EncodedSource {380+ frames: frame_receiver,381+ current_gop: mut current_gop_receiver,382+ task: source_task,383+ pointer_surface_dimensions,384+ encoder_control,385+ } = match source {386+ Ok(source) => source,387+ Err(error) => {388+ send_rejection(&mut send, "video source is unavailable").await?;389+ return Err(error);390+ }391+ };392+ let source_task = SourceTaskGuard::new(source_task);393+ sessions.install_encoder_control(remote, lease.session_id, encoder_control.clone())?;394+ let _encoder_control = encoder_control;395+396+ let mut media_hello = HostHello::accepted(397+ lease.session_id,398+ negotiated_capabilities(&hello, MEDIA_CAPABILITIES),399+ );400+ if let Some(dimensions) = pointer_surface_dimensions {401+ media_hello = media_hello.with_pointer_surface_dimensions(dimensions);402+ }403+ write_host_hello(&mut send, &media_hello).await?;404+ send.finish()405+ .context("finishing media v2 handshake response")?;406+ drop(send);407+ drop(recv);408+ info!(%remote, session_id = lease.session_id, "media v2 client accepted");409+410+ let session_result =411+ run_media_v2_session(&connection, &config, &mut current_gop_receiver, remote).await;412+413+ drop(current_gop_receiver);414+ drop(frame_receiver);415+ source_task.wait_or_abort(SOURCE_REAP_GRACE_TIMEOUT).await;416+ drop(lease);417+ info!(%remote, "media v2 client released");418+ session_result419+}420+421+async fn run_media_v2_session(422+ connection: &Connection,423+ config: &HostConfig,424+ current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,425+ remote: EndpointId,426+) -> Result<()> {427+ let maximum_replay_age = maximum_media_replay_age(config.framerate);428+ let mut replay_cursor = MediaReplayCursor::default();429+ let mut scheduler = MediaV2Scheduler::default();430+ let mut send_tasks = tokio::task::JoinSet::new();431+432+ let result = loop {433+ tokio::select! {434+ closed = connection.closed() => {435+ debug!(%remote, ?closed, "media v2 connection closed");436+ break Ok(());437+ }438+ task = send_tasks.join_next(), if !send_tasks.is_empty() => {439+ match task.expect("guarded by non-empty send task set") {440+ Ok((sequence, Ok(()))) => scheduler.complete(sequence),441+ Ok((sequence, Err(error))) => {442+ warn!(sequence, %error, "media v2 object send failed; waiting for keyframe");443+ if !scheduler.fail(sequence).is_empty() {444+ send_tasks.abort_all();445+ }446+ }447+ Err(error) if error.is_cancelled() => {}448+ Err(error) => {449+ warn!(%error, "media v2 object task failed; waiting for keyframe");450+ if !scheduler.fail_all().is_empty() {451+ send_tasks.abort_all();452+ }453+ }454+ }455+ }456+ changed = current_gop_receiver.changed() => {457+ if let Err(error) = changed {458+ break Err(error).context("encoded source stopped");459+ }460+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {461+ continue;462+ };463+ let initial_replay_started_at =464+ replay_cursor.last_sequence.is_none().then(Instant::now);465+ let replay_through_sequence = current_gop466+ .frames467+ .last()468+ .map(|frame| frame.sequence)469+ .context("current GOP snapshot is empty")?;470+471+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {472+ let replay_discontinuity = match replay_cursor.classify(473+ &frame,474+ replay_through_sequence,475+ initial_replay_started_at,476+ Instant::now(),477+ maximum_replay_age,478+ ) {479+ MediaReplayDecision::Send { discontinuity } => discontinuity,480+ MediaReplayDecision::SkipUntilKeyframe => continue,481+ MediaReplayDecision::DiscardStaleSuffix { .. } => break,482+ };483+ let independently_decodable = frame.keyframe && frame.codec_config;484+ let (scheduler_discontinuity, cancel_sequences) =485+ match scheduler.schedule(frame.sequence, independently_decodable) {486+ MediaV2ScheduleDecision::Send {487+ discontinuity,488+ cancel_sequences,489+ } => (discontinuity, cancel_sequences),490+ MediaV2ScheduleDecision::SkipUntilKeyframe => continue,491+ MediaV2ScheduleDecision::EnterResync { cancel_sequences } => {492+ if !cancel_sequences.is_empty() {493+ send_tasks.abort_all();494+ }495+ continue;496+ }497+ };498+ if !cancel_sequences.is_empty() {499+ debug!(500+ sequence = frame.sequence,501+ ?cancel_sequences,502+ "keyframe superseding media v2 objects"503+ );504+ send_tasks.abort_all();505+ }506+507+ let media_frame = media_frame_for_encoded(508+ config,509+ &frame,510+ replay_discontinuity || scheduler_discontinuity,511+ )?;512+ let sequence = frame.sequence;513+ let keyframe = independently_decodable;514+ // Reserve stream IDs in encoded-frame order. Opening them515+ // inside the concurrent writer tasks would let task516+ // scheduling reorder the QUIC object sequence even though517+ // each object carries a monotonic media sequence number.518+ let stream = match tokio::time::timeout(519+ MEDIA_WRITE_TIMEOUT,520+ connection.open_uni(),521+ )522+ .await523+ {524+ Ok(Ok(stream)) => stream,525+ Ok(Err(error)) => {526+ warn!(sequence, %error, "opening media v2 object stream failed");527+ if !scheduler.fail(sequence).is_empty() {528+ send_tasks.abort_all();529+ }530+ continue;531+ }532+ Err(_) => {533+ warn!(sequence, "opening media v2 object stream timed out");534+ if !scheduler.fail(sequence).is_empty() {535+ send_tasks.abort_all();536+ }537+ continue;538+ }539+ };540+ let stream = ResetOnDropSendStream::new(stream);541+ if let Err(error) = stream542+ .stream()543+ .set_priority(media_v2_priority(keyframe))544+ {545+ warn!(sequence, %error, "setting media v2 object priority failed");546+ if !scheduler.fail(sequence).is_empty() {547+ send_tasks.abort_all();548+ }549+ continue;550+ }551+ send_tasks.spawn(async move {552+ (553+ sequence,554+ send_media_v2_object(stream, media_frame).await,555+ )556+ });557+ replay_cursor.commit_sent(&frame);558+ }559+ }560+ }561+ };562+563+ send_tasks.abort_all();564+ while send_tasks.join_next().await.is_some() {}565+ result566+}567+568+pub(super) fn media_frame_for_encoded(569+ _config: &HostConfig,570+ frame: &EncodedFrame,571+ discontinuity: bool,572+) -> Result<MediaFrame> {573+ let mut flags = FrameFlags::NONE;574+ if frame.keyframe {575+ flags = flags.union(FrameFlags::KEYFRAME);576+ }577+ if frame.codec_config {578+ flags = flags.union(FrameFlags::CODEC_CONFIG);579+ }580+ if discontinuity || frame.discontinuity {581+ flags = flags.union(FrameFlags::DISCONTINUITY);582+ }583+ let header = MediaFrameHeader::h264(584+ frame.width,585+ frame.height,586+ frame.data.len(),587+ frame.sequence,588+ frame.capture_timestamp_micros,589+ frame.presentation_timestamp_micros,590+ flags,591+ )?;592+ MediaFrame::new(header, frame.data.as_ref().to_vec()).map_err(Into::into)593+}594+595+async fn send_media_v2_object(mut stream: ResetOnDropSendStream, frame: MediaFrame) -> Result<()> {596+ tokio::time::timeout(597+ MEDIA_WRITE_TIMEOUT,598+ write_media_frame(stream.stream_mut(), &frame),599+ )600+ .await601+ .context("timed out writing media v2 object")??;602+ stream603+ .stream_mut()604+ .finish()605+ .context("finishing media v2 object stream")?;606+ match tokio::time::timeout(MEDIA_V2_PEER_ACK_TIMEOUT, stream.stream().stopped())607+ .await608+ .context("timed out waiting for media v2 object acknowledgement")?609+ .context("waiting for media v2 object acknowledgement")?610+ {611+ None => {612+ stream.disarm();613+ Ok(())614+ }615+ Some(code) => bail!("peer stopped media v2 object stream with code {code}"),616+ }617+}618+619+fn media_v2_priority(keyframe: bool) -> i32 {620+ if keyframe {621+ MEDIA_V2_KEYFRAME_PRIORITY622+ } else {623+ MEDIA_V2_DELTA_PRIORITY624+ }625+}626+627+628+mod tests {629+ use super::super::{DropNotify, endpoint};630+ use super::*;631+632+ 633+ fn media_v2_scheduler_is_bounded_and_drops_a_saturated_delta_suffix() {634+ let mut scheduler = MediaV2Scheduler::default();635+ assert_eq!(636+ scheduler.schedule(10, true),637+ MediaV2ScheduleDecision::Send {638+ discontinuity: false,639+ cancel_sequences: vec![],640+ }641+ );642+ for sequence in 11..14 {643+ assert!(matches!(644+ scheduler.schedule(sequence, false),645+ MediaV2ScheduleDecision::Send { .. }646+ ));647+ }648+ assert_eq!(scheduler.in_flight.len(), MEDIA_V2_IN_FLIGHT_CAPACITY);649+650+ assert_eq!(651+ scheduler.schedule(14, false),652+ MediaV2ScheduleDecision::EnterResync {653+ cancel_sequences: vec![10, 11, 12, 13],654+ }655+ );656+ assert!(scheduler.in_flight.is_empty());657+ assert_eq!(658+ scheduler.schedule(15, false),659+ MediaV2ScheduleDecision::SkipUntilKeyframe660+ );661+ }662+663+ 664+ fn blocked_old_streams_cannot_prevent_a_keyframe_decision() {665+ let mut scheduler = MediaV2Scheduler::default();666+ for sequence in 20..24 {667+ assert!(matches!(668+ scheduler.schedule(sequence, sequence == 20),669+ MediaV2ScheduleDecision::Send { .. }670+ ));671+ }672+ assert_eq!(scheduler.in_flight.len(), MEDIA_V2_IN_FLIGHT_CAPACITY);673+674+ assert_eq!(675+ scheduler.schedule(30, true),676+ MediaV2ScheduleDecision::Send {677+ discontinuity: true,678+ cancel_sequences: vec![20, 21, 22, 23],679+ }680+ );681+ assert_eq!(scheduler.in_flight, vec![30]);682+ }683+684+ 685+ fn media_v2_send_failure_cancels_dependents_and_resyncs_on_keyframe() {686+ let mut scheduler = MediaV2Scheduler::default();687+ assert!(matches!(688+ scheduler.schedule(40, true),689+ MediaV2ScheduleDecision::Send { .. }690+ ));691+ assert!(matches!(692+ scheduler.schedule(41, false),693+ MediaV2ScheduleDecision::Send { .. }694+ ));695+696+ assert_eq!(scheduler.fail(40), vec![41]);697+ assert_eq!(698+ scheduler.schedule(42, false),699+ MediaV2ScheduleDecision::SkipUntilKeyframe700+ );701+ assert_eq!(702+ scheduler.schedule(50, true),703+ MediaV2ScheduleDecision::Send {704+ discontinuity: true,705+ cancel_sequences: vec![],706+ }707+ );708+ scheduler.complete(50);709+ assert!(scheduler.in_flight.is_empty());710+ assert!(matches!(711+ scheduler.schedule(51, false),712+ MediaV2ScheduleDecision::Send {713+ discontinuity: false,714+ ..715+ }716+ ));717+ }718+719+ 720+ fn cancelled_old_completion_cannot_poison_the_replacement_gop() {721+ let mut scheduler = MediaV2Scheduler::default();722+ assert!(matches!(723+ scheduler.schedule(60, true),724+ MediaV2ScheduleDecision::Send { .. }725+ ));726+ assert!(matches!(727+ scheduler.schedule(61, false),728+ MediaV2ScheduleDecision::Send { .. }729+ ));730+ assert!(matches!(731+ scheduler.schedule(70, true),732+ MediaV2ScheduleDecision::Send {733+ discontinuity: true,734+ ..735+ }736+ ));737+738+ assert!(scheduler.fail(60).is_empty());739+ assert!(matches!(740+ scheduler.schedule(71, false),741+ MediaV2ScheduleDecision::Send {742+ discontinuity: false,743+ ..744+ }745+ ));746+ }747+748+ 749+ fn media_v2_keyframes_have_strictly_higher_transport_priority() {750+ assert!(media_v2_priority(true) > media_v2_priority(false));751+ }752+753+ 754+ async fn no_frame_peer_drop_reaps_source_and_allows_reconnect() {755+ let sessions = Arc::new(SessionRegistry::default());756+ let remote = endpoint(1);757+ let nonce = [7; 16];758+ let media = sessions759+ .claim(remote, nonce, InvitationGrants::ALL)760+ .unwrap();761+ let input = sessions.claim_input(remote, nonce).unwrap();762+ let session_id = media.session_id;763+764+ let input_sessions = Arc::clone(&sessions);765+ let input_shutdown = tokio::spawn(async move {766+ loop {767+ let notified = input_sessions.session_changed.notified();768+ if !input_sessions.is_active(remote, session_id) {769+ break;770+ }771+ notified.await;772+ }773+ drop(input);774+ });775+776+ let (_frame_sender, mut frame_receiver) =777+ tokio::sync::watch::channel(Option::<EncodedFrame>::None);778+ let (source_started_tx, source_started_rx) = tokio::sync::oneshot::channel();779+ let (source_reaped_tx, source_reaped_rx) = tokio::sync::oneshot::channel();780+ let source_task = tokio::spawn(async move {781+ let _notify = DropNotify(Some(source_reaped_tx));782+ let _ = source_started_tx.send(());783+ std::future::pending::<Result<()>>().await784+ });785+ let source_task = SourceTaskGuard::new(source_task);786+ source_started_rx.await.unwrap();787+788+ let (peer_alive, peer_disconnected) = tokio::sync::oneshot::channel::<()>();789+ let disconnected = async move {790+ let _ = peer_disconnected.await;791+ };792+ tokio::pin!(disconnected);793+ drop(peer_alive);794+795+ let activity = tokio::time::timeout(796+ Duration::from_millis(100),797+ wait_for_media_activity(&mut frame_receiver, disconnected.as_mut()),798+ )799+ .await800+ .expect("no-frame media loop ignored peer disconnect")801+ .unwrap();802+ assert_eq!(activity, MediaActivity::PeerDisconnected);803+804+ source_task.abort_and_wait().await;805+ tokio::time::timeout(Duration::from_millis(100), source_reaped_rx)806+ .await807+ .expect("source task was not reaped after peer disconnect")808+ .unwrap();809+ drop(media);810+ tokio::time::timeout(Duration::from_millis(100), input_shutdown)811+ .await812+ .expect("input lease did not observe media shutdown")813+ .unwrap();814+815+ assert!(816+ sessions817+ .claim(endpoint(2), [8; 16], InvitationGrants::ALL)818+ .is_ok()819+ );820+ }821+}