1use std::sync::atomic::{AtomicU64, Ordering};2use std::sync::{Arc, Mutex};3use std::time::{Duration, Instant};45use anyhow::{Context, Result, bail, ensure};6use iroh::EndpointId;7use iroh::endpoint::Connection;8use moq_net::BroadcastConsumer;9use sigil_protocol::{InvitationGrants, KeyframeRequestReasonV3};10use tracing::{debug, warn};1112use super::{ENCODER_CONTROL_COMMIT_TIMEOUT, VideoDimensions};13use crate::clock::SessionClock;14use crate::source::EncoderControl;1516const MAX_PENDING_HANDSHAKES: usize = 4;171819pub struct SessionRegistry {20 active: Mutex<Option<ActiveSession>>,21 pending_moq: Mutex<Option<PendingMoqAttachment>>,22 next_session_id: AtomicU64,23 pub(super) session_changed: tokio::sync::Notify,24 pub(super) pending_handshakes: tokio::sync::Semaphore,25}2627struct PendingMoqAttachment {28 remote: EndpointId,29 session_id: u64,30 broadcast_name: String,31 broadcast: BroadcastConsumer,32 attached: tokio::sync::oneshot::Sender<()>,33 closed: tokio::sync::oneshot::Sender<()>,34 telemetry: Arc<MediaV3Telemetry>,35}3637impl std::fmt::Debug for PendingMoqAttachment {38 fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {39 formatter40 .debug_struct("PendingMoqAttachment")41 .field("remote", &self.remote)42 .field("session_id", &self.session_id)43 .field("broadcast_name", &self.broadcast_name)44 .finish_non_exhaustive()45 }46}4748pub(super) struct ClaimedMoqAttachment {49 pub(super) session_id: u64,50 pub(super) broadcast_name: String,51 pub(super) broadcast: BroadcastConsumer,52 pub(super) attached: tokio::sync::oneshot::Sender<()>,53 pub(super) closed: tokio::sync::oneshot::Sender<()>,54 pub(super) telemetry: Arc<MediaV3Telemetry>,55}5657pub(super) struct MoqAttachmentWait {58 pub(super) attached: tokio::sync::oneshot::Receiver<()>,59 pub(super) closed: tokio::sync::oneshot::Receiver<()>,60}616263struct ActiveSession {64 remote: EndpointId,65 session_id: u64,66 nonce: [u8; 16],67 session_clock: SessionClock,68 grants: InvitationGrants,69 media_active: bool,70 input_claimed: bool,71 audio_claimed: bool,72 feedback_claimed: bool,73 media_v3_telemetry: Arc<MediaV3Telemetry>,74 encoder_control: Option<EncoderControl>,75}767778pub(super) struct MediaV3Telemetry {79 pub(super) scheduler_cancellations: AtomicU64,80 pub(super) send_failures: AtomicU64,81 selected_path_rtt_micros: AtomicU64,82 selected_path_lost_packets: AtomicU64,83 selected_path_congestion_events: AtomicU64,84 keyframe_control_requests: AtomicU64,85 encoder_force_requests: AtomicU64,86 encoder_force_acknowledgements: AtomicU64,87 encoder_force_failures: AtomicU64,88 last_encoder_force_ack_micros: AtomicU64,89}909192pub(super) struct MediaV3TelemetrySnapshot {93 pub(super) scheduler_cancellations: u64,94 pub(super) send_failures: u64,95 pub(super) selected_path_rtt_micros: u64,96 pub(super) selected_path_lost_packets: u64,97 pub(super) selected_path_congestion_events: u64,98}99100101pub(super) struct AdaptiveEncoderProposal {102 pub(super) control: EncoderControl,103 pub(super) target_kbps: u32,104 pub(super) bitrate_revision: u64,105 pub(super) force_keyframe_revision: Option<u64>,106}107108109pub(super) struct ResolutionEncoderProposal {110 pub(super) control: EncoderControl,111 pub(super) target: VideoDimensions,112 pub(super) revision: u64,113}114115116pub(super) enum ForcedIdrDisposition {117 JoinReplay,118 Unavailable,119 Requested { revision: u64 },120 Coalesced { revision: u64 },121 Failed { error: String },122}123124pub(super) struct ForcedIdrAcknowledgement {125 requested_revision: u64,126 elapsed: Duration,127 result: Result<crate::source::EncoderControlStatus>,128}129130pub(super) struct ForcedIdrCoordinator {131 control: Option<EncoderControl>,132 pub(super) pending_revision: Option<u64>,133 pub(super) acknowledgements: tokio::task::JoinSet<ForcedIdrAcknowledgement>,134 telemetry: Arc<MediaV3Telemetry>,135}136137impl ForcedIdrCoordinator {138 pub(super) fn new(control: Option<EncoderControl>, telemetry: Arc<MediaV3Telemetry>) -> Self {139 Self {140 control,141 pending_revision: None,142 acknowledgements: tokio::task::JoinSet::new(),143 telemetry,144 }145 }146147 pub(super) fn request(&mut self, reason: KeyframeRequestReasonV3) -> ForcedIdrDisposition {148 self.telemetry149 .keyframe_control_requests150 .fetch_add(1, Ordering::Relaxed);151 if reason == KeyframeRequestReasonV3::Join {152 return ForcedIdrDisposition::JoinReplay;153 }154 if let Some(revision) = self.pending_revision {155 return ForcedIdrDisposition::Coalesced { revision };156 }157 let Some(control) = self.control.clone() else {158 return ForcedIdrDisposition::Unavailable;159 };160 let revision = match control.request_force_keyframe() {161 Ok(revision) => revision,162 Err(error) => {163 self.telemetry164 .encoder_force_failures165 .fetch_add(1, Ordering::Relaxed);166 return ForcedIdrDisposition::Failed {167 error: error.to_string(),168 };169 }170 };171 self.pending_revision = Some(revision);172 self.telemetry173 .encoder_force_requests174 .fetch_add(1, Ordering::Relaxed);175 self.acknowledgements.spawn(async move {176 let started_at = Instant::now();177 let result = control178 .wait_for_recovery_keyframe_acknowledged(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)179 .await;180 ForcedIdrAcknowledgement {181 requested_revision: revision,182 elapsed: started_at.elapsed(),183 result,184 }185 });186 ForcedIdrDisposition::Requested { revision }187 }188189 pub(super) fn complete(190 &mut self,191 result: Option<Result<ForcedIdrAcknowledgement, tokio::task::JoinError>>,192 remote: EndpointId,193 transport: &'static str,194 ) {195 let pending_revision = self.pending_revision.take();196 match result {197 Some(Ok(acknowledgement)) => match acknowledgement.result {198 Ok(status) => {199 let elapsed_micros =200 u64::try_from(acknowledgement.elapsed.as_micros()).unwrap_or(u64::MAX);201 self.telemetry202 .encoder_force_acknowledgements203 .fetch_add(1, Ordering::Relaxed);204 self.telemetry205 .last_encoder_force_ack_micros206 .store(elapsed_micros, Ordering::Relaxed);207 debug!(208 %remote,209 transport,210 requested_revision = acknowledgement.requested_revision,211 acknowledged_revision = ?status.acknowledged_force_keyframe_revision,212 elapsed_micros,213 "forced-IDR recovery acknowledged"214 );215 }216 Err(error) => {217 self.telemetry218 .encoder_force_failures219 .fetch_add(1, Ordering::Relaxed);220 warn!(221 %remote,222 transport,223 requested_revision = acknowledgement.requested_revision,224 %error,225 "forced-IDR recovery was not acknowledged; retaining natural-IDR fallback"226 );227 }228 },229 Some(Err(error)) => {230 self.telemetry231 .encoder_force_failures232 .fetch_add(1, Ordering::Relaxed);233 warn!(234 %remote,235 transport,236 ?pending_revision,237 %error,238 "forced-IDR acknowledgement task failed; retaining natural-IDR fallback"239 );240 }241 None => {242 self.telemetry243 .encoder_force_failures244 .fetch_add(1, Ordering::Relaxed);245 warn!(246 %remote,247 transport,248 ?pending_revision,249 "forced-IDR acknowledgement task ended without a result"250 );251 }252 }253 }254255 pub(super) async fn abort_and_drain(&mut self, remote: EndpointId, transport: &'static str) {256 self.pending_revision = None;257 self.acknowledgements.abort_all();258 while self.acknowledgements.join_next().await.is_some() {}259 debug!(260 %remote,261 transport,262 keyframe_control_requests = self263 .telemetry264 .keyframe_control_requests265 .load(Ordering::Relaxed),266 encoder_force_requests = self267 .telemetry268 .encoder_force_requests269 .load(Ordering::Relaxed),270 encoder_force_acknowledgements = self271 .telemetry272 .encoder_force_acknowledgements273 .load(Ordering::Relaxed),274 encoder_force_failures = self275 .telemetry276 .encoder_force_failures277 .load(Ordering::Relaxed),278 last_encoder_force_ack_micros = self279 .telemetry280 .last_encoder_force_ack_micros281 .load(Ordering::Relaxed),282 "forced-IDR recovery session summary"283 );284 }285}286287impl MediaV3Telemetry {288 pub(super) fn snapshot(&self) -> MediaV3TelemetrySnapshot {289 MediaV3TelemetrySnapshot {290 scheduler_cancellations: self.scheduler_cancellations.load(Ordering::Relaxed),291 send_failures: self.send_failures.load(Ordering::Relaxed),292 selected_path_rtt_micros: self.selected_path_rtt_micros.load(Ordering::Relaxed),293 selected_path_lost_packets: self.selected_path_lost_packets.load(Ordering::Relaxed),294 selected_path_congestion_events: self295 .selected_path_congestion_events296 .load(Ordering::Relaxed),297 }298 }299300 pub(super) fn record_selected_path(&self, connection: &Connection) {301 let paths = connection.paths();302 let Some(path) = paths.iter().find(|path| path.is_selected()) else {303 return;304 };305 let stats = path.stats();306 self.selected_path_rtt_micros.store(307 u64::try_from(stats.rtt.as_micros()).unwrap_or(u64::MAX),308 Ordering::Relaxed,309 );310 self.selected_path_lost_packets311 .store(stats.lost_packets, Ordering::Relaxed);312 self.selected_path_congestion_events313 .store(stats.congestion_events, Ordering::Relaxed);314 }315}316317impl Default for SessionRegistry {318 fn default() -> Self {319 Self {320 active: Mutex::new(None),321 pending_moq: Mutex::new(None),322 next_session_id: AtomicU64::new(0),323 session_changed: tokio::sync::Notify::new(),324 pending_handshakes: tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES),325 }326 }327}328329impl SessionRegistry {330 pub fn has_session(&self) -> bool {331 self.active332 .lock()333 .expect("session registry poisoned")334 .is_some()335 }336337 pub(super) fn claim(338 self: &Arc<Self>,339 remote: EndpointId,340 nonce: [u8; 16],341 grants: InvitationGrants,342 ) -> Result<SessionLease> {343 let mut active = self.active.lock().expect("session registry poisoned");344 if let Some(current) = active.as_ref() {345 bail!("host already has active client {}", current.remote);346 }347 let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;348 let session_clock = SessionClock::start();349 let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());350 *active = Some(ActiveSession {351 remote,352 session_id,353 nonce,354 session_clock,355 grants,356 media_active: true,357 input_claimed: false,358 audio_claimed: false,359 feedback_claimed: false,360 media_v3_telemetry: Arc::clone(&media_v3_telemetry),361 encoder_control: None,362 });363 Ok(SessionLease {364 registry: Arc::clone(self),365 remote,366 session_id,367 session_clock,368 media_v3_telemetry,369 })370 }371372 pub(super) fn claim_input(373 self: &Arc<Self>,374 remote: EndpointId,375 nonce: [u8; 16],376 ) -> Result<InputLease> {377 let mut active = self.active.lock().expect("session registry poisoned");378 let session = active379 .as_mut()380 .filter(|session| {381 session.media_active && session.remote == remote && session.nonce == nonce382 })383 .context("input connection does not match the active media session")?;384 ensure!(385 !session.input_claimed,386 "active client already has an input stream"387 );388 session.input_claimed = true;389 Ok(InputLease {390 registry: Arc::clone(self),391 remote,392 session_id: session.session_id,393 grants: session.grants,394 })395 }396397 pub(super) fn install_encoder_control(398 &self,399 remote: EndpointId,400 session_id: u64,401 encoder_control: Option<EncoderControl>,402 ) -> Result<()> {403 let mut active = self.active.lock().expect("session registry poisoned");404 let session = active405 .as_mut()406 .filter(|session| {407 session.media_active && session.remote == remote && session.session_id == session_id408 })409 .context("encoder control does not match the active media session")?;410 ensure!(411 session.encoder_control.is_none(),412 "active media session already has encoder control"413 );414 session.encoder_control = encoder_control;415 Ok(())416 }417418 pub(super) fn propose_adaptive_encoder_update(419 &self,420 remote: EndpointId,421 session_id: u64,422 target_kbps: u32,423 force_keyframe: bool,424 ) -> Result<Option<AdaptiveEncoderProposal>> {425 let active = self.active.lock().expect("session registry poisoned");426 let session = active427 .as_ref()428 .filter(|session| {429 session.media_active && session.remote == remote && session.session_id == session_id430 })431 .context("adaptive encoder update does not match the active media session")?;432 let Some(control) = session.encoder_control.clone() else {433 return Ok(None);434 };435 let bitrate_revision = control.request_bitrate_kbps(target_kbps)?;436 let force_keyframe_revision = force_keyframe437 .then(|| control.request_force_keyframe())438 .transpose()?;439 Ok(Some(AdaptiveEncoderProposal {440 control,441 target_kbps,442 bitrate_revision,443 force_keyframe_revision,444 }))445 }446447 pub(super) fn propose_resolution_update(448 &self,449 remote: EndpointId,450 session_id: u64,451 target: VideoDimensions,452 ) -> Result<Option<ResolutionEncoderProposal>> {453 let active = self.active.lock().expect("session registry poisoned");454 let session = active455 .as_ref()456 .filter(|session| {457 session.media_active && session.remote == remote && session.session_id == session_id458 })459 .context("resolution update does not match the active media session")?;460 let Some(control) = session.encoder_control.clone() else {461 return Ok(None);462 };463 let revision = control.request_resolution(target.width, target.height)?;464 Ok(Some(ResolutionEncoderProposal {465 control,466 target,467 revision,468 }))469 }470471 pub(super) fn claim_feedback(472 self: &Arc<Self>,473 remote: EndpointId,474 nonce: [u8; 16],475 ) -> Result<FeedbackLease> {476 let mut active = self.active.lock().expect("session registry poisoned");477 let session = active478 .as_mut()479 .filter(|session| {480 session.media_active && session.remote == remote && session.nonce == nonce481 })482 .context("feedback connection does not match the active media session")?;483 ensure!(484 session.grants.contains(InvitationGrants::VIEW),485 "active Portal session lacks feedback view permission"486 );487 ensure!(488 !session.feedback_claimed,489 "active client already has a feedback connection"490 );491 session.feedback_claimed = true;492 Ok(FeedbackLease {493 registry: Arc::clone(self),494 remote,495 session_id: session.session_id,496 telemetry: Arc::clone(&session.media_v3_telemetry),497 encoder_control: session.encoder_control.clone(),498 })499 }500501 pub(super) fn claim_audio(502 self: &Arc<Self>,503 remote: EndpointId,504 nonce: [u8; 16],505 ) -> Result<AudioLease> {506 let mut active = self.active.lock().expect("session registry poisoned");507 let session = active508 .as_mut()509 .filter(|session| {510 session.media_active && session.remote == remote && session.nonce == nonce511 })512 .context("audio connection does not match the active media session")?;513 ensure!(514 !session.audio_claimed,515 "active client already has an audio connection"516 );517 session.audio_claimed = true;518 Ok(AudioLease {519 registry: Arc::clone(self),520 remote,521 session_id: session.session_id,522 session_clock: session.session_clock,523 grants: session.grants,524 })525 }526527 pub(super) fn expect_moq(528 &self,529 remote: EndpointId,530 session_id: u64,531 broadcast_name: String,532 broadcast: BroadcastConsumer,533 ) -> Result<MoqAttachmentWait> {534 let active = self.active.lock().expect("session registry poisoned");535 let telemetry = active536 .as_ref()537 .filter(|session| {538 session.media_active && session.remote == remote && session.session_id == session_id539 })540 .map(|session| Arc::clone(&session.media_v3_telemetry))541 .context("MoQ expectation does not match the active control session")?;542 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");543 ensure!(544 pending.is_none(),545 "active control session already expects MoQ"546 );547 let (attached, attached_rx) = tokio::sync::oneshot::channel();548 let (closed, closed_rx) = tokio::sync::oneshot::channel();549 *pending = Some(PendingMoqAttachment {550 remote,551 session_id,552 broadcast_name,553 broadcast,554 attached,555 closed,556 telemetry,557 });558 Ok(MoqAttachmentWait {559 attached: attached_rx,560 closed: closed_rx,561 })562 }563564 pub(super) fn claim_moq(&self, remote: EndpointId) -> Result<ClaimedMoqAttachment> {565 let active = self.active.lock().expect("session registry poisoned");566 let session = active567 .as_ref()568 .filter(|session| session.media_active && session.remote == remote)569 .context("MoQ connection does not match the active control session")?;570 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");571 let attachment = pending572 .as_ref()573 .filter(|attachment| {574 attachment.remote == remote && attachment.session_id == session.session_id575 })576 .context("active control session is not expecting a MoQ connection")?;577 debug_assert_eq!(attachment.remote, session.remote);578 let attachment = pending579 .take()580 .expect("validated pending MoQ attachment disappeared");581 Ok(ClaimedMoqAttachment {582 session_id: attachment.session_id,583 broadcast_name: attachment.broadcast_name,584 broadcast: attachment.broadcast,585 attached: attachment.attached,586 closed: attachment.closed,587 telemetry: attachment.telemetry,588 })589 }590591 fn release(&self, remote: EndpointId, session_id: u64) {592 let mut active = self.active.lock().expect("session registry poisoned");593 if let Some(session) = active.as_mut()594 && session.remote == remote595 && session.session_id == session_id596 {597 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");598 if pending.as_ref().is_some_and(|attachment| {599 attachment.remote == remote && attachment.session_id == session_id600 }) {601 *pending = None;602 }603 // Keep the registry occupied until the input handler has observed604 // media shutdown and released all held uinput transitions. This605 // prevents a reconnect from sharing the device with a draining606 // predecessor session.607 session.media_active = false;608 session.encoder_control = None;609 if !session.input_claimed && !session.audio_claimed && !session.feedback_claimed {610 *active = None;611 }612 drop(active);613 self.session_changed.notify_waiters();614 }615 }616617 pub(super) fn is_active(&self, remote: EndpointId, session_id: u64) -> bool {618 self.active619 .lock()620 .expect("session registry poisoned")621 .as_ref()622 .is_some_and(|active| {623 active.media_active && active.remote == remote && active.session_id == session_id624 })625 }626627 fn release_input(&self, remote: EndpointId, session_id: u64) {628 let mut active = self.active.lock().expect("session registry poisoned");629 if let Some(session) = active.as_mut()630 && session.remote == remote631 && session.session_id == session_id632 {633 session.input_claimed = false;634 if !session.media_active && !session.audio_claimed && !session.feedback_claimed {635 *active = None;636 }637 }638 }639640 fn release_audio(&self, remote: EndpointId, session_id: u64) {641 let mut active = self.active.lock().expect("session registry poisoned");642 if let Some(session) = active.as_mut()643 && session.remote == remote644 && session.session_id == session_id645 {646 session.audio_claimed = false;647 if !session.media_active && !session.input_claimed && !session.feedback_claimed {648 *active = None;649 }650 }651 }652653 fn release_feedback(&self, remote: EndpointId, session_id: u64) {654 let mut active = self.active.lock().expect("session registry poisoned");655 if let Some(session) = active.as_mut()656 && session.remote == remote657 && session.session_id == session_id658 {659 session.feedback_claimed = false;660 if !session.media_active && !session.input_claimed && !session.audio_claimed {661 *active = None;662 }663 }664 }665}666667668pub(super) struct SessionLease {669 registry: Arc<SessionRegistry>,670 remote: EndpointId,671 pub(super) session_id: u64,672 pub(super) session_clock: SessionClock,673 pub(super) media_v3_telemetry: Arc<MediaV3Telemetry>,674}675676impl Drop for SessionLease {677 fn drop(&mut self) {678 self.registry.release(self.remote, self.session_id);679 }680}681682683pub(super) struct InputLease {684 registry: Arc<SessionRegistry>,685 remote: EndpointId,686 pub(super) session_id: u64,687 pub(super) grants: InvitationGrants,688}689690691pub(super) struct AudioLease {692 registry: Arc<SessionRegistry>,693 remote: EndpointId,694 pub(super) session_id: u64,695 pub(super) session_clock: SessionClock,696 pub(super) grants: InvitationGrants,697}698699700pub(super) struct FeedbackLease {701 registry: Arc<SessionRegistry>,702 remote: EndpointId,703 pub(super) session_id: u64,704 pub(super) telemetry: Arc<MediaV3Telemetry>,705 pub(super) encoder_control: Option<EncoderControl>,706}707708709pub(super) struct SourceTaskGuard(Option<tokio::task::JoinHandle<Result<()>>>);710711impl SourceTaskGuard {712 pub(super) fn new(task: tokio::task::JoinHandle<Result<()>>) -> Self {713 Self(Some(task))714 }715716 pub(super) async fn abort_and_wait(mut self) {717 if let Some(task) = self.0.take() {718 task.abort();719 let _ = task.await;720 }721 }722723 pub(super) async fn wait_or_abort(mut self, grace_timeout: Duration) {724 let Some(mut task) = self.0.take() else {725 return;726 };727 if tokio::time::timeout(grace_timeout, &mut task)728 .await729 .is_err()730 {731 task.abort();732 let _ = task.await;733 }734 }735}736737impl Drop for SourceTaskGuard {738 fn drop(&mut self) {739 if let Some(task) = self.0.take() {740 task.abort();741 }742 }743}744745impl Drop for InputLease {746 fn drop(&mut self) {747 self.registry.release_input(self.remote, self.session_id);748 }749}750751impl Drop for AudioLease {752 fn drop(&mut self) {753 self.registry.release_audio(self.remote, self.session_id);754 }755}756757impl Drop for FeedbackLease {758 fn drop(&mut self) {759 self.registry.release_feedback(self.remote, self.session_id);760 }761}762763764mod tests {765 use super::super::endpoint;766 use super::*;767768 use moq_net::{Broadcast, BroadcastProducer};769 use sigil_protocol::media_moq_broadcast_name;770771 772 async fn forced_idr_recovery_is_one_slot_join_safe_and_rearms_after_ack() {773 let telemetry = Arc::new(MediaV3Telemetry::default());774 let harness = crate::source::EncoderControlTestHarness::new();775 let mut coordinator =776 ForcedIdrCoordinator::new(Some(harness.control.clone()), Arc::clone(&telemetry));777778 assert_eq!(779 coordinator.request(KeyframeRequestReasonV3::Join),780 ForcedIdrDisposition::JoinReplay781 );782 assert_eq!(harness.requested_force_keyframe_revision(), None);783784 let requested_revision = match coordinator.request(KeyframeRequestReasonV3::DecoderReset) {785 ForcedIdrDisposition::Requested { revision } => revision,786 disposition => panic!("unexpected forced-IDR disposition: {disposition:?}"),787 };788 assert_eq!(789 harness.requested_force_keyframe_revision(),790 Some(requested_revision)791 );792 assert_eq!(793 coordinator.request(KeyframeRequestReasonV3::TransportGap),794 ForcedIdrDisposition::Coalesced {795 revision: requested_revision796 }797 );798 assert_eq!(telemetry.encoder_force_requests.load(Ordering::Relaxed), 1);799800 let newer_revision = harness.control.request_force_keyframe().unwrap();801 harness.status.send_modify(|status| {802 status.requested_force_keyframe_revision = Some(newer_revision);803 status.acknowledged_force_keyframe_revision = Some(newer_revision);804 });805 let acknowledgement = coordinator.acknowledgements.join_next().await;806 coordinator.complete(acknowledgement, endpoint(1), "test");807 assert_eq!(coordinator.pending_revision, None);808 assert_eq!(809 telemetry810 .encoder_force_acknowledgements811 .load(Ordering::Relaxed),812 1813 );814 assert_eq!(telemetry.encoder_force_failures.load(Ordering::Relaxed), 0);815816 assert!(matches!(817 coordinator.request(KeyframeRequestReasonV3::DecoderReset),818 ForcedIdrDisposition::Requested { revision } if revision > newer_revision819 ));820 coordinator.abort_and_drain(endpoint(1), "test").await;821822 let fallback_telemetry = Arc::new(MediaV3Telemetry::default());823 let mut fallback = ForcedIdrCoordinator::new(None, fallback_telemetry);824 assert_eq!(825 fallback.request(KeyframeRequestReasonV3::DecoderReset),826 ForcedIdrDisposition::Unavailable827 );828 assert!(fallback.acknowledgements.is_empty());829 }830831 832 fn only_one_remote_can_hold_session() {833 let sessions = Arc::new(SessionRegistry::default());834 assert!(!sessions.has_session());835 let nonce = [7; 16];836 let first = sessions837 .claim(endpoint(1), nonce, InvitationGrants::ALL)838 .unwrap();839 assert!(sessions.has_session());840 assert!(841 sessions842 .claim(endpoint(2), nonce, InvitationGrants::ALL)843 .is_err()844 );845 assert!(sessions.claim_input(endpoint(1), [8; 16]).is_err());846 let input = sessions.claim_input(endpoint(1), nonce).unwrap();847 assert_eq!(input.session_id, first.session_id);848 assert!(sessions.claim_input(endpoint(1), nonce).is_err());849 let audio = sessions.claim_audio(endpoint(1), nonce).unwrap();850 assert!(sessions.claim_audio(endpoint(1), nonce).is_err());851 drop(input);852 let draining_input = sessions.claim_input(endpoint(1), nonce).unwrap();853 drop(first);854 assert!(855 sessions856 .claim(endpoint(2), nonce, InvitationGrants::ALL)857 .is_err()858 );859 drop(draining_input);860 assert!(861 sessions862 .claim(endpoint(2), nonce, InvitationGrants::ALL)863 .is_err()864 );865 drop(audio);866 assert!(!sessions.has_session());867 assert!(868 sessions869 .claim(endpoint(2), nonce, InvitationGrants::ALL)870 .is_ok()871 );872 }873874 fn test_moq_broadcast() -> (BroadcastProducer, BroadcastConsumer) {875 let producer = Broadcast::new().produce();876 let consumer = producer.consume();877 (producer, consumer)878 }879880 881 fn moq_attachment_requires_exact_active_control_remote_and_is_single_use() {882 let sessions = Arc::new(SessionRegistry::default());883 assert!(sessions.claim_moq(endpoint(1)).is_err());884 let lease = sessions885 .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)886 .unwrap();887 let (_producer, consumer) = test_moq_broadcast();888 let _wait = sessions889 .expect_moq(890 endpoint(1),891 lease.session_id,892 media_moq_broadcast_name(lease.session_id).unwrap(),893 consumer,894 )895 .unwrap();896897 // A wrong peer cannot consume the exact pending attachment.898 assert!(sessions.claim_moq(endpoint(2)).is_err());899 let attachment = sessions.claim_moq(endpoint(1)).unwrap();900 assert_eq!(attachment.session_id, lease.session_id);901 assert_eq!(902 attachment.broadcast_name,903 media_moq_broadcast_name(lease.session_id).unwrap()904 );905 // The pending token was atomically consumed before the MoQ handshake.906 assert!(sessions.claim_moq(endpoint(1)).is_err());907 }908909 910 fn competing_moq_connections_cannot_both_claim_one_control_attachment() {911 let sessions = Arc::new(SessionRegistry::default());912 let lease = sessions913 .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)914 .unwrap();915 let (_producer, consumer) = test_moq_broadcast();916 let _wait = sessions917 .expect_moq(918 endpoint(1),919 lease.session_id,920 media_moq_broadcast_name(lease.session_id).unwrap(),921 consumer,922 )923 .unwrap();924 let barrier = Arc::new(std::sync::Barrier::new(3));925 let contenders = (0..2)926 .map(|_| {927 let sessions = Arc::clone(&sessions);928 let barrier = Arc::clone(&barrier);929 std::thread::spawn(move || {930 barrier.wait();931 sessions.claim_moq(endpoint(1)).is_ok()932 })933 })934 .collect::<Vec<_>>();935 barrier.wait();936 let claimed = contenders937 .into_iter()938 .map(|thread| usize::from(thread.join().unwrap()))939 .sum::<usize>();940 assert_eq!(claimed, 1);941 }942943 944 async fn releasing_control_clears_an_unclaimed_moq_attachment() {945 let sessions = Arc::new(SessionRegistry::default());946 let lease = sessions947 .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)948 .unwrap();949 let (_producer, consumer) = test_moq_broadcast();950 let wait = sessions951 .expect_moq(952 endpoint(1),953 lease.session_id,954 media_moq_broadcast_name(lease.session_id).unwrap(),955 consumer,956 )957 .unwrap();958 drop(lease);959 assert!(sessions.claim_moq(endpoint(1)).is_err());960 assert!(wait.attached.await.is_err());961 assert!(wait.closed.await.is_err());962 }963964 965 fn feedback_attaches_only_to_exact_active_view_session() {966 let sessions = Arc::new(SessionRegistry::default());967 let nonce = [9; 16];968 assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());969970 let no_view = sessions971 .claim(endpoint(1), nonce, InvitationGrants::GAMEPAD)972 .unwrap();973 assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());974 drop(no_view);975976 let media = sessions977 .claim(endpoint(1), nonce, InvitationGrants::VIEW)978 .unwrap();979 assert!(sessions.claim_feedback(endpoint(2), nonce).is_err());980 assert!(sessions.claim_feedback(endpoint(1), [8; 16]).is_err());981 let feedback = sessions.claim_feedback(endpoint(1), nonce).unwrap();982 assert_eq!(feedback.session_id, media.session_id);983 assert!(feedback.encoder_control.is_none());984 assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());985986 drop(media);987 assert!(988 sessions989 .claim(endpoint(2), nonce, InvitationGrants::VIEW)990 .is_err(),991 "feedback teardown must keep the draining session isolated"992 );993 drop(feedback);994 assert!(995 sessions996 .claim(endpoint(2), nonce, InvitationGrants::VIEW)997 .is_ok()998 );999 }10001001 1002 fn adaptive_encoder_proposals_are_bound_to_the_exact_active_generation() {1003 let sessions = Arc::new(SessionRegistry::default());1004 let remote = endpoint(1);1005 let media = sessions1006 .claim(remote, [3; 16], InvitationGrants::VIEW)1007 .unwrap();1008 let harness = crate::source::EncoderControlTestHarness::new();10091010 assert!(1011 sessions1012 .install_encoder_control(1013 endpoint(2),1014 media.session_id,1015 Some(harness.control.clone())1016 )1017 .is_err()1018 );1019 sessions1020 .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))1021 .unwrap();1022 let feedback = sessions.claim_feedback(remote, [3; 16]).unwrap();1023 assert!(feedback.encoder_control.is_some());1024 let proposal = sessions1025 .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)1026 .unwrap()1027 .unwrap();1028 assert_eq!(proposal.target_kbps, 8_000);1029 assert!(proposal.force_keyframe_revision > Some(proposal.bitrate_revision));10301031 let old_session_id = media.session_id;1032 drop(media);1033 assert!(1034 sessions1035 .propose_adaptive_encoder_update(remote, old_session_id, 7_000, false)1036 .is_err(),1037 "a draining generation must not issue another encoder proposal"1038 );1039 drop(feedback);1040 }10411042 1043 fn pending_handshakes_are_bounded() {1044 let sessions = SessionRegistry::default();1045 let permits: Vec<_> = (0..MAX_PENDING_HANDSHAKES)1046 .map(|_| sessions.pending_handshakes.try_acquire().unwrap())1047 .collect();1048 assert!(sessions.pending_handshakes.try_acquire().is_err());1049 drop(permits);1050 assert!(sessions.pending_handshakes.try_acquire().is_ok());1051 }10521053 1054 fn session_substreams_inherit_the_exact_enrollment_grant() {1055 let sessions = Arc::new(SessionRegistry::default());1056 let grants = InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD);1057 let media = sessions.claim(endpoint(1), [3; 16], grants).unwrap();1058 let input = sessions.claim_input(endpoint(1), [3; 16]).unwrap();1059 let audio = sessions.claim_audio(endpoint(1), [3; 16]).unwrap();1060 assert_eq!(input.grants, grants);1061 assert_eq!(audio.grants, grants);1062 drop(input);1063 drop(audio);1064 drop(media);1065 }10661067 1068 fn audio_claim_requires_the_active_remote_and_nonce() {1069 let sessions = Arc::new(SessionRegistry::default());1070 let media = sessions1071 .claim(endpoint(1), [9; 16], InvitationGrants::ALL)1072 .unwrap();1073 assert!(sessions.claim_audio(endpoint(2), [9; 16]).is_err());1074 assert!(sessions.claim_audio(endpoint(1), [8; 16]).is_err());1075 let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();1076 drop(media);1077 assert!(1078 sessions1079 .claim(endpoint(2), [0; 16], InvitationGrants::ALL)1080 .is_err()1081 );1082 drop(audio);1083 assert!(1084 sessions1085 .claim(endpoint(2), [0; 16], InvitationGrants::ALL)1086 .is_ok()1087 );1088 }10891090 1091 fn media_and_audio_leases_share_one_session_clock() {1092 let sessions = Arc::new(SessionRegistry::default());1093 let media = sessions1094 .claim(endpoint(1), [9; 16], InvitationGrants::ALL)1095 .unwrap();1096 let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();1097 assert_eq!(media.session_clock, audio.session_clock);1098 }1099}