- commit
- bf395a4f0cc3ef0a4af09484faf697afcb4ee254
- parent
- 1f447d46610428431323b13a7efd769854788807
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 1084 | +----------------------------- |
| A | crates/ | 1097 | ++++++++++++++++++++++++++++++ |
@@ -1,7 +1,7 @@
11 use std::future::Future;22 use std::pin::Pin;3-use std::sync::atomic::{AtomicU64, Ordering};4-use std::sync::{Arc, Mutex};3+use std::sync::Arc;4+use std::sync::atomic::Ordering;55 use std::time::{Duration, Instant};66 77 use anyhow::{Context, Result, bail, ensure};@@ -9,8 +9,7 @@ use iroh::EndpointId;
99 use iroh::endpoint::{Connection, SendStream};1010 use iroh::protocol::ProtocolHandler;1111 use moq_net::{12- Broadcast, BroadcastConsumer, BroadcastProducer, Error as MoqError, GroupProducer, Origin,13- Track, TrackProducer,12+ Broadcast, BroadcastProducer, Error as MoqError, GroupProducer, Origin, Track, TrackProducer,1413 };1514 use sigil_protocol::{1615 AUDIO_HEADER_LEN, AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1,@@ -929,725 +928,19 @@ fn queue_ratio_at_least(depth: u8, capacity: u8, numerator: u16, denominator: u1
929928 >= u16::from(capacity).saturating_mul(numerator)930929 }931930 932-933-pub struct SessionRegistry {934- active: Mutex<Option<ActiveSession>>,935- pending_moq: Mutex<Option<PendingMoqAttachment>>,936- next_session_id: AtomicU64,937- session_changed: tokio::sync::Notify,938- pending_handshakes: tokio::sync::Semaphore,939-}940-941-struct PendingMoqAttachment {942- remote: EndpointId,943- session_id: u64,944- broadcast_name: String,945- broadcast: BroadcastConsumer,946- attached: tokio::sync::oneshot::Sender<()>,947- closed: tokio::sync::oneshot::Sender<()>,948- telemetry: Arc<MediaV3Telemetry>,949-}950-951-impl std::fmt::Debug for PendingMoqAttachment {952- fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {953- formatter954- .debug_struct("PendingMoqAttachment")955- .field("remote", &self.remote)956- .field("session_id", &self.session_id)957- .field("broadcast_name", &self.broadcast_name)958- .finish_non_exhaustive()959- }960-}961-962-struct ClaimedMoqAttachment {963- session_id: u64,964- broadcast_name: String,965- broadcast: BroadcastConsumer,966- attached: tokio::sync::oneshot::Sender<()>,967- closed: tokio::sync::oneshot::Sender<()>,968- telemetry: Arc<MediaV3Telemetry>,969-}970-971-struct MoqAttachmentWait {972- attached: tokio::sync::oneshot::Receiver<()>,973- closed: tokio::sync::oneshot::Receiver<()>,974-}975-976-977-struct ActiveSession {978- remote: EndpointId,979- session_id: u64,980- nonce: [u8; 16],981- session_clock: SessionClock,982- grants: InvitationGrants,983- media_active: bool,984- input_claimed: bool,985- audio_claimed: bool,986- feedback_claimed: bool,987- media_v3_telemetry: Arc<MediaV3Telemetry>,988- encoder_control: Option<EncoderControl>,989-}990-991-992-struct MediaV3Telemetry {993- scheduler_cancellations: AtomicU64,994- send_failures: AtomicU64,995- selected_path_rtt_micros: AtomicU64,996- selected_path_lost_packets: AtomicU64,997- selected_path_congestion_events: AtomicU64,998- keyframe_control_requests: AtomicU64,999- encoder_force_requests: AtomicU64,1000- encoder_force_acknowledgements: AtomicU64,1001- encoder_force_failures: AtomicU64,1002- last_encoder_force_ack_micros: AtomicU64,1003-}1004-1005-1006-struct MediaV3TelemetrySnapshot {1007- scheduler_cancellations: u64,1008- send_failures: u64,1009- selected_path_rtt_micros: u64,1010- selected_path_lost_packets: u64,1011- selected_path_congestion_events: u64,1012-}1013-1014-1015-struct AdaptiveEncoderProposal {1016- control: EncoderControl,1017- target_kbps: u32,1018- bitrate_revision: u64,1019- force_keyframe_revision: Option<u64>,1020-}1021-1022-1023-struct ResolutionEncoderProposal {1024- control: EncoderControl,1025- target: VideoDimensions,1026- revision: u64,1027-}1028-1029-1030-enum ForcedIdrDisposition {1031- JoinReplay,1032- Unavailable,1033- Requested { revision: u64 },1034- Coalesced { revision: u64 },1035- Failed { error: String },1036-}1037-1038-struct ForcedIdrAcknowledgement {1039- requested_revision: u64,1040- elapsed: Duration,1041- result: Result<crate::source::EncoderControlStatus>,1042-}1043-1044-struct ForcedIdrCoordinator {1045- control: Option<EncoderControl>,1046- pending_revision: Option<u64>,1047- acknowledgements: tokio::task::JoinSet<ForcedIdrAcknowledgement>,1048- telemetry: Arc<MediaV3Telemetry>,1049-}1050-1051-impl ForcedIdrCoordinator {1052- fn new(control: Option<EncoderControl>, telemetry: Arc<MediaV3Telemetry>) -> Self {1053- Self {1054- control,1055- pending_revision: None,1056- acknowledgements: tokio::task::JoinSet::new(),1057- telemetry,1058- }1059- }1060-1061- fn request(&mut self, reason: KeyframeRequestReasonV3) -> ForcedIdrDisposition {1062- self.telemetry1063- .keyframe_control_requests1064- .fetch_add(1, Ordering::Relaxed);1065- if reason == KeyframeRequestReasonV3::Join {1066- return ForcedIdrDisposition::JoinReplay;1067- }1068- if let Some(revision) = self.pending_revision {1069- return ForcedIdrDisposition::Coalesced { revision };1070- }1071- let Some(control) = self.control.clone() else {1072- return ForcedIdrDisposition::Unavailable;1073- };1074- let revision = match control.request_force_keyframe() {1075- Ok(revision) => revision,1076- Err(error) => {1077- self.telemetry1078- .encoder_force_failures1079- .fetch_add(1, Ordering::Relaxed);1080- return ForcedIdrDisposition::Failed {1081- error: error.to_string(),1082- };1083- }1084- };1085- self.pending_revision = Some(revision);1086- self.telemetry1087- .encoder_force_requests1088- .fetch_add(1, Ordering::Relaxed);1089- self.acknowledgements.spawn(async move {1090- let started_at = Instant::now();1091- let result = control1092- .wait_for_recovery_keyframe_acknowledged(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)1093- .await;1094- ForcedIdrAcknowledgement {1095- requested_revision: revision,1096- elapsed: started_at.elapsed(),1097- result,1098- }1099- });1100- ForcedIdrDisposition::Requested { revision }1101- }1102-1103- fn complete(1104- &mut self,1105- result: Option<Result<ForcedIdrAcknowledgement, tokio::task::JoinError>>,1106- remote: EndpointId,1107- transport: &'static str,1108- ) {1109- let pending_revision = self.pending_revision.take();1110- match result {1111- Some(Ok(acknowledgement)) => match acknowledgement.result {1112- Ok(status) => {1113- let elapsed_micros =1114- u64::try_from(acknowledgement.elapsed.as_micros()).unwrap_or(u64::MAX);1115- self.telemetry1116- .encoder_force_acknowledgements1117- .fetch_add(1, Ordering::Relaxed);1118- self.telemetry1119- .last_encoder_force_ack_micros1120- .store(elapsed_micros, Ordering::Relaxed);1121- debug!(1122- %remote,1123- transport,1124- requested_revision = acknowledgement.requested_revision,1125- acknowledged_revision = ?status.acknowledged_force_keyframe_revision,1126- elapsed_micros,1127- "forced-IDR recovery acknowledged"1128- );1129- }1130- Err(error) => {1131- self.telemetry1132- .encoder_force_failures1133- .fetch_add(1, Ordering::Relaxed);1134- warn!(1135- %remote,1136- transport,1137- requested_revision = acknowledgement.requested_revision,1138- %error,1139- "forced-IDR recovery was not acknowledged; retaining natural-IDR fallback"1140- );1141- }1142- },1143- Some(Err(error)) => {1144- self.telemetry1145- .encoder_force_failures1146- .fetch_add(1, Ordering::Relaxed);1147- warn!(1148- %remote,1149- transport,1150- ?pending_revision,1151- %error,1152- "forced-IDR acknowledgement task failed; retaining natural-IDR fallback"1153- );1154- }1155- None => {1156- self.telemetry1157- .encoder_force_failures1158- .fetch_add(1, Ordering::Relaxed);1159- warn!(1160- %remote,1161- transport,1162- ?pending_revision,1163- "forced-IDR acknowledgement task ended without a result"1164- );1165- }1166- }1167- }1168-1169- async fn abort_and_drain(&mut self, remote: EndpointId, transport: &'static str) {1170- self.pending_revision = None;1171- self.acknowledgements.abort_all();1172- while self.acknowledgements.join_next().await.is_some() {}1173- debug!(1174- %remote,1175- transport,1176- keyframe_control_requests = self1177- .telemetry1178- .keyframe_control_requests1179- .load(Ordering::Relaxed),1180- encoder_force_requests = self1181- .telemetry1182- .encoder_force_requests1183- .load(Ordering::Relaxed),1184- encoder_force_acknowledgements = self1185- .telemetry1186- .encoder_force_acknowledgements1187- .load(Ordering::Relaxed),1188- encoder_force_failures = self1189- .telemetry1190- .encoder_force_failures1191- .load(Ordering::Relaxed),1192- last_encoder_force_ack_micros = self1193- .telemetry1194- .last_encoder_force_ack_micros1195- .load(Ordering::Relaxed),1196- "forced-IDR recovery session summary"1197- );1198- }1199-}1200-1201-impl MediaV3Telemetry {1202- fn snapshot(&self) -> MediaV3TelemetrySnapshot {1203- MediaV3TelemetrySnapshot {1204- scheduler_cancellations: self.scheduler_cancellations.load(Ordering::Relaxed),1205- send_failures: self.send_failures.load(Ordering::Relaxed),1206- selected_path_rtt_micros: self.selected_path_rtt_micros.load(Ordering::Relaxed),1207- selected_path_lost_packets: self.selected_path_lost_packets.load(Ordering::Relaxed),1208- selected_path_congestion_events: self1209- .selected_path_congestion_events1210- .load(Ordering::Relaxed),1211- }1212- }1213-1214- fn record_selected_path(&self, connection: &Connection) {1215- let paths = connection.paths();1216- let Some(path) = paths.iter().find(|path| path.is_selected()) else {1217- return;1218- };1219- let stats = path.stats();1220- self.selected_path_rtt_micros.store(1221- u64::try_from(stats.rtt.as_micros()).unwrap_or(u64::MAX),1222- Ordering::Relaxed,1223- );1224- self.selected_path_lost_packets1225- .store(stats.lost_packets, Ordering::Relaxed);1226- self.selected_path_congestion_events1227- .store(stats.congestion_events, Ordering::Relaxed);1228- }1229-}1230-1231-impl Default for SessionRegistry {1232- fn default() -> Self {1233- Self {1234- active: Mutex::new(None),1235- pending_moq: Mutex::new(None),1236- next_session_id: AtomicU64::new(0),1237- session_changed: tokio::sync::Notify::new(),1238- pending_handshakes: tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES),1239- }1240- }1241-}1242-1243-impl SessionRegistry {1244- pub fn has_session(&self) -> bool {1245- self.active1246- .lock()1247- .expect("session registry poisoned")1248- .is_some()1249- }1250-1251- fn claim(1252- self: &Arc<Self>,1253- remote: EndpointId,1254- nonce: [u8; 16],1255- grants: InvitationGrants,1256- ) -> Result<SessionLease> {1257- let mut active = self.active.lock().expect("session registry poisoned");1258- if let Some(current) = active.as_ref() {1259- bail!("host already has active client {}", current.remote);1260- }1261- let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;1262- let session_clock = SessionClock::start();1263- let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());1264- *active = Some(ActiveSession {1265- remote,1266- session_id,1267- nonce,1268- session_clock,1269- grants,1270- media_active: true,1271- input_claimed: false,1272- audio_claimed: false,1273- feedback_claimed: false,1274- media_v3_telemetry: Arc::clone(&media_v3_telemetry),1275- encoder_control: None,1276- });1277- Ok(SessionLease {1278- registry: Arc::clone(self),1279- remote,1280- session_id,1281- session_clock,1282- media_v3_telemetry,1283- })1284- }1285-1286- fn claim_input(self: &Arc<Self>, remote: EndpointId, nonce: [u8; 16]) -> Result<InputLease> {1287- let mut active = self.active.lock().expect("session registry poisoned");1288- let session = active1289- .as_mut()1290- .filter(|session| {1291- session.media_active && session.remote == remote && session.nonce == nonce1292- })1293- .context("input connection does not match the active media session")?;1294- ensure!(1295- !session.input_claimed,1296- "active client already has an input stream"1297- );1298- session.input_claimed = true;1299- Ok(InputLease {1300- registry: Arc::clone(self),1301- remote,1302- session_id: session.session_id,1303- grants: session.grants,1304- })1305- }1306-1307- fn install_encoder_control(1308- &self,1309- remote: EndpointId,1310- session_id: u64,1311- encoder_control: Option<EncoderControl>,1312- ) -> Result<()> {1313- let mut active = self.active.lock().expect("session registry poisoned");1314- let session = active1315- .as_mut()1316- .filter(|session| {1317- session.media_active && session.remote == remote && session.session_id == session_id1318- })1319- .context("encoder control does not match the active media session")?;1320- ensure!(1321- session.encoder_control.is_none(),1322- "active media session already has encoder control"1323- );1324- session.encoder_control = encoder_control;1325- Ok(())1326- }1327-1328- fn propose_adaptive_encoder_update(1329- &self,1330- remote: EndpointId,1331- session_id: u64,1332- target_kbps: u32,1333- force_keyframe: bool,1334- ) -> Result<Option<AdaptiveEncoderProposal>> {1335- let active = self.active.lock().expect("session registry poisoned");1336- let session = active1337- .as_ref()1338- .filter(|session| {1339- session.media_active && session.remote == remote && session.session_id == session_id1340- })1341- .context("adaptive encoder update does not match the active media session")?;1342- let Some(control) = session.encoder_control.clone() else {1343- return Ok(None);1344- };1345- let bitrate_revision = control.request_bitrate_kbps(target_kbps)?;1346- let force_keyframe_revision = force_keyframe1347- .then(|| control.request_force_keyframe())1348- .transpose()?;1349- Ok(Some(AdaptiveEncoderProposal {1350- control,1351- target_kbps,1352- bitrate_revision,1353- force_keyframe_revision,1354- }))1355- }1356-1357- fn propose_resolution_update(1358- &self,1359- remote: EndpointId,1360- session_id: u64,1361- target: VideoDimensions,1362- ) -> Result<Option<ResolutionEncoderProposal>> {1363- let active = self.active.lock().expect("session registry poisoned");1364- let session = active1365- .as_ref()1366- .filter(|session| {1367- session.media_active && session.remote == remote && session.session_id == session_id1368- })1369- .context("resolution update does not match the active media session")?;1370- let Some(control) = session.encoder_control.clone() else {1371- return Ok(None);1372- };1373- let revision = control.request_resolution(target.width, target.height)?;1374- Ok(Some(ResolutionEncoderProposal {1375- control,1376- target,1377- revision,1378- }))1379- }1380-1381- fn claim_feedback(1382- self: &Arc<Self>,1383- remote: EndpointId,1384- nonce: [u8; 16],1385- ) -> Result<FeedbackLease> {1386- let mut active = self.active.lock().expect("session registry poisoned");1387- let session = active1388- .as_mut()1389- .filter(|session| {1390- session.media_active && session.remote == remote && session.nonce == nonce1391- })1392- .context("feedback connection does not match the active media session")?;1393- ensure!(1394- session.grants.contains(InvitationGrants::VIEW),1395- "active Portal session lacks feedback view permission"1396- );1397- ensure!(1398- !session.feedback_claimed,1399- "active client already has a feedback connection"1400- );1401- session.feedback_claimed = true;1402- Ok(FeedbackLease {1403- registry: Arc::clone(self),1404- remote,1405- session_id: session.session_id,1406- telemetry: Arc::clone(&session.media_v3_telemetry),1407- encoder_control: session.encoder_control.clone(),1408- })1409- }1410-1411- fn claim_audio(self: &Arc<Self>, remote: EndpointId, nonce: [u8; 16]) -> Result<AudioLease> {1412- let mut active = self.active.lock().expect("session registry poisoned");1413- let session = active1414- .as_mut()1415- .filter(|session| {1416- session.media_active && session.remote == remote && session.nonce == nonce1417- })1418- .context("audio connection does not match the active media session")?;1419- ensure!(1420- !session.audio_claimed,1421- "active client already has an audio connection"1422- );1423- session.audio_claimed = true;1424- Ok(AudioLease {1425- registry: Arc::clone(self),1426- remote,1427- session_id: session.session_id,1428- session_clock: session.session_clock,1429- grants: session.grants,1430- })1431- }1432-1433- fn expect_moq(1434- &self,1435- remote: EndpointId,1436- session_id: u64,1437- broadcast_name: String,1438- broadcast: BroadcastConsumer,1439- ) -> Result<MoqAttachmentWait> {1440- let active = self.active.lock().expect("session registry poisoned");1441- let telemetry = active1442- .as_ref()1443- .filter(|session| {1444- session.media_active && session.remote == remote && session.session_id == session_id1445- })1446- .map(|session| Arc::clone(&session.media_v3_telemetry))1447- .context("MoQ expectation does not match the active control session")?;1448- let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1449- ensure!(1450- pending.is_none(),1451- "active control session already expects MoQ"1452- );1453- let (attached, attached_rx) = tokio::sync::oneshot::channel();1454- let (closed, closed_rx) = tokio::sync::oneshot::channel();1455- *pending = Some(PendingMoqAttachment {1456- remote,1457- session_id,1458- broadcast_name,1459- broadcast,1460- attached,1461- closed,1462- telemetry,1463- });1464- Ok(MoqAttachmentWait {1465- attached: attached_rx,1466- closed: closed_rx,1467- })1468- }1469-1470- fn claim_moq(&self, remote: EndpointId) -> Result<ClaimedMoqAttachment> {1471- let active = self.active.lock().expect("session registry poisoned");1472- let session = active1473- .as_ref()1474- .filter(|session| session.media_active && session.remote == remote)1475- .context("MoQ connection does not match the active control session")?;1476- let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1477- let attachment = pending1478- .as_ref()1479- .filter(|attachment| {1480- attachment.remote == remote && attachment.session_id == session.session_id1481- })1482- .context("active control session is not expecting a MoQ connection")?;1483- debug_assert_eq!(attachment.remote, session.remote);1484- let attachment = pending1485- .take()1486- .expect("validated pending MoQ attachment disappeared");1487- Ok(ClaimedMoqAttachment {1488- session_id: attachment.session_id,1489- broadcast_name: attachment.broadcast_name,1490- broadcast: attachment.broadcast,1491- attached: attachment.attached,1492- closed: attachment.closed,1493- telemetry: attachment.telemetry,1494- })1495- }1496-1497- fn release(&self, remote: EndpointId, session_id: u64) {1498- let mut active = self.active.lock().expect("session registry poisoned");1499- if let Some(session) = active.as_mut()1500- && session.remote == remote1501- && session.session_id == session_id1502- {1503- let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1504- if pending.as_ref().is_some_and(|attachment| {1505- attachment.remote == remote && attachment.session_id == session_id1506- }) {1507- *pending = None;1508- }1509- // Keep the registry occupied until the input handler has observed1510- // media shutdown and released all held uinput transitions. This1511- // prevents a reconnect from sharing the device with a draining1512- // predecessor session.1513- session.media_active = false;1514- session.encoder_control = None;1515- if !session.input_claimed && !session.audio_claimed && !session.feedback_claimed {1516- *active = None;1517- }1518- drop(active);1519- self.session_changed.notify_waiters();1520- }1521- }1522-1523- fn is_active(&self, remote: EndpointId, session_id: u64) -> bool {1524- self.active1525- .lock()1526- .expect("session registry poisoned")1527- .as_ref()1528- .is_some_and(|active| {1529- active.media_active && active.remote == remote && active.session_id == session_id1530- })1531- }1532-1533- fn release_input(&self, remote: EndpointId, session_id: u64) {1534- let mut active = self.active.lock().expect("session registry poisoned");1535- if let Some(session) = active.as_mut()1536- && session.remote == remote1537- && session.session_id == session_id1538- {1539- session.input_claimed = false;1540- if !session.media_active && !session.audio_claimed && !session.feedback_claimed {1541- *active = None;1542- }1543- }1544- }1545-1546- fn release_audio(&self, remote: EndpointId, session_id: u64) {1547- let mut active = self.active.lock().expect("session registry poisoned");1548- if let Some(session) = active.as_mut()1549- && session.remote == remote1550- && session.session_id == session_id1551- {1552- session.audio_claimed = false;1553- if !session.media_active && !session.input_claimed && !session.feedback_claimed {1554- *active = None;1555- }1556- }1557- }1558-1559- fn release_feedback(&self, remote: EndpointId, session_id: u64) {1560- let mut active = self.active.lock().expect("session registry poisoned");1561- if let Some(session) = active.as_mut()1562- && session.remote == remote1563- && session.session_id == session_id1564- {1565- session.feedback_claimed = false;1566- if !session.media_active && !session.input_claimed && !session.audio_claimed {1567- *active = None;1568- }1569- }1570- }1571-}1572-1573-1574-struct SessionLease {1575- registry: Arc<SessionRegistry>,1576- remote: EndpointId,1577- session_id: u64,1578- session_clock: SessionClock,1579- media_v3_telemetry: Arc<MediaV3Telemetry>,1580-}1581-1582-impl Drop for SessionLease {1583- fn drop(&mut self) {1584- self.registry.release(self.remote, self.session_id);1585- }1586-}1587-1588-1589-struct InputLease {1590- registry: Arc<SessionRegistry>,1591- remote: EndpointId,1592- session_id: u64,1593- grants: InvitationGrants,1594-}1595-1596-1597-struct AudioLease {1598- registry: Arc<SessionRegistry>,1599- remote: EndpointId,1600- session_id: u64,1601- session_clock: SessionClock,1602- grants: InvitationGrants,1603-}1604-1605-1606-struct FeedbackLease {1607- registry: Arc<SessionRegistry>,1608- remote: EndpointId,1609- session_id: u64,1610- telemetry: Arc<MediaV3Telemetry>,1611- encoder_control: Option<EncoderControl>,1612-}1613-1614-1615-struct SourceTaskGuard(Option<tokio::task::JoinHandle<Result<()>>>);1616-1617-impl SourceTaskGuard {1618- fn new(task: tokio::task::JoinHandle<Result<()>>) -> Self {1619- Self(Some(task))1620- }1621-1622- async fn abort_and_wait(mut self) {1623- if let Some(task) = self.0.take() {1624- task.abort();1625- let _ = task.await;1626- }1627- }1628-1629- async fn wait_or_abort(mut self, grace_timeout: Duration) {1630- let Some(mut task) = self.0.take() else {1631- return;1632- };1633- if tokio::time::timeout(grace_timeout, &mut task)1634- .await1635- .is_err()1636- {1637- task.abort();1638- let _ = task.await;1639- }1640- }931+932+fn endpoint(byte: u8) -> EndpointId {933+ iroh::SecretKey::from_bytes(&[byte; 32]).public()1641934 }1642935 1643-impl Drop for SourceTaskGuard {1644- fn drop(&mut self) {1645- if let Some(task) = self.0.take() {1646- task.abort();1647- }1648- }1649-}936+mod session;1650937 938+pub use session::SessionRegistry;939+use session::{940+ AdaptiveEncoderProposal, ClaimedMoqAttachment, ForcedIdrCoordinator, ForcedIdrDisposition,941+ MediaV3Telemetry, MediaV3TelemetrySnapshot, MoqAttachmentWait, ResolutionEncoderProposal,942+ SourceTaskGuard,943+};1651944 1652945 struct StartupCadenceSample {1653946 first_sequence: Option<u64>,@@ -2560,24 +1853,6 @@ where
25601853 }25611854 }25621855 2563-impl Drop for InputLease {2564- fn drop(&mut self) {2565- self.registry.release_input(self.remote, self.session_id);2566- }2567-}2568-2569-impl Drop for AudioLease {2570- fn drop(&mut self) {2571- self.registry.release_audio(self.remote, self.session_id);2572- }2573-}2574-2575-impl Drop for FeedbackLease {2576- fn drop(&mut self) {2577- self.registry.release_feedback(self.remote, self.session_id);2578- }2579-}2580-25811856 25821857 pub struct MediaHandler {25831858 pub config: HostConfig,@@ -6170,66 +5445,6 @@ mod tests {
61705445 );61715446 }61725447 6173- 6174- async fn forced_idr_recovery_is_one_slot_join_safe_and_rearms_after_ack() {6175- let telemetry = Arc::new(MediaV3Telemetry::default());6176- let harness = crate::source::EncoderControlTestHarness::new();6177- let mut coordinator =6178- ForcedIdrCoordinator::new(Some(harness.control.clone()), Arc::clone(&telemetry));6179-6180- assert_eq!(6181- coordinator.request(KeyframeRequestReasonV3::Join),6182- ForcedIdrDisposition::JoinReplay6183- );6184- assert_eq!(harness.requested_force_keyframe_revision(), None);6185-6186- let requested_revision = match coordinator.request(KeyframeRequestReasonV3::DecoderReset) {6187- ForcedIdrDisposition::Requested { revision } => revision,6188- disposition => panic!("unexpected forced-IDR disposition: {disposition:?}"),6189- };6190- assert_eq!(6191- harness.requested_force_keyframe_revision(),6192- Some(requested_revision)6193- );6194- assert_eq!(6195- coordinator.request(KeyframeRequestReasonV3::TransportGap),6196- ForcedIdrDisposition::Coalesced {6197- revision: requested_revision6198- }6199- );6200- assert_eq!(telemetry.encoder_force_requests.load(Ordering::Relaxed), 1);6201-6202- let newer_revision = harness.control.request_force_keyframe().unwrap();6203- harness.status.send_modify(|status| {6204- status.requested_force_keyframe_revision = Some(newer_revision);6205- status.acknowledged_force_keyframe_revision = Some(newer_revision);6206- });6207- let acknowledgement = coordinator.acknowledgements.join_next().await;6208- coordinator.complete(acknowledgement, endpoint(1), "test");6209- assert_eq!(coordinator.pending_revision, None);6210- assert_eq!(6211- telemetry6212- .encoder_force_acknowledgements6213- .load(Ordering::Relaxed),6214- 16215- );6216- assert_eq!(telemetry.encoder_force_failures.load(Ordering::Relaxed), 0);6217-6218- assert!(matches!(6219- coordinator.request(KeyframeRequestReasonV3::DecoderReset),6220- ForcedIdrDisposition::Requested { revision } if revision > newer_revision6221- ));6222- coordinator.abort_and_drain(endpoint(1), "test").await;6223-6224- let fallback_telemetry = Arc::new(MediaV3Telemetry::default());6225- let mut fallback = ForcedIdrCoordinator::new(None, fallback_telemetry);6226- assert_eq!(6227- fallback.request(KeyframeRequestReasonV3::DecoderReset),6228- ForcedIdrDisposition::Unavailable6229- );6230- assert!(fallback.acknowledgements.is_empty());6231- }6232-62335448 62345449 async fn upstream_moq_group_counts_envelope_bytes_before_upstream_cache_eviction() {62355450 let track_info = Track {@@ -6632,10 +5847,6 @@ mod tests {
66325847 assert!(media_v3_transport_priority(1).is_err());66335848 }66345849 6635- fn endpoint(byte: u8) -> EndpointId {6636- iroh::SecretKey::from_bytes(&[byte; 32]).public()6637- }6638-66395850 fn startup_sample(fps: f64, frames: u64, receiver_open: bool) -> StartupCadenceSample {66405851 if frames == 0 {66415852 return StartupCadenceSample {@@ -6764,217 +5975,6 @@ mod tests {
67645975 .unwrap();67655976 }67665977 6767- 6768- fn only_one_remote_can_hold_session() {6769- let sessions = Arc::new(SessionRegistry::default());6770- assert!(!sessions.has_session());6771- let nonce = [7; 16];6772- let first = sessions6773- .claim(endpoint(1), nonce, InvitationGrants::ALL)6774- .unwrap();6775- assert!(sessions.has_session());6776- assert!(6777- sessions6778- .claim(endpoint(2), nonce, InvitationGrants::ALL)6779- .is_err()6780- );6781- assert!(sessions.claim_input(endpoint(1), [8; 16]).is_err());6782- let input = sessions.claim_input(endpoint(1), nonce).unwrap();6783- assert_eq!(input.session_id, first.session_id);6784- assert!(sessions.claim_input(endpoint(1), nonce).is_err());6785- let audio = sessions.claim_audio(endpoint(1), nonce).unwrap();6786- assert!(sessions.claim_audio(endpoint(1), nonce).is_err());6787- drop(input);6788- let draining_input = sessions.claim_input(endpoint(1), nonce).unwrap();6789- drop(first);6790- assert!(6791- sessions6792- .claim(endpoint(2), nonce, InvitationGrants::ALL)6793- .is_err()6794- );6795- drop(draining_input);6796- assert!(6797- sessions6798- .claim(endpoint(2), nonce, InvitationGrants::ALL)6799- .is_err()6800- );6801- drop(audio);6802- assert!(!sessions.has_session());6803- assert!(6804- sessions6805- .claim(endpoint(2), nonce, InvitationGrants::ALL)6806- .is_ok()6807- );6808- }6809-6810- fn test_moq_broadcast() -> (BroadcastProducer, BroadcastConsumer) {6811- let producer = Broadcast::new().produce();6812- let consumer = producer.consume();6813- (producer, consumer)6814- }6815-6816- 6817- fn moq_attachment_requires_exact_active_control_remote_and_is_single_use() {6818- let sessions = Arc::new(SessionRegistry::default());6819- assert!(sessions.claim_moq(endpoint(1)).is_err());6820- let lease = sessions6821- .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)6822- .unwrap();6823- let (_producer, consumer) = test_moq_broadcast();6824- let _wait = sessions6825- .expect_moq(6826- endpoint(1),6827- lease.session_id,6828- media_moq_broadcast_name(lease.session_id).unwrap(),6829- consumer,6830- )6831- .unwrap();6832-6833- // A wrong peer cannot consume the exact pending attachment.6834- assert!(sessions.claim_moq(endpoint(2)).is_err());6835- let attachment = sessions.claim_moq(endpoint(1)).unwrap();6836- assert_eq!(attachment.session_id, lease.session_id);6837- assert_eq!(6838- attachment.broadcast_name,6839- media_moq_broadcast_name(lease.session_id).unwrap()6840- );6841- // The pending token was atomically consumed before the MoQ handshake.6842- assert!(sessions.claim_moq(endpoint(1)).is_err());6843- }6844-6845- 6846- fn competing_moq_connections_cannot_both_claim_one_control_attachment() {6847- let sessions = Arc::new(SessionRegistry::default());6848- let lease = sessions6849- .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)6850- .unwrap();6851- let (_producer, consumer) = test_moq_broadcast();6852- let _wait = sessions6853- .expect_moq(6854- endpoint(1),6855- lease.session_id,6856- media_moq_broadcast_name(lease.session_id).unwrap(),6857- consumer,6858- )6859- .unwrap();6860- let barrier = Arc::new(std::sync::Barrier::new(3));6861- let contenders = (0..2)6862- .map(|_| {6863- let sessions = Arc::clone(&sessions);6864- let barrier = Arc::clone(&barrier);6865- std::thread::spawn(move || {6866- barrier.wait();6867- sessions.claim_moq(endpoint(1)).is_ok()6868- })6869- })6870- .collect::<Vec<_>>();6871- barrier.wait();6872- let claimed = contenders6873- .into_iter()6874- .map(|thread| usize::from(thread.join().unwrap()))6875- .sum::<usize>();6876- assert_eq!(claimed, 1);6877- }6878-6879- 6880- async fn releasing_control_clears_an_unclaimed_moq_attachment() {6881- let sessions = Arc::new(SessionRegistry::default());6882- let lease = sessions6883- .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)6884- .unwrap();6885- let (_producer, consumer) = test_moq_broadcast();6886- let wait = sessions6887- .expect_moq(6888- endpoint(1),6889- lease.session_id,6890- media_moq_broadcast_name(lease.session_id).unwrap(),6891- consumer,6892- )6893- .unwrap();6894- drop(lease);6895- assert!(sessions.claim_moq(endpoint(1)).is_err());6896- assert!(wait.attached.await.is_err());6897- assert!(wait.closed.await.is_err());6898- }6899-6900- 6901- fn feedback_attaches_only_to_exact_active_view_session() {6902- let sessions = Arc::new(SessionRegistry::default());6903- let nonce = [9; 16];6904- assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());6905-6906- let no_view = sessions6907- .claim(endpoint(1), nonce, InvitationGrants::GAMEPAD)6908- .unwrap();6909- assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());6910- drop(no_view);6911-6912- let media = sessions6913- .claim(endpoint(1), nonce, InvitationGrants::VIEW)6914- .unwrap();6915- assert!(sessions.claim_feedback(endpoint(2), nonce).is_err());6916- assert!(sessions.claim_feedback(endpoint(1), [8; 16]).is_err());6917- let feedback = sessions.claim_feedback(endpoint(1), nonce).unwrap();6918- assert_eq!(feedback.session_id, media.session_id);6919- assert!(feedback.encoder_control.is_none());6920- assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());6921-6922- drop(media);6923- assert!(6924- sessions6925- .claim(endpoint(2), nonce, InvitationGrants::VIEW)6926- .is_err(),6927- "feedback teardown must keep the draining session isolated"6928- );6929- drop(feedback);6930- assert!(6931- sessions6932- .claim(endpoint(2), nonce, InvitationGrants::VIEW)6933- .is_ok()6934- );6935- }6936-6937- 6938- fn adaptive_encoder_proposals_are_bound_to_the_exact_active_generation() {6939- let sessions = Arc::new(SessionRegistry::default());6940- let remote = endpoint(1);6941- let media = sessions6942- .claim(remote, [3; 16], InvitationGrants::VIEW)6943- .unwrap();6944- let harness = crate::source::EncoderControlTestHarness::new();6945-6946- assert!(6947- sessions6948- .install_encoder_control(6949- endpoint(2),6950- media.session_id,6951- Some(harness.control.clone())6952- )6953- .is_err()6954- );6955- sessions6956- .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))6957- .unwrap();6958- let feedback = sessions.claim_feedback(remote, [3; 16]).unwrap();6959- assert!(feedback.encoder_control.is_some());6960- let proposal = sessions6961- .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)6962- .unwrap()6963- .unwrap();6964- assert_eq!(proposal.target_kbps, 8_000);6965- assert!(proposal.force_keyframe_revision > Some(proposal.bitrate_revision));6966-6967- let old_session_id = media.session_id;6968- drop(media);6969- assert!(6970- sessions6971- .propose_adaptive_encoder_update(remote, old_session_id, 7_000, false)6972- .is_err(),6973- "a draining generation must not issue another encoder proposal"6974- );6975- drop(feedback);6976- }6977-69785978 69795979 async fn adaptive_encoder_commit_requires_exact_readback_and_tracks_recovery_separately() {69805980 let sessions = Arc::new(SessionRegistry::default());@@ -7184,17 +6184,6 @@ mod tests {
71846184 );71856185 }71866186 7187- 7188- fn pending_handshakes_are_bounded() {7189- let sessions = SessionRegistry::default();7190- let permits: Vec<_> = (0..MAX_PENDING_HANDSHAKES)7191- .map(|_| sessions.pending_handshakes.try_acquire().unwrap())7192- .collect();7193- assert!(sessions.pending_handshakes.try_acquire().is_err());7194- drop(permits);7195- assert!(sessions.pending_handshakes.try_acquire().is_ok());7196- }7197-71986187 71996188 fn capability_negotiation_is_an_exact_intersection() {72006189 let hello = ClientHello::new(@@ -7283,20 +6272,6 @@ mod tests {
72836272 );72846273 }72856274 7286- 7287- fn session_substreams_inherit_the_exact_enrollment_grant() {7288- let sessions = Arc::new(SessionRegistry::default());7289- let grants = InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD);7290- let media = sessions.claim(endpoint(1), [3; 16], grants).unwrap();7291- let input = sessions.claim_input(endpoint(1), [3; 16]).unwrap();7292- let audio = sessions.claim_audio(endpoint(1), [3; 16]).unwrap();7293- assert_eq!(input.grants, grants);7294- assert_eq!(audio.grants, grants);7295- drop(input);7296- drop(audio);7297- drop(media);7298- }7299-73006275 73016276 fn pointer_feedback_is_advertised_only_with_tracker_and_relative_input() {73026277 assert_eq!(@@ -7417,39 +6392,6 @@ mod tests {
74176392 );74186393 }74196394 7420- 7421- fn audio_claim_requires_the_active_remote_and_nonce() {7422- let sessions = Arc::new(SessionRegistry::default());7423- let media = sessions7424- .claim(endpoint(1), [9; 16], InvitationGrants::ALL)7425- .unwrap();7426- assert!(sessions.claim_audio(endpoint(2), [9; 16]).is_err());7427- assert!(sessions.claim_audio(endpoint(1), [8; 16]).is_err());7428- let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();7429- drop(media);7430- assert!(7431- sessions7432- .claim(endpoint(2), [0; 16], InvitationGrants::ALL)7433- .is_err()7434- );7435- drop(audio);7436- assert!(7437- sessions7438- .claim(endpoint(2), [0; 16], InvitationGrants::ALL)7439- .is_ok()7440- );7441- }7442-7443- 7444- fn media_and_audio_leases_share_one_session_clock() {7445- let sessions = Arc::new(SessionRegistry::default());7446- let media = sessions7447- .claim(endpoint(1), [9; 16], InvitationGrants::ALL)7448- .unwrap();7449- let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();7450- assert_eq!(media.session_clock, audio.session_clock);7451- }7452-74536395 74546396 fn current_gop_replay_is_complete_and_skips_already_sent_frames() {74556397 let frame = |sequence, keyframe| EncodedFrame {@@ -0,0 +1,1097 @@
1+use std::sync::atomic::{AtomicU64, Ordering};2+use std::sync::{Arc, Mutex};3+use std::time::{Duration, Instant};4+5+use anyhow::{Context, Result, bail, ensure};6+use iroh::EndpointId;7+use iroh::endpoint::Connection;8+use moq_net::BroadcastConsumer;9+use sigil_protocol::{InvitationGrants, KeyframeRequestReasonV3};10+use tracing::{debug, warn};11+12+use super::{ENCODER_CONTROL_COMMIT_TIMEOUT, MAX_PENDING_HANDSHAKES, VideoDimensions};13+use crate::clock::SessionClock;14+use crate::source::EncoderControl;15+16+17+pub struct SessionRegistry {18+ active: Mutex<Option<ActiveSession>>,19+ pending_moq: Mutex<Option<PendingMoqAttachment>>,20+ next_session_id: AtomicU64,21+ pub(super) session_changed: tokio::sync::Notify,22+ pub(super) pending_handshakes: tokio::sync::Semaphore,23+}24+25+struct PendingMoqAttachment {26+ remote: EndpointId,27+ session_id: u64,28+ broadcast_name: String,29+ broadcast: BroadcastConsumer,30+ attached: tokio::sync::oneshot::Sender<()>,31+ closed: tokio::sync::oneshot::Sender<()>,32+ telemetry: Arc<MediaV3Telemetry>,33+}34+35+impl std::fmt::Debug for PendingMoqAttachment {36+ fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {37+ formatter38+ .debug_struct("PendingMoqAttachment")39+ .field("remote", &self.remote)40+ .field("session_id", &self.session_id)41+ .field("broadcast_name", &self.broadcast_name)42+ .finish_non_exhaustive()43+ }44+}45+46+pub(super) struct ClaimedMoqAttachment {47+ pub(super) session_id: u64,48+ pub(super) broadcast_name: String,49+ pub(super) broadcast: BroadcastConsumer,50+ pub(super) attached: tokio::sync::oneshot::Sender<()>,51+ pub(super) closed: tokio::sync::oneshot::Sender<()>,52+ pub(super) telemetry: Arc<MediaV3Telemetry>,53+}54+55+pub(super) struct MoqAttachmentWait {56+ pub(super) attached: tokio::sync::oneshot::Receiver<()>,57+ pub(super) closed: tokio::sync::oneshot::Receiver<()>,58+}59+60+61+struct ActiveSession {62+ remote: EndpointId,63+ session_id: u64,64+ nonce: [u8; 16],65+ session_clock: SessionClock,66+ grants: InvitationGrants,67+ media_active: bool,68+ input_claimed: bool,69+ audio_claimed: bool,70+ feedback_claimed: bool,71+ media_v3_telemetry: Arc<MediaV3Telemetry>,72+ encoder_control: Option<EncoderControl>,73+}74+75+76+pub(super) struct MediaV3Telemetry {77+ pub(super) scheduler_cancellations: AtomicU64,78+ pub(super) send_failures: AtomicU64,79+ selected_path_rtt_micros: AtomicU64,80+ selected_path_lost_packets: AtomicU64,81+ selected_path_congestion_events: AtomicU64,82+ keyframe_control_requests: AtomicU64,83+ encoder_force_requests: AtomicU64,84+ encoder_force_acknowledgements: AtomicU64,85+ encoder_force_failures: AtomicU64,86+ last_encoder_force_ack_micros: AtomicU64,87+}88+89+90+pub(super) struct MediaV3TelemetrySnapshot {91+ pub(super) scheduler_cancellations: u64,92+ pub(super) send_failures: u64,93+ pub(super) selected_path_rtt_micros: u64,94+ pub(super) selected_path_lost_packets: u64,95+ pub(super) selected_path_congestion_events: u64,96+}97+98+99+pub(super) struct AdaptiveEncoderProposal {100+ pub(super) control: EncoderControl,101+ pub(super) target_kbps: u32,102+ pub(super) bitrate_revision: u64,103+ pub(super) force_keyframe_revision: Option<u64>,104+}105+106+107+pub(super) struct ResolutionEncoderProposal {108+ pub(super) control: EncoderControl,109+ pub(super) target: VideoDimensions,110+ pub(super) revision: u64,111+}112+113+114+pub(super) enum ForcedIdrDisposition {115+ JoinReplay,116+ Unavailable,117+ Requested { revision: u64 },118+ Coalesced { revision: u64 },119+ Failed { error: String },120+}121+122+pub(super) struct ForcedIdrAcknowledgement {123+ requested_revision: u64,124+ elapsed: Duration,125+ result: Result<crate::source::EncoderControlStatus>,126+}127+128+pub(super) struct ForcedIdrCoordinator {129+ control: Option<EncoderControl>,130+ pub(super) pending_revision: Option<u64>,131+ pub(super) acknowledgements: tokio::task::JoinSet<ForcedIdrAcknowledgement>,132+ telemetry: Arc<MediaV3Telemetry>,133+}134+135+impl ForcedIdrCoordinator {136+ pub(super) fn new(control: Option<EncoderControl>, telemetry: Arc<MediaV3Telemetry>) -> Self {137+ Self {138+ control,139+ pending_revision: None,140+ acknowledgements: tokio::task::JoinSet::new(),141+ telemetry,142+ }143+ }144+145+ pub(super) fn request(&mut self, reason: KeyframeRequestReasonV3) -> ForcedIdrDisposition {146+ self.telemetry147+ .keyframe_control_requests148+ .fetch_add(1, Ordering::Relaxed);149+ if reason == KeyframeRequestReasonV3::Join {150+ return ForcedIdrDisposition::JoinReplay;151+ }152+ if let Some(revision) = self.pending_revision {153+ return ForcedIdrDisposition::Coalesced { revision };154+ }155+ let Some(control) = self.control.clone() else {156+ return ForcedIdrDisposition::Unavailable;157+ };158+ let revision = match control.request_force_keyframe() {159+ Ok(revision) => revision,160+ Err(error) => {161+ self.telemetry162+ .encoder_force_failures163+ .fetch_add(1, Ordering::Relaxed);164+ return ForcedIdrDisposition::Failed {165+ error: error.to_string(),166+ };167+ }168+ };169+ self.pending_revision = Some(revision);170+ self.telemetry171+ .encoder_force_requests172+ .fetch_add(1, Ordering::Relaxed);173+ self.acknowledgements.spawn(async move {174+ let started_at = Instant::now();175+ let result = control176+ .wait_for_recovery_keyframe_acknowledged(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)177+ .await;178+ ForcedIdrAcknowledgement {179+ requested_revision: revision,180+ elapsed: started_at.elapsed(),181+ result,182+ }183+ });184+ ForcedIdrDisposition::Requested { revision }185+ }186+187+ pub(super) fn complete(188+ &mut self,189+ result: Option<Result<ForcedIdrAcknowledgement, tokio::task::JoinError>>,190+ remote: EndpointId,191+ transport: &'static str,192+ ) {193+ let pending_revision = self.pending_revision.take();194+ match result {195+ Some(Ok(acknowledgement)) => match acknowledgement.result {196+ Ok(status) => {197+ let elapsed_micros =198+ u64::try_from(acknowledgement.elapsed.as_micros()).unwrap_or(u64::MAX);199+ self.telemetry200+ .encoder_force_acknowledgements201+ .fetch_add(1, Ordering::Relaxed);202+ self.telemetry203+ .last_encoder_force_ack_micros204+ .store(elapsed_micros, Ordering::Relaxed);205+ debug!(206+ %remote,207+ transport,208+ requested_revision = acknowledgement.requested_revision,209+ acknowledged_revision = ?status.acknowledged_force_keyframe_revision,210+ elapsed_micros,211+ "forced-IDR recovery acknowledged"212+ );213+ }214+ Err(error) => {215+ self.telemetry216+ .encoder_force_failures217+ .fetch_add(1, Ordering::Relaxed);218+ warn!(219+ %remote,220+ transport,221+ requested_revision = acknowledgement.requested_revision,222+ %error,223+ "forced-IDR recovery was not acknowledged; retaining natural-IDR fallback"224+ );225+ }226+ },227+ Some(Err(error)) => {228+ self.telemetry229+ .encoder_force_failures230+ .fetch_add(1, Ordering::Relaxed);231+ warn!(232+ %remote,233+ transport,234+ ?pending_revision,235+ %error,236+ "forced-IDR acknowledgement task failed; retaining natural-IDR fallback"237+ );238+ }239+ None => {240+ self.telemetry241+ .encoder_force_failures242+ .fetch_add(1, Ordering::Relaxed);243+ warn!(244+ %remote,245+ transport,246+ ?pending_revision,247+ "forced-IDR acknowledgement task ended without a result"248+ );249+ }250+ }251+ }252+253+ pub(super) async fn abort_and_drain(&mut self, remote: EndpointId, transport: &'static str) {254+ self.pending_revision = None;255+ self.acknowledgements.abort_all();256+ while self.acknowledgements.join_next().await.is_some() {}257+ debug!(258+ %remote,259+ transport,260+ keyframe_control_requests = self261+ .telemetry262+ .keyframe_control_requests263+ .load(Ordering::Relaxed),264+ encoder_force_requests = self265+ .telemetry266+ .encoder_force_requests267+ .load(Ordering::Relaxed),268+ encoder_force_acknowledgements = self269+ .telemetry270+ .encoder_force_acknowledgements271+ .load(Ordering::Relaxed),272+ encoder_force_failures = self273+ .telemetry274+ .encoder_force_failures275+ .load(Ordering::Relaxed),276+ last_encoder_force_ack_micros = self277+ .telemetry278+ .last_encoder_force_ack_micros279+ .load(Ordering::Relaxed),280+ "forced-IDR recovery session summary"281+ );282+ }283+}284+285+impl MediaV3Telemetry {286+ pub(super) fn snapshot(&self) -> MediaV3TelemetrySnapshot {287+ MediaV3TelemetrySnapshot {288+ scheduler_cancellations: self.scheduler_cancellations.load(Ordering::Relaxed),289+ send_failures: self.send_failures.load(Ordering::Relaxed),290+ selected_path_rtt_micros: self.selected_path_rtt_micros.load(Ordering::Relaxed),291+ selected_path_lost_packets: self.selected_path_lost_packets.load(Ordering::Relaxed),292+ selected_path_congestion_events: self293+ .selected_path_congestion_events294+ .load(Ordering::Relaxed),295+ }296+ }297+298+ pub(super) fn record_selected_path(&self, connection: &Connection) {299+ let paths = connection.paths();300+ let Some(path) = paths.iter().find(|path| path.is_selected()) else {301+ return;302+ };303+ let stats = path.stats();304+ self.selected_path_rtt_micros.store(305+ u64::try_from(stats.rtt.as_micros()).unwrap_or(u64::MAX),306+ Ordering::Relaxed,307+ );308+ self.selected_path_lost_packets309+ .store(stats.lost_packets, Ordering::Relaxed);310+ self.selected_path_congestion_events311+ .store(stats.congestion_events, Ordering::Relaxed);312+ }313+}314+315+impl Default for SessionRegistry {316+ fn default() -> Self {317+ Self {318+ active: Mutex::new(None),319+ pending_moq: Mutex::new(None),320+ next_session_id: AtomicU64::new(0),321+ session_changed: tokio::sync::Notify::new(),322+ pending_handshakes: tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES),323+ }324+ }325+}326+327+impl SessionRegistry {328+ pub fn has_session(&self) -> bool {329+ self.active330+ .lock()331+ .expect("session registry poisoned")332+ .is_some()333+ }334+335+ pub(super) fn claim(336+ self: &Arc<Self>,337+ remote: EndpointId,338+ nonce: [u8; 16],339+ grants: InvitationGrants,340+ ) -> Result<SessionLease> {341+ let mut active = self.active.lock().expect("session registry poisoned");342+ if let Some(current) = active.as_ref() {343+ bail!("host already has active client {}", current.remote);344+ }345+ let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;346+ let session_clock = SessionClock::start();347+ let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());348+ *active = Some(ActiveSession {349+ remote,350+ session_id,351+ nonce,352+ session_clock,353+ grants,354+ media_active: true,355+ input_claimed: false,356+ audio_claimed: false,357+ feedback_claimed: false,358+ media_v3_telemetry: Arc::clone(&media_v3_telemetry),359+ encoder_control: None,360+ });361+ Ok(SessionLease {362+ registry: Arc::clone(self),363+ remote,364+ session_id,365+ session_clock,366+ media_v3_telemetry,367+ })368+ }369+370+ pub(super) fn claim_input(371+ self: &Arc<Self>,372+ remote: EndpointId,373+ nonce: [u8; 16],374+ ) -> Result<InputLease> {375+ let mut active = self.active.lock().expect("session registry poisoned");376+ let session = active377+ .as_mut()378+ .filter(|session| {379+ session.media_active && session.remote == remote && session.nonce == nonce380+ })381+ .context("input connection does not match the active media session")?;382+ ensure!(383+ !session.input_claimed,384+ "active client already has an input stream"385+ );386+ session.input_claimed = true;387+ Ok(InputLease {388+ registry: Arc::clone(self),389+ remote,390+ session_id: session.session_id,391+ grants: session.grants,392+ })393+ }394+395+ pub(super) fn install_encoder_control(396+ &self,397+ remote: EndpointId,398+ session_id: u64,399+ encoder_control: Option<EncoderControl>,400+ ) -> Result<()> {401+ let mut active = self.active.lock().expect("session registry poisoned");402+ let session = active403+ .as_mut()404+ .filter(|session| {405+ session.media_active && session.remote == remote && session.session_id == session_id406+ })407+ .context("encoder control does not match the active media session")?;408+ ensure!(409+ session.encoder_control.is_none(),410+ "active media session already has encoder control"411+ );412+ session.encoder_control = encoder_control;413+ Ok(())414+ }415+416+ pub(super) fn propose_adaptive_encoder_update(417+ &self,418+ remote: EndpointId,419+ session_id: u64,420+ target_kbps: u32,421+ force_keyframe: bool,422+ ) -> Result<Option<AdaptiveEncoderProposal>> {423+ let active = self.active.lock().expect("session registry poisoned");424+ let session = active425+ .as_ref()426+ .filter(|session| {427+ session.media_active && session.remote == remote && session.session_id == session_id428+ })429+ .context("adaptive encoder update does not match the active media session")?;430+ let Some(control) = session.encoder_control.clone() else {431+ return Ok(None);432+ };433+ let bitrate_revision = control.request_bitrate_kbps(target_kbps)?;434+ let force_keyframe_revision = force_keyframe435+ .then(|| control.request_force_keyframe())436+ .transpose()?;437+ Ok(Some(AdaptiveEncoderProposal {438+ control,439+ target_kbps,440+ bitrate_revision,441+ force_keyframe_revision,442+ }))443+ }444+445+ pub(super) fn propose_resolution_update(446+ &self,447+ remote: EndpointId,448+ session_id: u64,449+ target: VideoDimensions,450+ ) -> Result<Option<ResolutionEncoderProposal>> {451+ let active = self.active.lock().expect("session registry poisoned");452+ let session = active453+ .as_ref()454+ .filter(|session| {455+ session.media_active && session.remote == remote && session.session_id == session_id456+ })457+ .context("resolution update does not match the active media session")?;458+ let Some(control) = session.encoder_control.clone() else {459+ return Ok(None);460+ };461+ let revision = control.request_resolution(target.width, target.height)?;462+ Ok(Some(ResolutionEncoderProposal {463+ control,464+ target,465+ revision,466+ }))467+ }468+469+ pub(super) fn claim_feedback(470+ self: &Arc<Self>,471+ remote: EndpointId,472+ nonce: [u8; 16],473+ ) -> Result<FeedbackLease> {474+ let mut active = self.active.lock().expect("session registry poisoned");475+ let session = active476+ .as_mut()477+ .filter(|session| {478+ session.media_active && session.remote == remote && session.nonce == nonce479+ })480+ .context("feedback connection does not match the active media session")?;481+ ensure!(482+ session.grants.contains(InvitationGrants::VIEW),483+ "active Portal session lacks feedback view permission"484+ );485+ ensure!(486+ !session.feedback_claimed,487+ "active client already has a feedback connection"488+ );489+ session.feedback_claimed = true;490+ Ok(FeedbackLease {491+ registry: Arc::clone(self),492+ remote,493+ session_id: session.session_id,494+ telemetry: Arc::clone(&session.media_v3_telemetry),495+ encoder_control: session.encoder_control.clone(),496+ })497+ }498+499+ pub(super) fn claim_audio(500+ self: &Arc<Self>,501+ remote: EndpointId,502+ nonce: [u8; 16],503+ ) -> Result<AudioLease> {504+ let mut active = self.active.lock().expect("session registry poisoned");505+ let session = active506+ .as_mut()507+ .filter(|session| {508+ session.media_active && session.remote == remote && session.nonce == nonce509+ })510+ .context("audio connection does not match the active media session")?;511+ ensure!(512+ !session.audio_claimed,513+ "active client already has an audio connection"514+ );515+ session.audio_claimed = true;516+ Ok(AudioLease {517+ registry: Arc::clone(self),518+ remote,519+ session_id: session.session_id,520+ session_clock: session.session_clock,521+ grants: session.grants,522+ })523+ }524+525+ pub(super) fn expect_moq(526+ &self,527+ remote: EndpointId,528+ session_id: u64,529+ broadcast_name: String,530+ broadcast: BroadcastConsumer,531+ ) -> Result<MoqAttachmentWait> {532+ let active = self.active.lock().expect("session registry poisoned");533+ let telemetry = active534+ .as_ref()535+ .filter(|session| {536+ session.media_active && session.remote == remote && session.session_id == session_id537+ })538+ .map(|session| Arc::clone(&session.media_v3_telemetry))539+ .context("MoQ expectation does not match the active control session")?;540+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");541+ ensure!(542+ pending.is_none(),543+ "active control session already expects MoQ"544+ );545+ let (attached, attached_rx) = tokio::sync::oneshot::channel();546+ let (closed, closed_rx) = tokio::sync::oneshot::channel();547+ *pending = Some(PendingMoqAttachment {548+ remote,549+ session_id,550+ broadcast_name,551+ broadcast,552+ attached,553+ closed,554+ telemetry,555+ });556+ Ok(MoqAttachmentWait {557+ attached: attached_rx,558+ closed: closed_rx,559+ })560+ }561+562+ pub(super) fn claim_moq(&self, remote: EndpointId) -> Result<ClaimedMoqAttachment> {563+ let active = self.active.lock().expect("session registry poisoned");564+ let session = active565+ .as_ref()566+ .filter(|session| session.media_active && session.remote == remote)567+ .context("MoQ connection does not match the active control session")?;568+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");569+ let attachment = pending570+ .as_ref()571+ .filter(|attachment| {572+ attachment.remote == remote && attachment.session_id == session.session_id573+ })574+ .context("active control session is not expecting a MoQ connection")?;575+ debug_assert_eq!(attachment.remote, session.remote);576+ let attachment = pending577+ .take()578+ .expect("validated pending MoQ attachment disappeared");579+ Ok(ClaimedMoqAttachment {580+ session_id: attachment.session_id,581+ broadcast_name: attachment.broadcast_name,582+ broadcast: attachment.broadcast,583+ attached: attachment.attached,584+ closed: attachment.closed,585+ telemetry: attachment.telemetry,586+ })587+ }588+589+ fn release(&self, remote: EndpointId, session_id: u64) {590+ let mut active = self.active.lock().expect("session registry poisoned");591+ if let Some(session) = active.as_mut()592+ && session.remote == remote593+ && session.session_id == session_id594+ {595+ let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");596+ if pending.as_ref().is_some_and(|attachment| {597+ attachment.remote == remote && attachment.session_id == session_id598+ }) {599+ *pending = None;600+ }601+ // Keep the registry occupied until the input handler has observed602+ // media shutdown and released all held uinput transitions. This603+ // prevents a reconnect from sharing the device with a draining604+ // predecessor session.605+ session.media_active = false;606+ session.encoder_control = None;607+ if !session.input_claimed && !session.audio_claimed && !session.feedback_claimed {608+ *active = None;609+ }610+ drop(active);611+ self.session_changed.notify_waiters();612+ }613+ }614+615+ pub(super) fn is_active(&self, remote: EndpointId, session_id: u64) -> bool {616+ self.active617+ .lock()618+ .expect("session registry poisoned")619+ .as_ref()620+ .is_some_and(|active| {621+ active.media_active && active.remote == remote && active.session_id == session_id622+ })623+ }624+625+ fn release_input(&self, remote: EndpointId, session_id: u64) {626+ let mut active = self.active.lock().expect("session registry poisoned");627+ if let Some(session) = active.as_mut()628+ && session.remote == remote629+ && session.session_id == session_id630+ {631+ session.input_claimed = false;632+ if !session.media_active && !session.audio_claimed && !session.feedback_claimed {633+ *active = None;634+ }635+ }636+ }637+638+ fn release_audio(&self, remote: EndpointId, session_id: u64) {639+ let mut active = self.active.lock().expect("session registry poisoned");640+ if let Some(session) = active.as_mut()641+ && session.remote == remote642+ && session.session_id == session_id643+ {644+ session.audio_claimed = false;645+ if !session.media_active && !session.input_claimed && !session.feedback_claimed {646+ *active = None;647+ }648+ }649+ }650+651+ fn release_feedback(&self, remote: EndpointId, session_id: u64) {652+ let mut active = self.active.lock().expect("session registry poisoned");653+ if let Some(session) = active.as_mut()654+ && session.remote == remote655+ && session.session_id == session_id656+ {657+ session.feedback_claimed = false;658+ if !session.media_active && !session.input_claimed && !session.audio_claimed {659+ *active = None;660+ }661+ }662+ }663+}664+665+666+pub(super) struct SessionLease {667+ registry: Arc<SessionRegistry>,668+ remote: EndpointId,669+ pub(super) session_id: u64,670+ pub(super) session_clock: SessionClock,671+ pub(super) media_v3_telemetry: Arc<MediaV3Telemetry>,672+}673+674+impl Drop for SessionLease {675+ fn drop(&mut self) {676+ self.registry.release(self.remote, self.session_id);677+ }678+}679+680+681+pub(super) struct InputLease {682+ registry: Arc<SessionRegistry>,683+ remote: EndpointId,684+ pub(super) session_id: u64,685+ pub(super) grants: InvitationGrants,686+}687+688+689+pub(super) struct AudioLease {690+ registry: Arc<SessionRegistry>,691+ remote: EndpointId,692+ pub(super) session_id: u64,693+ pub(super) session_clock: SessionClock,694+ pub(super) grants: InvitationGrants,695+}696+697+698+pub(super) struct FeedbackLease {699+ registry: Arc<SessionRegistry>,700+ remote: EndpointId,701+ pub(super) session_id: u64,702+ pub(super) telemetry: Arc<MediaV3Telemetry>,703+ pub(super) encoder_control: Option<EncoderControl>,704+}705+706+707+pub(super) struct SourceTaskGuard(Option<tokio::task::JoinHandle<Result<()>>>);708+709+impl SourceTaskGuard {710+ pub(super) fn new(task: tokio::task::JoinHandle<Result<()>>) -> Self {711+ Self(Some(task))712+ }713+714+ pub(super) async fn abort_and_wait(mut self) {715+ if let Some(task) = self.0.take() {716+ task.abort();717+ let _ = task.await;718+ }719+ }720+721+ pub(super) async fn wait_or_abort(mut self, grace_timeout: Duration) {722+ let Some(mut task) = self.0.take() else {723+ return;724+ };725+ if tokio::time::timeout(grace_timeout, &mut task)726+ .await727+ .is_err()728+ {729+ task.abort();730+ let _ = task.await;731+ }732+ }733+}734+735+impl Drop for SourceTaskGuard {736+ fn drop(&mut self) {737+ if let Some(task) = self.0.take() {738+ task.abort();739+ }740+ }741+}742+743+impl Drop for InputLease {744+ fn drop(&mut self) {745+ self.registry.release_input(self.remote, self.session_id);746+ }747+}748+749+impl Drop for AudioLease {750+ fn drop(&mut self) {751+ self.registry.release_audio(self.remote, self.session_id);752+ }753+}754+755+impl Drop for FeedbackLease {756+ fn drop(&mut self) {757+ self.registry.release_feedback(self.remote, self.session_id);758+ }759+}760+761+762+mod tests {763+ use super::super::endpoint;764+ use super::*;765+766+ use moq_net::{Broadcast, BroadcastProducer};767+ use sigil_protocol::media_moq_broadcast_name;768+769+ 770+ async fn forced_idr_recovery_is_one_slot_join_safe_and_rearms_after_ack() {771+ let telemetry = Arc::new(MediaV3Telemetry::default());772+ let harness = crate::source::EncoderControlTestHarness::new();773+ let mut coordinator =774+ ForcedIdrCoordinator::new(Some(harness.control.clone()), Arc::clone(&telemetry));775+776+ assert_eq!(777+ coordinator.request(KeyframeRequestReasonV3::Join),778+ ForcedIdrDisposition::JoinReplay779+ );780+ assert_eq!(harness.requested_force_keyframe_revision(), None);781+782+ let requested_revision = match coordinator.request(KeyframeRequestReasonV3::DecoderReset) {783+ ForcedIdrDisposition::Requested { revision } => revision,784+ disposition => panic!("unexpected forced-IDR disposition: {disposition:?}"),785+ };786+ assert_eq!(787+ harness.requested_force_keyframe_revision(),788+ Some(requested_revision)789+ );790+ assert_eq!(791+ coordinator.request(KeyframeRequestReasonV3::TransportGap),792+ ForcedIdrDisposition::Coalesced {793+ revision: requested_revision794+ }795+ );796+ assert_eq!(telemetry.encoder_force_requests.load(Ordering::Relaxed), 1);797+798+ let newer_revision = harness.control.request_force_keyframe().unwrap();799+ harness.status.send_modify(|status| {800+ status.requested_force_keyframe_revision = Some(newer_revision);801+ status.acknowledged_force_keyframe_revision = Some(newer_revision);802+ });803+ let acknowledgement = coordinator.acknowledgements.join_next().await;804+ coordinator.complete(acknowledgement, endpoint(1), "test");805+ assert_eq!(coordinator.pending_revision, None);806+ assert_eq!(807+ telemetry808+ .encoder_force_acknowledgements809+ .load(Ordering::Relaxed),810+ 1811+ );812+ assert_eq!(telemetry.encoder_force_failures.load(Ordering::Relaxed), 0);813+814+ assert!(matches!(815+ coordinator.request(KeyframeRequestReasonV3::DecoderReset),816+ ForcedIdrDisposition::Requested { revision } if revision > newer_revision817+ ));818+ coordinator.abort_and_drain(endpoint(1), "test").await;819+820+ let fallback_telemetry = Arc::new(MediaV3Telemetry::default());821+ let mut fallback = ForcedIdrCoordinator::new(None, fallback_telemetry);822+ assert_eq!(823+ fallback.request(KeyframeRequestReasonV3::DecoderReset),824+ ForcedIdrDisposition::Unavailable825+ );826+ assert!(fallback.acknowledgements.is_empty());827+ }828+829+ 830+ fn only_one_remote_can_hold_session() {831+ let sessions = Arc::new(SessionRegistry::default());832+ assert!(!sessions.has_session());833+ let nonce = [7; 16];834+ let first = sessions835+ .claim(endpoint(1), nonce, InvitationGrants::ALL)836+ .unwrap();837+ assert!(sessions.has_session());838+ assert!(839+ sessions840+ .claim(endpoint(2), nonce, InvitationGrants::ALL)841+ .is_err()842+ );843+ assert!(sessions.claim_input(endpoint(1), [8; 16]).is_err());844+ let input = sessions.claim_input(endpoint(1), nonce).unwrap();845+ assert_eq!(input.session_id, first.session_id);846+ assert!(sessions.claim_input(endpoint(1), nonce).is_err());847+ let audio = sessions.claim_audio(endpoint(1), nonce).unwrap();848+ assert!(sessions.claim_audio(endpoint(1), nonce).is_err());849+ drop(input);850+ let draining_input = sessions.claim_input(endpoint(1), nonce).unwrap();851+ drop(first);852+ assert!(853+ sessions854+ .claim(endpoint(2), nonce, InvitationGrants::ALL)855+ .is_err()856+ );857+ drop(draining_input);858+ assert!(859+ sessions860+ .claim(endpoint(2), nonce, InvitationGrants::ALL)861+ .is_err()862+ );863+ drop(audio);864+ assert!(!sessions.has_session());865+ assert!(866+ sessions867+ .claim(endpoint(2), nonce, InvitationGrants::ALL)868+ .is_ok()869+ );870+ }871+872+ fn test_moq_broadcast() -> (BroadcastProducer, BroadcastConsumer) {873+ let producer = Broadcast::new().produce();874+ let consumer = producer.consume();875+ (producer, consumer)876+ }877+878+ 879+ fn moq_attachment_requires_exact_active_control_remote_and_is_single_use() {880+ let sessions = Arc::new(SessionRegistry::default());881+ assert!(sessions.claim_moq(endpoint(1)).is_err());882+ let lease = sessions883+ .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)884+ .unwrap();885+ let (_producer, consumer) = test_moq_broadcast();886+ let _wait = sessions887+ .expect_moq(888+ endpoint(1),889+ lease.session_id,890+ media_moq_broadcast_name(lease.session_id).unwrap(),891+ consumer,892+ )893+ .unwrap();894+895+ // A wrong peer cannot consume the exact pending attachment.896+ assert!(sessions.claim_moq(endpoint(2)).is_err());897+ let attachment = sessions.claim_moq(endpoint(1)).unwrap();898+ assert_eq!(attachment.session_id, lease.session_id);899+ assert_eq!(900+ attachment.broadcast_name,901+ media_moq_broadcast_name(lease.session_id).unwrap()902+ );903+ // The pending token was atomically consumed before the MoQ handshake.904+ assert!(sessions.claim_moq(endpoint(1)).is_err());905+ }906+907+ 908+ fn competing_moq_connections_cannot_both_claim_one_control_attachment() {909+ let sessions = Arc::new(SessionRegistry::default());910+ let lease = sessions911+ .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)912+ .unwrap();913+ let (_producer, consumer) = test_moq_broadcast();914+ let _wait = sessions915+ .expect_moq(916+ endpoint(1),917+ lease.session_id,918+ media_moq_broadcast_name(lease.session_id).unwrap(),919+ consumer,920+ )921+ .unwrap();922+ let barrier = Arc::new(std::sync::Barrier::new(3));923+ let contenders = (0..2)924+ .map(|_| {925+ let sessions = Arc::clone(&sessions);926+ let barrier = Arc::clone(&barrier);927+ std::thread::spawn(move || {928+ barrier.wait();929+ sessions.claim_moq(endpoint(1)).is_ok()930+ })931+ })932+ .collect::<Vec<_>>();933+ barrier.wait();934+ let claimed = contenders935+ .into_iter()936+ .map(|thread| usize::from(thread.join().unwrap()))937+ .sum::<usize>();938+ assert_eq!(claimed, 1);939+ }940+941+ 942+ async fn releasing_control_clears_an_unclaimed_moq_attachment() {943+ let sessions = Arc::new(SessionRegistry::default());944+ let lease = sessions945+ .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)946+ .unwrap();947+ let (_producer, consumer) = test_moq_broadcast();948+ let wait = sessions949+ .expect_moq(950+ endpoint(1),951+ lease.session_id,952+ media_moq_broadcast_name(lease.session_id).unwrap(),953+ consumer,954+ )955+ .unwrap();956+ drop(lease);957+ assert!(sessions.claim_moq(endpoint(1)).is_err());958+ assert!(wait.attached.await.is_err());959+ assert!(wait.closed.await.is_err());960+ }961+962+ 963+ fn feedback_attaches_only_to_exact_active_view_session() {964+ let sessions = Arc::new(SessionRegistry::default());965+ let nonce = [9; 16];966+ assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());967+968+ let no_view = sessions969+ .claim(endpoint(1), nonce, InvitationGrants::GAMEPAD)970+ .unwrap();971+ assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());972+ drop(no_view);973+974+ let media = sessions975+ .claim(endpoint(1), nonce, InvitationGrants::VIEW)976+ .unwrap();977+ assert!(sessions.claim_feedback(endpoint(2), nonce).is_err());978+ assert!(sessions.claim_feedback(endpoint(1), [8; 16]).is_err());979+ let feedback = sessions.claim_feedback(endpoint(1), nonce).unwrap();980+ assert_eq!(feedback.session_id, media.session_id);981+ assert!(feedback.encoder_control.is_none());982+ assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());983+984+ drop(media);985+ assert!(986+ sessions987+ .claim(endpoint(2), nonce, InvitationGrants::VIEW)988+ .is_err(),989+ "feedback teardown must keep the draining session isolated"990+ );991+ drop(feedback);992+ assert!(993+ sessions994+ .claim(endpoint(2), nonce, InvitationGrants::VIEW)995+ .is_ok()996+ );997+ }998+999+ 1000+ fn adaptive_encoder_proposals_are_bound_to_the_exact_active_generation() {1001+ let sessions = Arc::new(SessionRegistry::default());1002+ let remote = endpoint(1);1003+ let media = sessions1004+ .claim(remote, [3; 16], InvitationGrants::VIEW)1005+ .unwrap();1006+ let harness = crate::source::EncoderControlTestHarness::new();1007+1008+ assert!(1009+ sessions1010+ .install_encoder_control(1011+ endpoint(2),1012+ media.session_id,1013+ Some(harness.control.clone())1014+ )1015+ .is_err()1016+ );1017+ sessions1018+ .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))1019+ .unwrap();1020+ let feedback = sessions.claim_feedback(remote, [3; 16]).unwrap();1021+ assert!(feedback.encoder_control.is_some());1022+ let proposal = sessions1023+ .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)1024+ .unwrap()1025+ .unwrap();1026+ assert_eq!(proposal.target_kbps, 8_000);1027+ assert!(proposal.force_keyframe_revision > Some(proposal.bitrate_revision));1028+1029+ let old_session_id = media.session_id;1030+ drop(media);1031+ assert!(1032+ sessions1033+ .propose_adaptive_encoder_update(remote, old_session_id, 7_000, false)1034+ .is_err(),1035+ "a draining generation must not issue another encoder proposal"1036+ );1037+ drop(feedback);1038+ }1039+1040+ 1041+ fn pending_handshakes_are_bounded() {1042+ let sessions = SessionRegistry::default();1043+ let permits: Vec<_> = (0..MAX_PENDING_HANDSHAKES)1044+ .map(|_| sessions.pending_handshakes.try_acquire().unwrap())1045+ .collect();1046+ assert!(sessions.pending_handshakes.try_acquire().is_err());1047+ drop(permits);1048+ assert!(sessions.pending_handshakes.try_acquire().is_ok());1049+ }1050+1051+ 1052+ fn session_substreams_inherit_the_exact_enrollment_grant() {1053+ let sessions = Arc::new(SessionRegistry::default());1054+ let grants = InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD);1055+ let media = sessions.claim(endpoint(1), [3; 16], grants).unwrap();1056+ let input = sessions.claim_input(endpoint(1), [3; 16]).unwrap();1057+ let audio = sessions.claim_audio(endpoint(1), [3; 16]).unwrap();1058+ assert_eq!(input.grants, grants);1059+ assert_eq!(audio.grants, grants);1060+ drop(input);1061+ drop(audio);1062+ drop(media);1063+ }1064+1065+ 1066+ fn audio_claim_requires_the_active_remote_and_nonce() {1067+ let sessions = Arc::new(SessionRegistry::default());1068+ let media = sessions1069+ .claim(endpoint(1), [9; 16], InvitationGrants::ALL)1070+ .unwrap();1071+ assert!(sessions.claim_audio(endpoint(2), [9; 16]).is_err());1072+ assert!(sessions.claim_audio(endpoint(1), [8; 16]).is_err());1073+ let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();1074+ drop(media);1075+ assert!(1076+ sessions1077+ .claim(endpoint(2), [0; 16], InvitationGrants::ALL)1078+ .is_err()1079+ );1080+ drop(audio);1081+ assert!(1082+ sessions1083+ .claim(endpoint(2), [0; 16], InvitationGrants::ALL)1084+ .is_ok()1085+ );1086+ }1087+1088+ 1089+ fn media_and_audio_leases_share_one_session_clock() {1090+ let sessions = Arc::new(SessionRegistry::default());1091+ let media = sessions1092+ .claim(endpoint(1), [9; 16], InvitationGrants::ALL)1093+ .unwrap();1094+ let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();1095+ assert_eq!(media.session_clock, audio.session_clock);1096+ }1097+}