crates/sigil-host/src/server/session.rs - fix/drop-hero-notarization-note

fix/drop-hero-notarization-note
goq / crates / sigil-host / src / server / session.rs
session.rs 115.6K wrap lines
1use std::collections::{HashMap, VecDeque};2use std::sync::atomic::{AtomicU64, Ordering};3use std::sync::{Arc, Mutex};4use std::time::{Duration, Instant};5use std::time::{SystemTime, UNIX_EPOCH};67use anyhow::{Context, Result, bail, ensure};8use iroh::EndpointId;9use iroh::endpoint::Connection;10use moq_net::BroadcastConsumer;11use sigil_protocol::{12    ControllerSlot, FocusCommandActionV2, FocusCommandV2, FocusStateV2, FocusTransitionReasonV2,13    InvitationGrants, KeyframeRequestReasonV3, MediaGenerationDescriptorV2, SessionSnapshotV2,14    SignedSubscriptionCapability, SubscriptionTracks, ViewerPresenceId, ViewerPresenceV2,15    media_generation_moq_broadcast_name, media_moq_broadcast_name,16};17use tracing::{debug, info, warn};1819use super::focus::{FocusArbiter, FocusCandidate, FocusNeutralization};20use super::{ENCODER_CONTROL_COMMIT_TIMEOUT, VideoDimensions};21use crate::authorization::{AuthorizationMutation, AuthorizedViewer};22use crate::clock::SessionClock;23use crate::source::EncoderControl;2425const MAX_PENDING_HANDSHAKES: usize = 4;26// The hard eight-viewer ceiling gives each admitted viewer four independent27// handshake slots (control, MoQ, input, and feedback). The semaphore therefore28// grows linearly with configured concurrency and never with connection churn.29const MAX_TRACKED_ADMISSION_PEERS: usize = crate::config::MAX_VIEWERS as usize * 4;30const ADMISSION_RATE_WINDOW: Duration = Duration::from_secs(10);31const MAX_ADMISSIONS_PER_WINDOW: usize = 12;32const ATTACHMENT_RATE_WINDOW: Duration = Duration::from_secs(5);33const MAX_ATTACHMENTS_PER_WINDOW: usize = 4;34const KEYFRAME_RATE_WINDOW: Duration = Duration::from_secs(2);35const MAX_KEYFRAME_REQUESTS_PER_WINDOW: usize = 8;36const FEEDBACK_CLAIM_RATE_WINDOW: Duration = Duration::from_secs(10);37const MAX_FEEDBACK_CLAIMS_PER_WINDOW: usize = 4;3839#[derive(Debug)]40pub struct SessionRegistry {41    active: Mutex<Option<ActiveSession>>,42    pending_moq: Mutex<HashMap<(EndpointId, u64), PendingMoqAttachment>>,43    max_viewers: usize,44    next_session_id: AtomicU64,45    authorization_committed_revision: AtomicU64,46    admission_rates: Mutex<HashMap<EndpointId, RateWindow>>,47    v2_state: Mutex<V2SessionState>,48    pub(super) session_changed: tokio::sync::Notify,49    pub(super) pending_handshakes: tokio::sync::Semaphore,50}5152struct PendingMoqAttachment {53    remote: EndpointId,54    session_id: u64,55    broadcast_name: String,56    broadcast: BroadcastConsumer,57    attached: tokio::sync::oneshot::Sender<()>,58    closed: tokio::sync::oneshot::Sender<()>,59    telemetry: Arc<MediaV3Telemetry>,60    subscription_capability: Option<SignedSubscriptionCapability>,61    expected_host: Option<[u8; 32]>,62}6364impl std::fmt::Debug for PendingMoqAttachment {65    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {66        formatter67            .debug_struct("PendingMoqAttachment")68            .field("remote", &self.remote)69            .field("session_id", &self.session_id)70            .field("broadcast_name", &self.broadcast_name)71            .finish_non_exhaustive()72    }73}7475pub(super) struct ClaimedMoqAttachment {76    pub(super) session_id: u64,77    pub(super) broadcast_name: String,78    pub(super) broadcast: BroadcastConsumer,79    pub(super) attached: tokio::sync::oneshot::Sender<()>,80    pub(super) closed: tokio::sync::oneshot::Sender<()>,81    pub(super) telemetry: Arc<MediaV3Telemetry>,82}8384pub(super) struct MoqAttachmentWait {85    pub(super) attached: tokio::sync::oneshot::Receiver<()>,86    pub(super) closed: tokio::sync::oneshot::Receiver<()>,87}8889#[derive(Clone, Debug)]90struct ActiveSession {91    remote: EndpointId,92    session_id: u64,93    nonce: [u8; 16],94    session_clock: SessionClock,95    media_generation_id: u64,96    media_broadcast_name: String,97    grants: InvitationGrants,98    viewer_handle: Option<String>,99    authorization_revision: u64,100    authorization_committed_revision: u64,101    media_active: bool,102    input_claimed: bool,103    audio_claimed: bool,104    feedback_claimed: bool,105    media_v3_telemetry: Arc<MediaV3Telemetry>,106    encoder_control: Option<EncoderControl>,107    mode: SessionMode,108}109110#[derive(Clone, Copy, Debug, PartialEq, Eq)]111enum SessionMode {112    LegacyExclusive,113    V2Single,114}115116#[derive(Debug)]117struct V2SessionState {118    viewers: HashMap<EndpointId, V2ViewerSession>,119    revision: u64,120    focus: FocusArbiter,121    media: Option<MediaGenerationDescriptorV2>,122    live_control_leases: usize,123}124125impl V2SessionState {126    fn new(configured_owner: Option<ViewerPresenceId>) -> Self {127        Self {128            viewers: HashMap::new(),129            revision: 0,130            focus: FocusArbiter::new(configured_owner),131            media: None,132            live_control_leases: 0,133        }134    }135}136137#[derive(Debug)]138struct V2ViewerSession {139    session: ActiveSession,140    presence_id: ViewerPresenceId,141    authorization_neutralizing: bool,142    snapshots: tokio::sync::watch::Sender<Option<SessionSnapshotV2>>,143    rates: ViewerRateLimits,144}145146#[derive(Debug, Default)]147struct ViewerRateLimits {148    attachments: RateWindow,149    keyframes: RateWindow,150    feedback_claims: RateWindow,151}152153#[derive(Debug, Default)]154struct RateWindow {155    events: VecDeque<Instant>,156    last_seen: Option<Instant>,157}158159impl RateWindow {160    fn check(161        &mut self,162        now: Instant,163        window: Duration,164        maximum: usize,165        label: &'static str,166    ) -> Result<()> {167        while self168            .events169            .front()170            .is_some_and(|event| now.saturating_duration_since(*event) >= window)171        {172            self.events.pop_front();173        }174        self.last_seen = Some(now);175        ensure!(self.events.len() < maximum, "{label} rate limit exceeded");176        self.events.push_back(now);177        Ok(())178    }179180    fn expired(&self, now: Instant, window: Duration) -> bool {181        self.last_seen182            .is_none_or(|seen| now.saturating_duration_since(seen) >= window)183    }184}185186#[derive(Clone, Copy, Debug, PartialEq, Eq)]187pub(crate) struct SessionRuntimeStatus {188    pub(crate) mode: &'static str,189    pub(crate) active_viewers: usize,190    pub(crate) media_generation_id: Option<u64>,191    pub(crate) roster_revision: Option<u64>,192    pub(crate) focus_occupied: Option<bool>,193    pub(crate) configured_capacity: usize,194    pub(crate) authorization_revision: u64,195}196197#[derive(Clone, Debug)]198pub(crate) struct FocusCommandEffect {199    pub(crate) snapshot: SessionSnapshotV2,200    pub(crate) neutralization: Option<FocusNeutralization>,201}202203#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]204pub struct AuthorizationSessionEffect {205    pub disconnected: bool,206    pub neutralize_input: bool,207    pub focus_transition_id: Option<u64>,208}209210#[derive(Debug, Default)]211pub(super) struct MediaV3Telemetry {212    pub(super) scheduler_cancellations: AtomicU64,213    pub(super) send_failures: AtomicU64,214    selected_path_rtt_micros: AtomicU64,215    selected_path_lost_packets: AtomicU64,216    selected_path_congestion_events: AtomicU64,217    keyframe_control_requests: AtomicU64,218    encoder_force_requests: AtomicU64,219    encoder_force_acknowledgements: AtomicU64,220    encoder_force_failures: AtomicU64,221    last_encoder_force_ack_micros: AtomicU64,222}223224#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]225pub(super) struct MediaV3TelemetrySnapshot {226    pub(super) scheduler_cancellations: u64,227    pub(super) send_failures: u64,228    pub(super) selected_path_rtt_micros: u64,229    pub(super) selected_path_lost_packets: u64,230    pub(super) selected_path_congestion_events: u64,231}232233#[derive(Clone, Debug)]234pub(super) struct AdaptiveEncoderProposal {235    pub(super) control: EncoderControl,236    pub(super) target_kbps: u32,237    pub(super) bitrate_revision: u64,238    pub(super) force_keyframe_revision: Option<u64>,239}240241#[derive(Clone, Debug)]242pub(super) struct ResolutionEncoderProposal {243    pub(super) control: EncoderControl,244    pub(super) target: VideoDimensions,245    pub(super) revision: u64,246}247248#[derive(Debug, PartialEq, Eq)]249pub(super) enum ForcedIdrDisposition {250    JoinReplay,251    Unavailable,252    Requested { revision: u64 },253    Coalesced { revision: u64 },254    Failed { error: String },255}256257pub(super) struct ForcedIdrAcknowledgement {258    requested_revision: u64,259    elapsed: Duration,260    result: Result<crate::source::EncoderControlStatus>,261}262263pub(super) struct ForcedIdrCoordinator {264    control: Option<EncoderControl>,265    pub(super) pending_revision: Option<u64>,266    pub(super) acknowledgements: tokio::task::JoinSet<ForcedIdrAcknowledgement>,267    telemetry: Arc<MediaV3Telemetry>,268}269270impl ForcedIdrCoordinator {271    pub(super) fn new(control: Option<EncoderControl>, telemetry: Arc<MediaV3Telemetry>) -> Self {272        Self {273            control,274            pending_revision: None,275            acknowledgements: tokio::task::JoinSet::new(),276            telemetry,277        }278    }279280    pub(super) fn request(&mut self, reason: KeyframeRequestReasonV3) -> ForcedIdrDisposition {281        self.telemetry282            .keyframe_control_requests283            .fetch_add(1, Ordering::Relaxed);284        if reason == KeyframeRequestReasonV3::Join {285            return ForcedIdrDisposition::JoinReplay;286        }287        if let Some(revision) = self.pending_revision {288            return ForcedIdrDisposition::Coalesced { revision };289        }290        let Some(control) = self.control.clone() else {291            return ForcedIdrDisposition::Unavailable;292        };293        let revision = match control.request_force_keyframe() {294            Ok(revision) => revision,295            Err(error) => {296                self.telemetry297                    .encoder_force_failures298                    .fetch_add(1, Ordering::Relaxed);299                return ForcedIdrDisposition::Failed {300                    error: error.to_string(),301                };302            }303        };304        self.pending_revision = Some(revision);305        self.telemetry306            .encoder_force_requests307            .fetch_add(1, Ordering::Relaxed);308        self.acknowledgements.spawn(async move {309            let started_at = Instant::now();310            let result = control311                .wait_for_recovery_keyframe_acknowledged(revision, ENCODER_CONTROL_COMMIT_TIMEOUT)312                .await;313            ForcedIdrAcknowledgement {314                requested_revision: revision,315                elapsed: started_at.elapsed(),316                result,317            }318        });319        ForcedIdrDisposition::Requested { revision }320    }321322    pub(super) fn complete(323        &mut self,324        result: Option<Result<ForcedIdrAcknowledgement, tokio::task::JoinError>>,325        remote: EndpointId,326        transport: &'static str,327    ) {328        let pending_revision = self.pending_revision.take();329        match result {330            Some(Ok(acknowledgement)) => match acknowledgement.result {331                Ok(status) => {332                    let elapsed_micros =333                        u64::try_from(acknowledgement.elapsed.as_micros()).unwrap_or(u64::MAX);334                    self.telemetry335                        .encoder_force_acknowledgements336                        .fetch_add(1, Ordering::Relaxed);337                    self.telemetry338                        .last_encoder_force_ack_micros339                        .store(elapsed_micros, Ordering::Relaxed);340                    debug!(341                        %remote,342                        transport,343                        requested_revision = acknowledgement.requested_revision,344                        acknowledged_revision = ?status.acknowledged_force_keyframe_revision,345                        elapsed_micros,346                        "forced-IDR recovery acknowledged"347                    );348                }349                Err(error) => {350                    self.telemetry351                        .encoder_force_failures352                        .fetch_add(1, Ordering::Relaxed);353                    warn!(354                        %remote,355                        transport,356                        requested_revision = acknowledgement.requested_revision,357                        %error,358                        "forced-IDR recovery was not acknowledged; retaining natural-IDR fallback"359                    );360                }361            },362            Some(Err(error)) => {363                self.telemetry364                    .encoder_force_failures365                    .fetch_add(1, Ordering::Relaxed);366                warn!(367                    %remote,368                    transport,369                    ?pending_revision,370                    %error,371                    "forced-IDR acknowledgement task failed; retaining natural-IDR fallback"372                );373            }374            None => {375                self.telemetry376                    .encoder_force_failures377                    .fetch_add(1, Ordering::Relaxed);378                warn!(379                    %remote,380                    transport,381                    ?pending_revision,382                    "forced-IDR acknowledgement task ended without a result"383                );384            }385        }386    }387388    pub(super) async fn abort_and_drain(&mut self, remote: EndpointId, transport: &'static str) {389        self.pending_revision = None;390        self.acknowledgements.abort_all();391        while self.acknowledgements.join_next().await.is_some() {}392        debug!(393            %remote,394            transport,395            keyframe_control_requests = self396                .telemetry397                .keyframe_control_requests398                .load(Ordering::Relaxed),399            encoder_force_requests = self400                .telemetry401                .encoder_force_requests402                .load(Ordering::Relaxed),403            encoder_force_acknowledgements = self404                .telemetry405                .encoder_force_acknowledgements406                .load(Ordering::Relaxed),407            encoder_force_failures = self408                .telemetry409                .encoder_force_failures410                .load(Ordering::Relaxed),411            last_encoder_force_ack_micros = self412                .telemetry413                .last_encoder_force_ack_micros414                .load(Ordering::Relaxed),415            "forced-IDR recovery session summary"416        );417    }418}419420impl MediaV3Telemetry {421    pub(super) fn snapshot(&self) -> MediaV3TelemetrySnapshot {422        MediaV3TelemetrySnapshot {423            scheduler_cancellations: self.scheduler_cancellations.load(Ordering::Relaxed),424            send_failures: self.send_failures.load(Ordering::Relaxed),425            selected_path_rtt_micros: self.selected_path_rtt_micros.load(Ordering::Relaxed),426            selected_path_lost_packets: self.selected_path_lost_packets.load(Ordering::Relaxed),427            selected_path_congestion_events: self428                .selected_path_congestion_events429                .load(Ordering::Relaxed),430        }431    }432433    pub(super) fn record_selected_path(&self, connection: &Connection) {434        let paths = connection.paths();435        let Some(path) = paths.iter().find(|path| path.is_selected()) else {436            return;437        };438        let stats = path.stats();439        self.selected_path_rtt_micros.store(440            u64::try_from(stats.rtt.as_micros()).unwrap_or(u64::MAX),441            Ordering::Relaxed,442        );443        self.selected_path_lost_packets444            .store(stats.lost_packets, Ordering::Relaxed);445        self.selected_path_congestion_events446            .store(stats.congestion_events, Ordering::Relaxed);447    }448}449450impl Default for SessionRegistry {451    fn default() -> Self {452        Self::new(crate::config::DEFAULT_MAX_VIEWERS)453    }454}455456impl SessionRegistry {457    pub fn new(max_viewers: u8) -> Self {458        Self::new_with_focus_owner(max_viewers, None)459    }460461    pub fn new_with_focus_owner(max_viewers: u8, focus_owner: Option<String>) -> Self {462        assert!(463            (1..=crate::config::MAX_VIEWERS).contains(&max_viewers),464            "validated max_viewers must be between 1 and {}",465            crate::config::MAX_VIEWERS466        );467        let focus_owner = focus_owner468            .map(ViewerPresenceId::new)469            .transpose()470            .expect("validated focus_owner must be an opaque viewer handle");471        Self {472            active: Mutex::new(None),473            pending_moq: Mutex::new(HashMap::with_capacity(crate::config::MAX_VIEWERS.into())),474            max_viewers: usize::from(max_viewers),475            next_session_id: AtomicU64::new(0),476            authorization_committed_revision: AtomicU64::new(1),477            admission_rates: Mutex::new(HashMap::with_capacity(MAX_TRACKED_ADMISSION_PEERS)),478            v2_state: Mutex::new(V2SessionState::new(focus_owner)),479            session_changed: tokio::sync::Notify::new(),480            pending_handshakes: tokio::sync::Semaphore::new(481                MAX_PENDING_HANDSHAKES * usize::from(max_viewers),482            ),483        }484    }485486    #[allow(dead_code)]487    pub fn has_session(&self) -> bool {488        if self489            .active490            .lock()491            .expect("session registry poisoned")492            .is_some()493        {494            return true;495        }496        !self497            .v2_state498            .lock()499            .expect("v2 session state poisoned")500            .viewers501            .is_empty()502    }503504    pub(crate) fn runtime_status(&self) -> SessionRuntimeStatus {505        if let Some(active) = self506            .active507            .lock()508            .expect("session registry poisoned")509            .as_ref()510            .filter(|session| session.media_active)511        {512            return SessionRuntimeStatus {513                mode: "legacy_exclusive",514                active_viewers: 1,515                media_generation_id: Some(active.media_generation_id),516                roster_revision: None,517                focus_occupied: None,518                configured_capacity: self.max_viewers,519                authorization_revision: self520                    .authorization_committed_revision521                    .load(Ordering::SeqCst),522            };523        }524        let state = self.v2_state.lock().expect("v2 session state poisoned");525        SessionRuntimeStatus {526            mode: if state.viewers.is_empty() {527                "inactive"528            } else {529                "moq_multi_viewer"530            },531            active_viewers: state.viewers.len(),532            media_generation_id: state.media.as_ref().map(|media| media.generation_id),533            roster_revision: (!state.viewers.is_empty()).then_some(state.revision),534            focus_occupied: (!state.viewers.is_empty())535                .then_some(matches!(state.focus.state(), FocusStateV2::Held { .. })),536            configured_capacity: self.max_viewers,537            authorization_revision: self.authorization_committed_revision.load(Ordering::SeqCst),538        }539    }540541    fn check_admission_rate(&self, remote: EndpointId, now: Instant) -> Result<()> {542        let mut rates = self543            .admission_rates544            .lock()545            .expect("viewer admission rate state poisoned");546        rates.retain(|_, rate| !rate.expired(now, ADMISSION_RATE_WINDOW));547        ensure!(548            rates.contains_key(&remote) || rates.len() < MAX_TRACKED_ADMISSION_PEERS,549            "viewer admission rate table is full"550        );551        rates.entry(remote).or_default().check(552            now,553            ADMISSION_RATE_WINDOW,554            MAX_ADMISSIONS_PER_WINDOW,555            "viewer reconnect",556        )557    }558559    pub(super) fn claim(560        self: &Arc<Self>,561        remote: EndpointId,562        nonce: [u8; 16],563        grants: InvitationGrants,564    ) -> Result<SessionLease> {565        let mut active = self.active.lock().expect("session registry poisoned");566        if let Some(current) = active.as_ref() {567            bail!("host already has active client {}", current.remote);568        }569        ensure!(570            self.v2_state571                .lock()572                .expect("v2 session state poisoned")573                .live_control_leases574                == 0,575            "host already has an active multi-viewer generation"576        );577        let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;578        let session_clock = SessionClock::start();579        let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());580        *active = Some(ActiveSession {581            remote,582            session_id,583            nonce,584            session_clock,585            media_generation_id: session_id,586            media_broadcast_name: media_moq_broadcast_name(session_id)?,587            grants,588            viewer_handle: None,589            authorization_revision: 1,590            authorization_committed_revision: 1,591            media_active: true,592            input_claimed: false,593            audio_claimed: false,594            feedback_claimed: false,595            media_v3_telemetry: Arc::clone(&media_v3_telemetry),596            encoder_control: None,597            mode: SessionMode::LegacyExclusive,598        });599        Ok(SessionLease {600            registry: Arc::clone(self),601            remote,602            session_id,603            session_clock,604            media_v3_telemetry,605        })606    }607608    #[allow(dead_code)]609    pub(super) fn claim_v2(610        self: &Arc<Self>,611        remote: EndpointId,612        nonce: [u8; 16],613        grants: InvitationGrants,614    ) -> Result<V2SessionLease> {615        self.claim_v2_authorized(616            remote,617            nonce,618            AuthorizedViewer {619                handle: "viewer-0000000000000000".to_owned(),620                grants,621                authorization_revision: 1,622                committed_revision: 1,623            },624        )625    }626627    pub(crate) fn claim_v2_authorized(628        self: &Arc<Self>,629        remote: EndpointId,630        nonce: [u8; 16],631        authorized: AuthorizedViewer,632    ) -> Result<V2SessionLease> {633        authorized.grants.validate()?;634        ensure!(635            authorized.grants.contains(InvitationGrants::VIEW),636            "viewer lacks view permission"637        );638        ensure!(639            authorized.authorization_revision != 0 && authorized.committed_revision != 0,640            "viewer authorization revision is invalid"641        );642        self.check_admission_rate(remote, Instant::now())?;643        self.authorization_committed_revision644            .fetch_max(authorized.committed_revision, Ordering::SeqCst);645        ensure!(646            self.authorization_committed_revision.load(Ordering::SeqCst)647                == authorized.committed_revision,648            "viewer admission used a stale authorization revision"649        );650        // Hold the legacy registry lock across the whole v2 admission. `claim`651        // takes `active` and then `v2_state`; releasing `active` here first652        // would let a legacy client and a v2 viewer both pass their mutual653        // exclusion check and be admitted together.654        // Hold the legacy registry lock across the whole v2 admission. `claim`655        // takes `active` and then `v2_state`; releasing `active` here first656        // would let a legacy client and a v2 viewer both pass their mutual657        // exclusion check and be admitted together.658        let active = self.active.lock().expect("session registry poisoned");659        ensure!(active.is_none(), "host already has an active legacy client");660        let mut state = self.v2_state.lock().expect("v2 session state poisoned");661        let replacing = state.viewers.contains_key(&remote);662        ensure!(663            replacing || state.viewers.len() < self.max_viewers,664            "multi-viewer capacity of {} is full",665            self.max_viewers666        );667        let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;668        let session_clock = SessionClock::start();669        let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());670        let presence_id = ViewerPresenceId::new(authorized.handle.clone())?;671        ensure!(672            state673                .viewers674                .iter()675                .all(|(peer, viewer)| { *peer == remote || viewer.presence_id != presence_id }),676            "viewer presence id is already active for another peer"677        );678        let replaced = state.viewers.remove(&remote);679        let replaced_candidate = replaced.as_ref().map(|viewer| FocusCandidate {680            presence_id: viewer.presence_id.clone(),681            session_id: viewer.session.session_id,682        });683        let replacement_transition = replaced_candidate684            .as_ref()685            .map(|candidate| {686                state687                    .focus688                    .begin_invalidation(candidate, FocusTransitionReasonV2::Replaced)689            })690            .transpose()?691            .and_then(|mutation| mutation.neutralization);692        if let Some(candidate) = &replaced_candidate {693            state.focus.retire_candidate(candidate);694        }695        let media = state.media.clone().unwrap_or(MediaGenerationDescriptorV2 {696            generation_id: session_id,697            broadcast_name: media_generation_moq_broadcast_name(session_id)?,698        });699        let (snapshots, _receiver) = tokio::sync::watch::channel(None);700        let session = ActiveSession {701            remote,702            session_id,703            nonce,704            session_clock,705            media_generation_id: media.generation_id,706            media_broadcast_name: media.broadcast_name,707            grants: authorized.grants,708            viewer_handle: Some(authorized.handle.clone()),709            authorization_revision: authorized.authorization_revision,710            authorization_committed_revision: authorized.committed_revision,711            media_active: true,712            input_claimed: false,713            audio_claimed: false,714            feedback_claimed: false,715            media_v3_telemetry: Arc::clone(&media_v3_telemetry),716            encoder_control: None,717            mode: SessionMode::V2Single,718        };719        state.viewers.insert(720            remote,721            V2ViewerSession {722                session,723                presence_id,724                authorization_neutralizing: false,725                snapshots,726                rates: ViewerRateLimits::default(),727            },728        );729        state.live_control_leases = state730            .live_control_leases731            .checked_add(1)732            .context("v2 control lease count overflowed")?;733        advance_v2_revision(&mut state)?;734        publish_v2_snapshots(&state)?;735        let snapshot = v2_snapshot_for(&state, remote)?;736        let active_viewers = state.viewers.len();737        let viewer_handle = authorized.handle.clone();738        let replaced_snapshots = replaced.map(|viewer| viewer.snapshots);739        let lease = V2SessionLease {740            registry: Arc::clone(self),741            remote,742            session_id,743            session_clock,744            media_v3_telemetry,745            initial_snapshot: snapshot,746            authorization_revision: authorized.authorization_revision,747            authorization_committed_revision: authorized.committed_revision,748            replaced_snapshots,749            replacement_transition,750        };751        drop(state);752        drop(active);753        self.session_changed.notify_waiters();754        info!(755            viewer_handle,756            session_id,757            active_viewers,758            reason = if replacing {759                "replacement"760            } else {761                "connected"762            },763            "multi-viewer session membership changed"764        );765        Ok(lease)766    }767768    pub(super) fn bind_v2_generation(769        &self,770        remote: EndpointId,771        session_id: u64,772        generation_id: u64,773        session_clock: SessionClock,774        _telemetry: Arc<MediaV3Telemetry>,775        encoder_control: Option<EncoderControl>,776    ) -> Result<SessionSnapshotV2> {777        let mut state = self.v2_state.lock().expect("v2 session state poisoned");778        let descriptor = MediaGenerationDescriptorV2 {779            generation_id,780            broadcast_name: media_generation_moq_broadcast_name(generation_id)?,781        };782        if let Some(current) = &state.media {783            ensure!(784                current == &descriptor,785                "viewer attempted to join a different active media generation"786            );787        } else {788            state.media = Some(descriptor.clone());789        }790        let viewer = state791            .viewers792            .get_mut(&remote)793            .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)794            .context("generation binding does not match the active v2 viewer")?;795        ensure!(generation_id != 0, "media generation id must be non-zero");796        viewer.session.session_clock = session_clock;797        viewer.session.media_generation_id = generation_id;798        viewer.session.media_broadcast_name = descriptor.broadcast_name;799        // Keep transport telemetry viewer-scoped. Producer telemetry belongs800        // to the shared generation and is merged only by its adaptive owner.801        viewer.session.encoder_control = encoder_control;802        publish_v2_snapshots(&state)?;803        let snapshot = v2_snapshot_for(&state, remote)?;804        Ok(snapshot)805    }806807    pub(super) fn claim_input(808        self: &Arc<Self>,809        remote: EndpointId,810        nonce: [u8; 16],811    ) -> Result<InputLease> {812        let mut active = self.active.lock().expect("session registry poisoned");813        let session = active814            .as_mut()815            .filter(|session| {816                session.mode == SessionMode::LegacyExclusive817                    && session.media_active818                    && session.remote == remote819                    && session.nonce == nonce820            })821            .context("input connection does not match the active media session")?;822        ensure!(823            !session.input_claimed,824            "active client already has an input stream"825        );826        session.input_claimed = true;827        Ok(InputLease {828            registry: Arc::clone(self),829            remote,830            session_id: session.session_id,831            grants: session.grants,832            authorization_revision: session.authorization_revision,833        })834    }835836    pub(super) fn subscribe_v2_snapshots(837        &self,838        remote: EndpointId,839        session_id: u64,840    ) -> Result<tokio::sync::watch::Receiver<Option<SessionSnapshotV2>>> {841        let state = self.v2_state.lock().expect("v2 session state poisoned");842        let viewer = state843            .viewers844            .get(&remote)845            .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)846            .context("snapshot subscription does not match the active v2 session")?;847        Ok(viewer.snapshots.subscribe())848    }849850    pub(super) fn v2_revision(&self, remote: EndpointId, session_id: u64) -> Result<u64> {851        let state = self.v2_state.lock().expect("v2 session state poisoned");852        state853            .viewers854            .get(&remote)855            .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)856            .context("revision request does not match the active v2 session")?;857        Ok(state.revision)858    }859860    pub(crate) fn apply_focus_command(861        &self,862        remote: EndpointId,863        session_id: u64,864        command: &FocusCommandV2,865    ) -> Result<FocusCommandEffect> {866        command.validate()?;867        let mut state = self.v2_state.lock().expect("v2 session state poisoned");868        let (presence_id, grants, authorization_neutralizing) = state869            .viewers870            .get(&remote)871            .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)872            .map(|viewer| {873                (874                    viewer.presence_id.clone(),875                    viewer.session.grants,876                    viewer.authorization_neutralizing,877                )878            })879            .context("focus command does not match the active v2 session")?;880        ensure!(881            command.expected_revision == state.revision,882            "focus command expected stale revision {} instead of {}",883            command.expected_revision,884            state.revision885        );886887        let candidate = FocusCandidate {888            presence_id,889            session_id,890        };891        let audit_handle = candidate.presence_id.clone();892        state.focus.check_rate_limit(&candidate, Instant::now())?;893        let mutation = match command.action {894            FocusCommandActionV2::Request => {895                ensure!(896                    !authorization_neutralizing,897                    "input authorization is still neutralizing"898                );899                ensure!(900                    grants.contains(InvitationGrants::POINTER_KEYBOARD)901                        || grants.contains(InvitationGrants::GAMEPAD),902                    "viewer is not input-capable"903                );904                state.focus.request(candidate, Instant::now())?905            }906            FocusCommandActionV2::Approve => state.focus.approve(907                &candidate,908                command909                    .expected_focus_generation910                    .expect("validated approval carries a focus generation"),911                command912                    .expected_proposal_id913                    .expect("validated approval carries a proposal id"),914            )?,915            FocusCommandActionV2::Deny => state.focus.deny(916                &candidate,917                command918                    .expected_focus_generation919                    .expect("validated denial carries a focus generation"),920                command921                    .expected_proposal_id922                    .expect("validated denial carries a proposal id"),923            )?,924            FocusCommandActionV2::Preempt => {925                ensure!(926                    !authorization_neutralizing,927                    "input authorization is still neutralizing"928                );929                ensure!(930                    grants.contains(InvitationGrants::POINTER_KEYBOARD)931                        || grants.contains(InvitationGrants::GAMEPAD),932                    "viewer is not input-capable"933                );934                state.focus.preempt(935                    candidate,936                    command937                        .expected_focus_generation938                        .expect("validated preemption carries a focus generation"),939                )?940            }941            FocusCommandActionV2::Release => {942                let expected_focus_generation = command943                    .expected_focus_generation944                    .expect("validated focus release carries a generation");945                state.focus.release(&candidate, expected_focus_generation)?946            }947        };948        if mutation.changed {949            advance_v2_revision(&mut state)?;950            publish_v2_snapshots(&state)?;951        }952        let snapshot = v2_snapshot_for(&state, remote)?;953        if mutation.changed {954            info!(955                viewer_handle = audit_handle.as_str(),956                session_id,957                action = ?command.action,958                roster_revision = snapshot.revision,959                transition_reason = ?snapshot.transition_reason,960                "slot-0 focus state changed"961            );962        }963        self.session_changed.notify_waiters();964        Ok(FocusCommandEffect {965            snapshot,966            neutralization: mutation.neutralization,967        })968    }969970    pub(super) fn invalidate_v2_focus(971        &self,972        remote: EndpointId,973        session_id: u64,974        reason: FocusTransitionReasonV2,975    ) -> Result<Option<FocusNeutralization>> {976        let mut state = self.v2_state.lock().expect("v2 session state poisoned");977        let Some(presence_id) = state978            .viewers979            .get(&remote)980            .filter(|viewer| viewer.session.session_id == session_id)981            .map(|viewer| viewer.presence_id.clone())982        else {983            return Ok(None);984        };985        let mutation = state.focus.begin_invalidation(986            &FocusCandidate {987                presence_id,988                session_id,989            },990            reason,991        )?;992        if mutation.changed {993            advance_v2_revision(&mut state)?;994            publish_v2_snapshots(&state)?;995        }996        self.session_changed.notify_waiters();997        Ok(mutation.neutralization)998    }9991000    pub(super) fn complete_v2_focus_transition(&self, transition_id: u64) -> Result<()> {1001        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1002        let successor_is_valid = state1003            .focus1004            .transition_successor(transition_id)1005            .and_then(|candidate| {1006                state.viewers.values().find(|viewer| {1007                    viewer.presence_id == candidate.presence_id1008                        && viewer.session.session_id == candidate.session_id1009                        && viewer.session.media_active1010                        && !viewer.authorization_neutralizing1011                        && (viewer1012                            .session1013                            .grants1014                            .contains(InvitationGrants::POINTER_KEYBOARD)1015                            || viewer.session.grants.contains(InvitationGrants::GAMEPAD))1016                })1017            })1018            .is_some();1019        if state1020            .focus1021            .complete_transition(transition_id, successor_is_valid, Instant::now())?1022        {1023            advance_v2_revision(&mut state)?;1024            publish_v2_snapshots(&state)?;1025        }1026        drop(state);1027        self.session_changed.notify_waiters();1028        Ok(())1029    }10301031    pub(super) fn expire_v2_focus(&self) -> Result<Option<FocusNeutralization>> {1032        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1033        let now = Instant::now();1034        let proposal_expired = state.focus.expire_proposal(now);1035        let activation = state.focus.begin_activation_expiry(now)?;1036        if proposal_expired || activation.changed {1037            advance_v2_revision(&mut state)?;1038            publish_v2_snapshots(&state)?;1039            self.session_changed.notify_waiters();1040        }1041        Ok(activation.neutralization)1042    }10431044    pub fn apply_authorization_mutation(1045        &self,1046        mutation: &AuthorizationMutation,1047    ) -> Result<AuthorizationSessionEffect> {1048        ensure!(1049            mutation.committed_revision != 0 && mutation.authorization_revision != 0,1050            "authorization mutation revision is invalid"1051        );1052        self.authorization_committed_revision1053            .fetch_max(mutation.committed_revision, Ordering::SeqCst);10541055        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1056        let Some(viewer) = state.viewers.get(&mutation.peer).filter(|viewer| {1057            viewer.session.viewer_handle.as_deref() == Some(mutation.handle.as_str())1058        }) else {1059            return Ok(AuthorizationSessionEffect::default());1060        };1061        ensure!(1062            mutation.authorization_revision > viewer.session.authorization_revision,1063            "authorization mutation is stale for the active viewer"1064        );1065        let current_grants = mutation.current_grants.unwrap_or(InvitationGrants::VIEW);1066        let input_reduced = mutation1067            .previous_grants1068            .contains(InvitationGrants::POINTER_KEYBOARD)1069            && !current_grants.contains(InvitationGrants::POINTER_KEYBOARD)1070            || mutation.previous_grants.contains(InvitationGrants::GAMEPAD)1071                && !current_grants.contains(InvitationGrants::GAMEPAD);10721073        let focus_candidate = FocusCandidate {1074            presence_id: viewer.presence_id.clone(),1075            session_id: viewer.session.session_id,1076        };1077        let input_claimed = viewer.session.input_claimed;1078        let disconnected = mutation.current_grants.is_none();1079        let focus_mutation = if disconnected || input_reduced {1080            state1081                .focus1082                .begin_invalidation(&focus_candidate, FocusTransitionReasonV2::Revoked)?1083        } else {1084            Default::default()1085        };10861087        if disconnected {1088            let viewer = state1089                .viewers1090                .remove(&mutation.peer)1091                .expect("validated authorization viewer disappeared");1092            viewer.snapshots.send_replace(None);1093            // `release` reclaims focus rate-limit state only for a viewer it1094            // still finds in the roster. Revocation removes it here, so retire1095            // the candidate now or its timestamps outlive the host.1096            state.focus.retire_candidate(&focus_candidate);1097            let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1098            pending.remove(&(mutation.peer, viewer.session.session_id));1099        } else if let Some(grants) = mutation.current_grants {1100            let viewer = state1101                .viewers1102                .get_mut(&mutation.peer)1103                .expect("validated authorization viewer disappeared");1104            viewer.session.authorization_revision = mutation.authorization_revision;1105            viewer.session.authorization_committed_revision = mutation.committed_revision;1106            viewer.session.grants = grants;1107            if input_reduced {1108                viewer.authorization_neutralizing =1109                    input_claimed || focus_mutation.neutralization.is_some();1110            }1111        }1112        advance_v2_revision(&mut state)?;1113        publish_v2_snapshots(&state)?;1114        info!(1115            viewer_handle = %mutation.handle,1116            authorization_revision = mutation.authorization_revision,1117            committed_revision = mutation.committed_revision,1118            reason = if disconnected { "view_revoked" } else if input_reduced { "input_reduced" } else { "grants_changed" },1119            "live viewer authorization changed"1120        );1121        drop(state);1122        self.session_changed.notify_waiters();1123        Ok(AuthorizationSessionEffect {1124            disconnected,1125            neutralize_input: input_claimed || focus_mutation.neutralization.is_some(),1126            focus_transition_id: focus_mutation1127                .neutralization1128                .map(|transition| transition.transition_id),1129        })1130    }11311132    pub fn complete_authorization_neutralization(1133        &self,1134        remote: EndpointId,1135        authorization_revision: u64,1136        focus_transition_id: Option<u64>,1137    ) -> Result<()> {1138        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1139        if let Some(viewer) = state.viewers.get_mut(&remote)1140            && viewer.session.authorization_revision == authorization_revision1141        {1142            viewer.authorization_neutralizing = false;1143        }1144        let completed_focus = if let Some(transition_id) = focus_transition_id {1145            state1146                .focus1147                .complete_transition(transition_id, false, Instant::now())?1148        } else {1149            false1150        };1151        if completed_focus {1152            advance_v2_revision(&mut state)?;1153            publish_v2_snapshots(&state)?;1154        }1155        drop(state);1156        self.session_changed.notify_waiters();1157        Ok(())1158    }11591160    pub(super) fn claim_input_v2(1161        self: &Arc<Self>,1162        remote: EndpointId,1163        nonce: [u8; 16],1164        session_id: u64,1165        slot: ControllerSlot,1166        focus_generation: u64,1167    ) -> Result<InputV2Lease> {1168        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1169        let (presence_id, grants, authorization_revision) = state1170            .viewers1171            .get(&remote)1172            .filter(|viewer| {1173                viewer.session.media_active1174                    && viewer.session.nonce == nonce1175                    && viewer.session.session_id == session_id1176            })1177            .map(|viewer| {1178                (1179                    viewer.presence_id.clone(),1180                    viewer.session.grants,1181                    viewer.session.authorization_revision,1182                )1183            })1184            .context("input v2 connection does not match the active session")?;1185        ensure!(1186            matches!(1187                state.focus.state(),1188                FocusStateV2::Held {1189                    holder,1190                    session_id: holder_session_id,1191                    slot: holder_slot,1192                    focus_generation: holder_generation,1193                } if holder == &presence_id1194                    && *holder_session_id == session_id1195                    && *holder_slot == slot1196                    && *holder_generation == focus_generation1197            ),1198            "input v2 connection does not own the authoritative focus generation"1199        );1200        state.focus.mark_activated(1201            &FocusCandidate {1202                presence_id: presence_id.clone(),1203                session_id,1204            },1205            focus_generation,1206        );1207        let viewer = state1208            .viewers1209            .get_mut(&remote)1210            .expect("validated v2 viewer disappeared");1211        ensure!(1212            !viewer.session.input_claimed,1213            "active viewer already has an input stream"1214        );1215        viewer.session.input_claimed = true;1216        Ok(InputV2Lease {1217            registry: Arc::clone(self),1218            remote,1219            session_id,1220            slot,1221            focus_generation,1222            grants,1223            authorization_revision,1224        })1225    }12261227    pub(crate) fn is_v2_focus_owner(1228        &self,1229        remote: EndpointId,1230        session_id: u64,1231        slot: ControllerSlot,1232        focus_generation: u64,1233    ) -> bool {1234        let state = self.v2_state.lock().expect("v2 session state poisoned");1235        let Some(viewer) = state.viewers.get(&remote).filter(|viewer| {1236            viewer.session.media_active && viewer.session.session_id == session_id1237        }) else {1238            return false;1239        };1240        matches!(1241            state.focus.state(),1242            FocusStateV2::Held {1243                holder,1244                session_id: holder_session_id,1245                slot: holder_slot,1246                focus_generation: holder_generation,1247            } if holder == &viewer.presence_id1248                && *holder_session_id == viewer.session.session_id1249                && *holder_slot == slot1250                && *holder_generation == focus_generation1251        )1252    }12531254    pub(super) fn install_encoder_control(1255        &self,1256        remote: EndpointId,1257        session_id: u64,1258        encoder_control: Option<EncoderControl>,1259    ) -> Result<()> {1260        let mut active = self.active.lock().expect("session registry poisoned");1261        let session = active1262            .as_mut()1263            .filter(|session| {1264                session.media_active && session.remote == remote && session.session_id == session_id1265            })1266            .context("encoder control does not match the active media session")?;1267        ensure!(1268            session.encoder_control.is_none(),1269            "active media session already has encoder control"1270        );1271        session.encoder_control = encoder_control;1272        Ok(())1273    }12741275    pub(super) fn propose_adaptive_encoder_update(1276        &self,1277        remote: EndpointId,1278        session_id: u64,1279        target_kbps: u32,1280        force_keyframe: bool,1281    ) -> Result<Option<AdaptiveEncoderProposal>> {1282        let legacy_control = self1283            .active1284            .lock()1285            .expect("session registry poisoned")1286            .as_ref()1287            .filter(|session| {1288                session.media_active && session.remote == remote && session.session_id == session_id1289            })1290            .and_then(|session| session.encoder_control.clone());1291        let control = legacy_control.or_else(|| {1292            self.v2_state1293                .lock()1294                .expect("v2 session state poisoned")1295                .viewers1296                .get(&remote)1297                .filter(|viewer| {1298                    viewer.session.media_active && viewer.session.session_id == session_id1299                })1300                .and_then(|viewer| viewer.session.encoder_control.clone())1301        });1302        let Some(control) = control else {1303            ensure!(1304                self.is_active(remote, session_id),1305                "adaptive encoder update does not match the active media session"1306            );1307            return Ok(None);1308        };1309        let bitrate_revision = control.request_bitrate_kbps(target_kbps)?;1310        let force_keyframe_revision = force_keyframe1311            .then(|| control.request_force_keyframe())1312            .transpose()?;1313        Ok(Some(AdaptiveEncoderProposal {1314            control,1315            target_kbps,1316            bitrate_revision,1317            force_keyframe_revision,1318        }))1319    }13201321    pub(super) fn propose_resolution_update(1322        &self,1323        remote: EndpointId,1324        session_id: u64,1325        target: VideoDimensions,1326    ) -> Result<Option<ResolutionEncoderProposal>> {1327        let legacy_control = self1328            .active1329            .lock()1330            .expect("session registry poisoned")1331            .as_ref()1332            .filter(|session| {1333                session.media_active && session.remote == remote && session.session_id == session_id1334            })1335            .and_then(|session| session.encoder_control.clone());1336        let control = legacy_control.or_else(|| {1337            self.v2_state1338                .lock()1339                .expect("v2 session state poisoned")1340                .viewers1341                .get(&remote)1342                .filter(|viewer| {1343                    viewer.session.media_active && viewer.session.session_id == session_id1344                })1345                .and_then(|viewer| viewer.session.encoder_control.clone())1346        });1347        let Some(control) = control else {1348            ensure!(1349                self.is_active(remote, session_id),1350                "resolution update does not match the active media session"1351            );1352            return Ok(None);1353        };1354        let revision = control.request_resolution(target.width, target.height)?;1355        Ok(Some(ResolutionEncoderProposal {1356            control,1357            target,1358            revision,1359        }))1360    }13611362    pub(super) fn claim_feedback(1363        self: &Arc<Self>,1364        remote: EndpointId,1365        nonce: [u8; 16],1366    ) -> Result<FeedbackLease> {1367        let mut active = self.active.lock().expect("session registry poisoned");1368        if let Some(session) = active.as_mut().filter(|session| {1369            session.media_active && session.remote == remote && session.nonce == nonce1370        }) {1371            ensure!(1372                session.grants.contains(InvitationGrants::VIEW),1373                "active Portal session lacks feedback view permission"1374            );1375            ensure!(1376                !session.feedback_claimed,1377                "active client already has a feedback connection"1378            );1379            session.feedback_claimed = true;1380            return Ok(FeedbackLease {1381                registry: Arc::clone(self),1382                remote,1383                session_id: session.session_id,1384                telemetry: Arc::clone(&session.media_v3_telemetry),1385                encoder_control: session.encoder_control.clone(),1386                authorization_revision: session.authorization_revision,1387                media_generation_id: session.media_generation_id,1388                scope: FeedbackScope::Legacy,1389            });1390        }1391        drop(active);1392        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1393        let viewer = state1394            .viewers1395            .get_mut(&remote)1396            .filter(|viewer| viewer.session.media_active && viewer.session.nonce == nonce)1397            .context("feedback connection does not match the active media session")?;1398        viewer.rates.feedback_claims.check(1399            Instant::now(),1400            FEEDBACK_CLAIM_RATE_WINDOW,1401            MAX_FEEDBACK_CLAIMS_PER_WINDOW,1402            "media feedback claim",1403        )?;1404        let session = &mut viewer.session;1405        ensure!(1406            session.grants.contains(InvitationGrants::VIEW),1407            "active Portal session lacks feedback view permission"1408        );1409        ensure!(1410            session.authorization_committed_revision1411                == self.authorization_committed_revision.load(Ordering::SeqCst),1412            "feedback claim used a stale committed authorization revision"1413        );1414        ensure!(1415            !session.feedback_claimed,1416            "active client already has a feedback connection"1417        );1418        session.feedback_claimed = true;1419        Ok(FeedbackLease {1420            registry: Arc::clone(self),1421            remote,1422            session_id: session.session_id,1423            telemetry: Arc::clone(&session.media_v3_telemetry),1424            encoder_control: session.encoder_control.clone(),1425            authorization_revision: session.authorization_revision,1426            media_generation_id: session.media_generation_id,1427            scope: FeedbackScope::SharedGeneration,1428        })1429    }14301431    pub(super) fn claim_audio(1432        self: &Arc<Self>,1433        remote: EndpointId,1434        nonce: [u8; 16],1435    ) -> Result<AudioLease> {1436        let mut active = self.active.lock().expect("session registry poisoned");1437        let session = active1438            .as_mut()1439            .filter(|session| {1440                session.mode == SessionMode::LegacyExclusive1441                    && session.media_active1442                    && session.remote == remote1443                    && session.nonce == nonce1444            })1445            .context("audio connection does not match the active media session")?;1446        ensure!(1447            !session.audio_claimed,1448            "active client already has an audio connection"1449        );1450        session.audio_claimed = true;1451        Ok(AudioLease {1452            registry: Arc::clone(self),1453            remote,1454            session_id: session.session_id,1455            session_clock: session.session_clock,1456            grants: session.grants,1457            authorization_revision: session.authorization_revision,1458        })1459    }14601461    pub(super) fn expect_moq(1462        &self,1463        remote: EndpointId,1464        session_id: u64,1465        broadcast_name: String,1466        broadcast: BroadcastConsumer,1467    ) -> Result<MoqAttachmentWait> {1468        let active = self.active.lock().expect("session registry poisoned");1469        let telemetry = active1470            .as_ref()1471            .filter(|session| {1472                session.media_active && session.remote == remote && session.session_id == session_id1473            })1474            .map(|session| Arc::clone(&session.media_v3_telemetry))1475            .context("MoQ expectation does not match the active control session")?;1476        let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1477        ensure!(1478            !pending.contains_key(&(remote, session_id)),1479            "active control session already expects MoQ"1480        );1481        let (attached, attached_rx) = tokio::sync::oneshot::channel();1482        let (closed, closed_rx) = tokio::sync::oneshot::channel();1483        pending.insert(1484            (remote, session_id),1485            PendingMoqAttachment {1486                remote,1487                session_id,1488                broadcast_name,1489                broadcast,1490                attached,1491                closed,1492                telemetry,1493                subscription_capability: None,1494                expected_host: None,1495            },1496        );1497        Ok(MoqAttachmentWait {1498            attached: attached_rx,1499            closed: closed_rx,1500        })1501    }15021503    pub(super) fn expect_moq_v2(1504        &self,1505        remote: EndpointId,1506        session_id: u64,1507        broadcast_name: String,1508        broadcast: BroadcastConsumer,1509        subscription_capability: SignedSubscriptionCapability,1510        expected_host: [u8; 32],1511    ) -> Result<MoqAttachmentWait> {1512        let state = self.v2_state.lock().expect("v2 session state poisoned");1513        let session = state1514            .viewers1515            .get(&remote)1516            .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)1517            .map(|viewer| &viewer.session)1518            .context("authenticated MoQ expectation does not match the active v2 session")?;1519        ensure!(1520            subscription_capability.claims.host_node_id == expected_host1521                && subscription_capability.claims.media_generation_id1522                    == session.media_generation_id1523                && subscription_capability.claims.subscriber_endpoint_id == *remote.as_bytes()1524                && subscription_capability1525                    .claims1526                    .tracks1527                    .contains(SubscriptionTracks::VIDEO_H264)1528                && subscription_capability.claims.authorization_revision1529                    == session.authorization_revision,1530            "subscription capability does not match the v2 viewer attachment"1531        );1532        let telemetry = Arc::clone(&session.media_v3_telemetry);1533        let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1534        ensure!(1535            !pending.contains_key(&(remote, session_id)),1536            "active control session already expects MoQ"1537        );1538        let (attached, attached_rx) = tokio::sync::oneshot::channel();1539        let (closed, closed_rx) = tokio::sync::oneshot::channel();1540        pending.insert(1541            (remote, session_id),1542            PendingMoqAttachment {1543                remote,1544                session_id,1545                broadcast_name,1546                broadcast,1547                attached,1548                closed,1549                telemetry,1550                subscription_capability: Some(subscription_capability),1551                expected_host: Some(expected_host),1552            },1553        );1554        Ok(MoqAttachmentWait {1555            attached: attached_rx,1556            closed: closed_rx,1557        })1558    }15591560    pub(super) fn claim_moq(&self, remote: EndpointId) -> Result<ClaimedMoqAttachment> {1561        let now_unix = SystemTime::now()1562            .duration_since(UNIX_EPOCH)1563            .context("system clock is before the Unix epoch")?1564            .as_secs();1565        self.claim_moq_at(remote, now_unix)1566    }15671568    fn claim_moq_at(&self, remote: EndpointId, now_unix: u64) -> Result<ClaimedMoqAttachment> {1569        let legacy = self1570            .active1571            .lock()1572            .expect("session registry poisoned")1573            .as_ref()1574            .filter(|session| session.media_active && session.remote == remote)1575            .cloned();1576        let v2 = {1577            let mut state = self.v2_state.lock().expect("v2 session state poisoned");1578            state1579                .viewers1580                .get_mut(&remote)1581                .filter(|viewer| viewer.session.media_active)1582                .map(|viewer| {1583                    viewer.rates.attachments.check(1584                        Instant::now(),1585                        ATTACHMENT_RATE_WINDOW,1586                        MAX_ATTACHMENTS_PER_WINDOW,1587                        "MoQ attachment",1588                    )?;1589                    Ok::<ActiveSession, anyhow::Error>(viewer.session.clone())1590                })1591                .transpose()?1592        };1593        let session = legacy1594            .or(v2)1595            .context("MoQ connection does not match an active control session")?;1596        let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1597        let attachment = pending1598            .get(&(remote, session.session_id))1599            .context("active control session is not expecting a MoQ connection")?;1600        if session.mode == SessionMode::V2Single {1601            let capability = attachment1602                .subscription_capability1603                .as_ref()1604                .context("v2 MoQ attachment has no subscription capability")?;1605            let expected_host = attachment1606                .expected_host1607                .context("v2 MoQ attachment has no expected host identity")?;1608            capability1609                .verify_binding(1610                    expected_host,1611                    session.media_generation_id,1612                    *remote.as_bytes(),1613                    capability.claims.tracks,1614                    session.authorization_revision,1615                    now_unix,1616                )1617                .context("verifying endpoint-bound MoQ subscription capability")?;1618        } else {1619            ensure!(1620                attachment.subscription_capability.is_none(),1621                "legacy MoQ attachment unexpectedly carries a subscription capability"1622            );1623        }1624        debug_assert_eq!(attachment.remote, session.remote);1625        let attachment = pending1626            .remove(&(remote, session.session_id))1627            .expect("validated pending MoQ attachment disappeared");1628        Ok(ClaimedMoqAttachment {1629            session_id: attachment.session_id,1630            broadcast_name: attachment.broadcast_name,1631            broadcast: attachment.broadcast,1632            attached: attachment.attached,1633            closed: attachment.closed,1634            telemetry: attachment.telemetry,1635        })1636    }16371638    pub(super) fn admit_v2_keyframe_request(1639        &self,1640        remote: EndpointId,1641        session_id: u64,1642    ) -> Result<()> {1643        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1644        let viewer = state1645            .viewers1646            .get_mut(&remote)1647            .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)1648            .context("keyframe request does not match the active viewer generation")?;1649        viewer.rates.keyframes.check(1650            Instant::now(),1651            KEYFRAME_RATE_WINDOW,1652            MAX_KEYFRAME_REQUESTS_PER_WINDOW,1653            "keyframe request",1654        )1655    }16561657    #[cfg(test)]1658    pub(super) fn focus_rate_limit_entries(&self) -> usize {1659        self.v2_state1660            .lock()1661            .expect("v2 session state poisoned")1662            .focus1663            .tracked_command_candidates()1664    }16651666    fn release(&self, remote: EndpointId, session_id: u64) {1667        let mut active = self.active.lock().expect("session registry poisoned");1668        if let Some(session) = active.as_mut()1669            && session.remote == remote1670            && session.session_id == session_id1671        {1672            let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1673            pending.remove(&(remote, session_id));1674            // Keep the registry occupied until the input handler has observed1675            // media shutdown and released all held uinput transitions. This1676            // prevents a reconnect from sharing the device with a draining1677            // predecessor session.1678            session.media_active = false;1679            session.encoder_control = None;1680            if !session.input_claimed && !session.audio_claimed && !session.feedback_claimed {1681                *active = None;1682            }1683            drop(active);1684            self.session_changed.notify_waiters();1685            return;1686        }1687        drop(active);16881689        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1690        state.live_control_leases = state.live_control_leases.saturating_sub(1);1691        let Some(viewer) = state1692            .viewers1693            .get(&remote)1694            .filter(|viewer| viewer.session.session_id == session_id)1695        else {1696            return;1697        };1698        let presence_id = viewer.presence_id.clone();1699        let viewer_handle = viewer1700            .session1701            .viewer_handle1702            .clone()1703            .unwrap_or_else(|| presence_id.as_str().to_owned());1704        let focus_mutation = state.focus.begin_invalidation(1705            &FocusCandidate {1706                presence_id: presence_id.clone(),1707                session_id,1708            },1709            FocusTransitionReasonV2::Disconnected,1710        );1711        let viewer = state1712            .viewers1713            .remove(&remote)1714            .expect("validated v2 viewer disappeared");1715        viewer.snapshots.send_replace(None);1716        self.pending_moq1717            .lock()1718            .expect("MoQ registry poisoned")1719            .remove(&(remote, session_id));1720        if let Err(error) = focus_mutation {1721            warn!(%error, session_id, "failed to invalidate focus during viewer release");1722        }1723        state.focus.retire_candidate(&FocusCandidate {1724            presence_id,1725            session_id,1726        });1727        let active_viewers = state.viewers.len();1728        if state.viewers.is_empty() {1729            state.media = None;1730        }1731        if advance_v2_revision(&mut state).is_ok() {1732            let _ = publish_v2_snapshots(&state);1733        }1734        drop(state);1735        self.session_changed.notify_waiters();1736        info!(1737            viewer_handle,1738            session_id,1739            active_viewers,1740            reason = "disconnected",1741            "multi-viewer session membership changed"1742        );1743    }17441745    pub(crate) fn is_active(&self, remote: EndpointId, session_id: u64) -> bool {1746        if self1747            .active1748            .lock()1749            .expect("session registry poisoned")1750            .as_ref()1751            .is_some_and(|active| {1752                active.media_active && active.remote == remote && active.session_id == session_id1753            })1754        {1755            return true;1756        }1757        self.v2_state1758            .lock()1759            .expect("v2 session state poisoned")1760            .viewers1761            .get(&remote)1762            .is_some_and(|viewer| {1763                viewer.session.media_active && viewer.session.session_id == session_id1764            })1765    }17661767    pub(super) fn disconnect_v2_viewer(&self, remote: EndpointId, session_id: u64) -> Result<bool> {1768        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1769        let Some(viewer) = state1770            .viewers1771            .get(&remote)1772            .filter(|viewer| viewer.session.session_id == session_id)1773        else {1774            return Ok(false);1775        };1776        let viewer_handle = viewer1777            .session1778            .viewer_handle1779            .clone()1780            .unwrap_or_else(|| viewer.presence_id.as_str().to_owned());1781        ensure!(1782            !matches!(1783                state.focus.state(),1784                FocusStateV2::Held {1785                    session_id: holder_session_id,1786                    ..1787                } if *holder_session_id == session_id1788            ),1789            "focused viewer must be neutralized before adaptive detachment"1790        );1791        let retired = FocusCandidate {1792            presence_id: viewer.presence_id.clone(),1793            session_id,1794        };1795        viewer.snapshots.send_replace(None);1796        state.viewers.remove(&remote);1797        // Detaching bypasses `release`'s roster lookup, so reclaim this1798        // viewer's focus rate-limit timestamps explicitly.1799        state.focus.retire_candidate(&retired);1800        if state.viewers.is_empty() {1801            state.media = None;1802        }1803        advance_v2_revision(&mut state)?;1804        publish_v2_snapshots(&state)?;1805        drop(state);1806        self.pending_moq1807            .lock()1808            .expect("MoQ registry poisoned")1809            .remove(&(remote, session_id));1810        self.session_changed.notify_waiters();1811        info!(1812            viewer_handle,1813            session_id,1814            active_viewers = self.runtime_status().active_viewers,1815            reason = "adaptive_recovery_detach",1816            "multi-viewer session membership changed"1817        );1818        Ok(true)1819    }18201821    fn release_input(&self, remote: EndpointId, session_id: u64) {1822        let mut active = self.active.lock().expect("session registry poisoned");1823        if let Some(session) = active.as_mut()1824            && session.remote == remote1825            && session.session_id == session_id1826        {1827            session.input_claimed = false;1828            if !session.media_active && !session.audio_claimed && !session.feedback_claimed {1829                *active = None;1830            }1831            return;1832        }1833        drop(active);1834        let mut state = self.v2_state.lock().expect("v2 session state poisoned");1835        if let Some(viewer) = state.viewers.get_mut(&remote)1836            && viewer.session.session_id == session_id1837        {1838            viewer.session.input_claimed = false;1839        }1840        let completed_neutralization = match state.focus.state() {1841            FocusStateV2::Neutralizing {1842                former_session_id,1843                transition_id,1844                ..1845            } if *former_session_id == session_id => Some(*transition_id),1846            _ => None,1847        };1848        if let Some(transition_id) = completed_neutralization {1849            let _ = state1850                .focus1851                .complete_transition(transition_id, false, Instant::now());1852            if advance_v2_revision(&mut state).is_ok() {1853                let _ = publish_v2_snapshots(&state);1854            }1855            drop(state);1856            self.session_changed.notify_waiters();1857        }1858    }18591860    fn release_audio(&self, remote: EndpointId, session_id: u64) {1861        let mut active = self.active.lock().expect("session registry poisoned");1862        if let Some(session) = active.as_mut()1863            && session.remote == remote1864            && session.session_id == session_id1865        {1866            session.audio_claimed = false;1867            if !session.media_active && !session.input_claimed && !session.feedback_claimed {1868                *active = None;1869            }1870        }1871    }18721873    fn release_feedback(&self, remote: EndpointId, session_id: u64) {1874        let mut active = self.active.lock().expect("session registry poisoned");1875        if let Some(session) = active.as_mut()1876            && session.remote == remote1877            && session.session_id == session_id1878        {1879            session.feedback_claimed = false;1880            if !session.media_active && !session.input_claimed && !session.audio_claimed {1881                *active = None;1882            }1883            return;1884        }1885        drop(active);1886        if let Some(viewer) = self1887            .v2_state1888            .lock()1889            .expect("v2 session state poisoned")1890            .viewers1891            .get_mut(&remote)1892            && viewer.session.session_id == session_id1893        {1894            viewer.session.feedback_claimed = false;1895        }1896    }1897}18981899fn advance_v2_revision(state: &mut V2SessionState) -> Result<()> {1900    state.revision = state1901        .revision1902        .checked_add(1)1903        .context("v2 snapshot revision overflowed")?;1904    Ok(())1905}19061907fn publish_v2_snapshots(state: &V2SessionState) -> Result<()> {1908    for (remote, viewer) in &state.viewers {1909        viewer1910            .snapshots1911            .send_replace(Some(v2_snapshot_for(state, *remote)?));1912    }1913    Ok(())1914}19151916fn v2_snapshot_for(state: &V2SessionState, remote: EndpointId) -> Result<SessionSnapshotV2> {1917    let self_viewer = state1918        .viewers1919        .get(&remote)1920        .context("active v2 viewer is missing from snapshot state")?;1921    let mut viewers = state1922        .viewers1923        .values()1924        .map(|viewer| ViewerPresenceV2 {1925            presence_id: viewer.presence_id.clone(),1926            session_id: viewer.session.session_id,1927            input_capable: viewer1928                .session1929                .grants1930                .contains(InvitationGrants::POINTER_KEYBOARD)1931                || viewer.session.grants.contains(InvitationGrants::GAMEPAD),1932            you: viewer.session.remote == remote,1933        })1934        .collect::<Vec<_>>();1935    viewers.sort_by(|left, right| left.presence_id.cmp(&right.presence_id));1936    let snapshot = SessionSnapshotV2 {1937        revision: state.revision,1938        self_presence_id: self_viewer.presence_id.clone(),1939        viewers,1940        focus: state.focus.state().clone(),1941        focus_proposal: state.focus.proposal().cloned(),1942        self_is_focus_owner: state.focus.is_configured_owner(&self_viewer.presence_id),1943        transition_reason: state.focus.transition_reason(),1944        media: state.media.clone().unwrap_or(MediaGenerationDescriptorV2 {1945            generation_id: self_viewer.session.media_generation_id,1946            broadcast_name: self_viewer.session.media_broadcast_name.clone(),1947        }),1948    };1949    snapshot.validate()?;1950    Ok(snapshot)1951}19521953#[derive(Debug)]1954pub(super) struct SessionLease {1955    registry: Arc<SessionRegistry>,1956    remote: EndpointId,1957    pub(super) session_id: u64,1958    pub(super) session_clock: SessionClock,1959    pub(super) media_v3_telemetry: Arc<MediaV3Telemetry>,1960}19611962#[derive(Debug)]1963pub(crate) struct V2SessionLease {1964    registry: Arc<SessionRegistry>,1965    remote: EndpointId,1966    pub(crate) session_id: u64,1967    #[allow(dead_code)]1968    pub(super) session_clock: SessionClock,1969    #[allow(dead_code)]1970    pub(super) media_v3_telemetry: Arc<MediaV3Telemetry>,1971    #[allow(dead_code)]1972    pub(super) initial_snapshot: SessionSnapshotV2,1973    pub(super) authorization_revision: u64,1974    pub(super) authorization_committed_revision: u64,1975    replaced_snapshots: Option<tokio::sync::watch::Sender<Option<SessionSnapshotV2>>>,1976    pub(super) replacement_transition: Option<FocusNeutralization>,1977}19781979impl V2SessionLease {1980    pub(super) fn retire_replaced_viewer(&mut self) {1981        if let Some(snapshots) = self.replaced_snapshots.take() {1982            snapshots.send_replace(None);1983        }1984    }1985}19861987impl Drop for V2SessionLease {1988    fn drop(&mut self) {1989        self.retire_replaced_viewer();1990        self.registry.release(self.remote, self.session_id);1991    }1992}19931994impl Drop for SessionLease {1995    fn drop(&mut self) {1996        self.registry.release(self.remote, self.session_id);1997    }1998}19992000#[derive(Debug)]2001pub(super) struct InputLease {2002    registry: Arc<SessionRegistry>,2003    remote: EndpointId,2004    pub(super) session_id: u64,2005    pub(super) grants: InvitationGrants,2006    pub(super) authorization_revision: u64,2007}20082009#[derive(Debug)]2010pub(super) struct InputV2Lease {2011    registry: Arc<SessionRegistry>,2012    remote: EndpointId,2013    pub(super) session_id: u64,2014    pub(super) slot: ControllerSlot,2015    pub(super) focus_generation: u64,2016    pub(super) grants: InvitationGrants,2017    pub(super) authorization_revision: u64,2018}20192020#[derive(Debug)]2021pub(super) struct AudioLease {2022    registry: Arc<SessionRegistry>,2023    remote: EndpointId,2024    pub(super) session_id: u64,2025    pub(super) session_clock: SessionClock,2026    pub(super) grants: InvitationGrants,2027    pub(super) authorization_revision: u64,2028}20292030#[derive(Debug)]2031pub(super) struct FeedbackLease {2032    registry: Arc<SessionRegistry>,2033    remote: EndpointId,2034    pub(super) session_id: u64,2035    pub(super) telemetry: Arc<MediaV3Telemetry>,2036    pub(super) encoder_control: Option<EncoderControl>,2037    pub(super) authorization_revision: u64,2038    pub(super) media_generation_id: u64,2039    pub(super) scope: FeedbackScope,2040}20412042#[derive(Clone, Copy, Debug, PartialEq, Eq)]2043pub(super) enum FeedbackScope {2044    Legacy,2045    SharedGeneration,2046}20472048#[derive(Debug)]2049pub(super) struct SourceTaskGuard(Option<tokio::task::JoinHandle<Result<()>>>);20502051impl SourceTaskGuard {2052    pub(super) fn new(task: tokio::task::JoinHandle<Result<()>>) -> Self {2053        Self(Some(task))2054    }20552056    pub(super) async fn abort_and_wait(mut self) {2057        if let Some(task) = self.0.take() {2058            task.abort();2059            let _ = task.await;2060        }2061    }20622063    pub(super) async fn wait_or_abort(mut self, grace_timeout: Duration) {2064        let Some(mut task) = self.0.take() else {2065            return;2066        };2067        if tokio::time::timeout(grace_timeout, &mut task)2068            .await2069            .is_err()2070        {2071            task.abort();2072            let _ = task.await;2073        }2074    }2075}20762077impl Drop for SourceTaskGuard {2078    fn drop(&mut self) {2079        if let Some(task) = self.0.take() {2080            task.abort();2081        }2082    }2083}20842085impl Drop for InputLease {2086    fn drop(&mut self) {2087        self.registry.release_input(self.remote, self.session_id);2088    }2089}20902091impl Drop for InputV2Lease {2092    fn drop(&mut self) {2093        self.registry.release_input(self.remote, self.session_id);2094    }2095}20962097impl Drop for AudioLease {2098    fn drop(&mut self) {2099        self.registry.release_audio(self.remote, self.session_id);2100    }2101}21022103impl Drop for FeedbackLease {2104    fn drop(&mut self) {2105        self.registry.release_feedback(self.remote, self.session_id);2106    }2107}21082109#[cfg(test)]2110mod tests {2111    use super::super::endpoint;2112    use super::*;21132114    use moq_net::{Broadcast, BroadcastProducer};2115    use sigil_protocol::media_moq_broadcast_name;21162117    #[tokio::test]2118    async fn forced_idr_recovery_is_one_slot_join_safe_and_rearms_after_ack() {2119        let telemetry = Arc::new(MediaV3Telemetry::default());2120        let harness = crate::source::EncoderControlTestHarness::new();2121        let mut coordinator =2122            ForcedIdrCoordinator::new(Some(harness.control.clone()), Arc::clone(&telemetry));21232124        assert_eq!(2125            coordinator.request(KeyframeRequestReasonV3::Join),2126            ForcedIdrDisposition::JoinReplay2127        );2128        assert_eq!(harness.requested_force_keyframe_revision(), None);21292130        let requested_revision = match coordinator.request(KeyframeRequestReasonV3::DecoderReset) {2131            ForcedIdrDisposition::Requested { revision } => revision,2132            disposition => panic!("unexpected forced-IDR disposition: {disposition:?}"),2133        };2134        assert_eq!(2135            harness.requested_force_keyframe_revision(),2136            Some(requested_revision)2137        );2138        assert_eq!(2139            coordinator.request(KeyframeRequestReasonV3::TransportGap),2140            ForcedIdrDisposition::Coalesced {2141                revision: requested_revision2142            }2143        );2144        assert_eq!(telemetry.encoder_force_requests.load(Ordering::Relaxed), 1);21452146        let newer_revision = harness.control.request_force_keyframe().unwrap();2147        harness.status.send_modify(|status| {2148            status.requested_force_keyframe_revision = Some(newer_revision);2149            status.acknowledged_force_keyframe_revision = Some(newer_revision);2150        });2151        let acknowledgement = coordinator.acknowledgements.join_next().await;2152        coordinator.complete(acknowledgement, endpoint(1), "test");2153        assert_eq!(coordinator.pending_revision, None);2154        assert_eq!(2155            telemetry2156                .encoder_force_acknowledgements2157                .load(Ordering::Relaxed),2158            12159        );2160        assert_eq!(telemetry.encoder_force_failures.load(Ordering::Relaxed), 0);21612162        assert!(matches!(2163            coordinator.request(KeyframeRequestReasonV3::DecoderReset),2164            ForcedIdrDisposition::Requested { revision } if revision > newer_revision2165        ));2166        coordinator.abort_and_drain(endpoint(1), "test").await;21672168        let fallback_telemetry = Arc::new(MediaV3Telemetry::default());2169        let mut fallback = ForcedIdrCoordinator::new(None, fallback_telemetry);2170        assert_eq!(2171            fallback.request(KeyframeRequestReasonV3::DecoderReset),2172            ForcedIdrDisposition::Unavailable2173        );2174        assert!(fallback.acknowledgements.is_empty());2175    }21762177    #[test]2178    fn only_one_remote_can_hold_session() {2179        let sessions = Arc::new(SessionRegistry::default());2180        assert!(!sessions.has_session());2181        let nonce = [7; 16];2182        let first = sessions2183            .claim(endpoint(1), nonce, InvitationGrants::ALL)2184            .unwrap();2185        assert!(sessions.has_session());2186        assert!(2187            sessions2188                .claim(endpoint(2), nonce, InvitationGrants::ALL)2189                .is_err()2190        );2191        assert!(sessions.claim_input(endpoint(1), [8; 16]).is_err());2192        let input = sessions.claim_input(endpoint(1), nonce).unwrap();2193        assert_eq!(input.session_id, first.session_id);2194        assert!(sessions.claim_input(endpoint(1), nonce).is_err());2195        let audio = sessions.claim_audio(endpoint(1), nonce).unwrap();2196        assert!(sessions.claim_audio(endpoint(1), nonce).is_err());2197        drop(input);2198        let draining_input = sessions.claim_input(endpoint(1), nonce).unwrap();2199        drop(first);2200        assert!(2201            sessions2202                .claim(endpoint(2), nonce, InvitationGrants::ALL)2203                .is_err()2204        );2205        drop(draining_input);2206        assert!(2207            sessions2208                .claim(endpoint(2), nonce, InvitationGrants::ALL)2209                .is_err()2210        );2211        drop(audio);2212        assert!(!sessions.has_session());2213        assert!(2214            sessions2215                .claim(endpoint(2), nonce, InvitationGrants::ALL)2216                .is_ok()2217        );2218    }22192220    #[test]2221    fn v2_single_viewer_mode_excludes_legacy_and_publishes_revisioned_focus() {2222        let sessions = Arc::new(SessionRegistry::default());2223        let remote = endpoint(1);2224        let nonce = [6; 16];2225        let media = sessions2226            .claim_v2(remote, nonce, InvitationGrants::ALL)2227            .unwrap();2228        assert_eq!(media.initial_snapshot.revision, 1);2229        assert!(matches!(2230            media.initial_snapshot.focus,2231            FocusStateV2::Vacant { .. }2232        ));2233        assert!(2234            sessions2235                .claim(endpoint(2), nonce, InvitationGrants::ALL)2236                .is_err()2237        );22382239        let request = FocusCommandV2 {2240            request_id: 1,2241            action: FocusCommandActionV2::Request,2242            slot: ControllerSlot::ZERO,2243            expected_revision: 1,2244            expected_focus_generation: None,2245            expected_proposal_id: None,2246        };2247        let granted = sessions2248            .apply_focus_command(remote, media.session_id, &request)2249            .unwrap();2250        assert_eq!(granted.snapshot.revision, 2);2251        assert!(granted.neutralization.is_none());2252        let focus_generation = granted.snapshot.self_focus_generation().unwrap();2253        assert!(sessions.is_v2_focus_owner(2254            remote,2255            media.session_id,2256            ControllerSlot::ZERO,2257            focus_generation2258        ));2259        assert!(2260            sessions2261                .claim_input_v2(2262                    remote,2263                    nonce,2264                    media.session_id,2265                    ControllerSlot::ZERO,2266                    focus_generation + 1,2267                )2268                .is_err()2269        );2270        let input = sessions2271            .claim_input_v2(2272                remote,2273                nonce,2274                media.session_id,2275                ControllerSlot::ZERO,2276                focus_generation,2277            )2278            .unwrap();22792280        let release = FocusCommandV2 {2281            request_id: 2,2282            action: FocusCommandActionV2::Release,2283            slot: ControllerSlot::ZERO,2284            expected_revision: 2,2285            expected_focus_generation: Some(focus_generation),2286            expected_proposal_id: None,2287        };2288        let released = sessions2289            .apply_focus_command(remote, media.session_id, &release)2290            .unwrap();2291        assert_eq!(released.snapshot.revision, 3);2292        let transition = released2293            .neutralization2294            .expect("release must publish neutralizing state");2295        assert!(!sessions.is_v2_focus_owner(2296            remote,2297            media.session_id,2298            ControllerSlot::ZERO,2299            focus_generation2300        ));2301        assert!(2302            sessions2303                .apply_focus_command(remote, media.session_id, &release)2304                .is_err(),2305            "a stale release must not mutate the newer snapshot revision"2306        );2307        sessions2308            .complete_v2_focus_transition(transition.transition_id)2309            .unwrap();2310        assert_eq!(sessions.v2_revision(remote, media.session_id).unwrap(), 4);2311        drop(input);2312        drop(media);2313    }23142315    fn authorized_viewer(handle: &str, grants: InvitationGrants) -> AuthorizedViewer {2316        AuthorizedViewer {2317            handle: handle.to_owned(),2318            grants,2319            authorization_revision: 1,2320            committed_revision: 1,2321        }2322    }23232324    // Regression: `claim_v2_authorized` used to drop the legacy registry lock2325    // before taking the v2 lock, so a legacy client and a v2 viewer could both2326    // clear their mutual-exclusion check and be admitted at the same time.2327    //2328    // The interleaving is too narrow to reproduce by racing threads, so assert2329    // the lock discipline that closes it directly: while v2 admission is parked2330    // on the v2 lock it must still be holding the legacy registry lock. Under2331    // the old ordering it had already released it, and `claim` could walk in.2332    #[test]2333    fn v2_admission_holds_the_legacy_registry_lock_across_the_v2_mutation() {2334        let sessions = Arc::new(SessionRegistry::default());2335        let blocked = Arc::clone(&sessions);2336        let state_guard = sessions.v2_state.lock().expect("v2 session state poisoned");23372338        let admission = std::thread::spawn(move || {2339            blocked.claim_v2_authorized(2340                endpoint(2),2341                [2; 16],2342                authorized_viewer("viewer-0000000000000002", InvitationGrants::ALL),2343            )2344        });2345        // Generous settle time: the thread is now parked on `v2_state`.2346        std::thread::sleep(Duration::from_millis(500));2347        assert!(2348            sessions.active.try_lock().is_err(),2349            "v2 admission released the legacy registry lock before mutating v2 state, \2350             which lets `claim` admit a legacy client concurrently"2351        );23522353        drop(state_guard);2354        // Hold the lease: dropping it would release the viewer and make the2355        // exclusion assertion below vacuous.2356        let _viewer = admission2357            .join()2358            .expect("v2 admission thread panicked")2359            .unwrap_or_else(|error| {2360                panic!("v2 admission must succeed once the v2 lock is free: {error:#}")2361            });2362        assert!(2363            sessions2364                .claim(endpoint(1), [1; 16], InvitationGrants::ALL)2365                .is_err(),2366            "a legacy client must be refused while a v2 viewer holds a control lease"2367        );2368    }23692370    #[test]2371    fn legacy_and_v2_admission_stay_mutually_exclusive_under_contention() {2372        for round in 0..300_u16 {2373            let sessions = Arc::new(SessionRegistry::default());2374            let legacy_registry = Arc::clone(&sessions);2375            let v2_registry = Arc::clone(&sessions);2376            let gate = Arc::new(std::sync::Barrier::new(2));2377            let legacy_gate = Arc::clone(&gate);2378            let v2_gate = Arc::clone(&gate);23792380            let legacy = std::thread::spawn(move || {2381                legacy_gate.wait();2382                legacy_registry2383                    .claim(endpoint(1), [1; 16], InvitationGrants::ALL)2384                    .ok()2385            });2386            let v2 = std::thread::spawn(move || {2387                v2_gate.wait();2388                v2_registry2389                    .claim_v2_authorized(2390                        endpoint(2),2391                        [2; 16],2392                        authorized_viewer("viewer-0000000000000002", InvitationGrants::ALL),2393                    )2394                    .ok()2395            });2396            let legacy = legacy.join().expect("legacy admission thread panicked");2397            let v2 = v2.join().expect("v2 admission thread panicked");2398            assert!(2399                legacy.is_none() || v2.is_none(),2400                "round {round} admitted a legacy client and a v2 viewer at once"2401            );2402            assert!(2403                legacy.is_some() || v2.is_some(),2404                "round {round} admitted neither client"2405            );2406        }2407    }24082409    // Regression: focus rate-limit timestamps were reclaimed only by `release`,2410    // which cannot find a viewer that revocation or adaptive detachment already2411    // removed from the roster.2412    #[test]2413    fn focus_rate_limit_state_is_reclaimed_on_every_viewer_exit() {2414        let request = FocusCommandV2 {2415            request_id: 1,2416            action: FocusCommandActionV2::Request,2417            slot: ControllerSlot::ZERO,2418            expected_revision: 1,2419            expected_focus_generation: None,2420            expected_proposal_id: None,2421        };24222423        // Ordinary disconnect.2424        let sessions = Arc::new(SessionRegistry::default());2425        let remote = endpoint(1);2426        let lease = sessions2427            .claim_v2_authorized(2428                remote,2429                [1; 16],2430                authorized_viewer("viewer-0000000000000001", InvitationGrants::ALL),2431            )2432            .unwrap();2433        sessions2434            .apply_focus_command(remote, lease.session_id, &request)2435            .unwrap();2436        assert_eq!(sessions.focus_rate_limit_entries(), 1);2437        drop(lease);2438        assert_eq!(2439            sessions.focus_rate_limit_entries(),2440            0,2441            "disconnect must reclaim focus rate-limit state"2442        );24432444        // Adaptive slow-viewer detachment.2445        let sessions = Arc::new(SessionRegistry::default());2446        let holder = endpoint(1);2447        let spectator = endpoint(2);2448        let holder_lease = sessions2449            .claim_v2_authorized(2450                holder,2451                [1; 16],2452                authorized_viewer("viewer-0000000000000001", InvitationGrants::ALL),2453            )2454            .unwrap();2455        let spectator_lease = sessions2456            .claim_v2_authorized(2457                spectator,2458                [2; 16],2459                authorized_viewer("viewer-0000000000000002", InvitationGrants::ALL),2460            )2461            .unwrap();2462        let mut spectator_request = request.clone();2463        spectator_request.expected_revision = sessions2464            .v2_revision(spectator, spectator_lease.session_id)2465            .unwrap();2466        sessions2467            .apply_focus_command(spectator, spectator_lease.session_id, &spectator_request)2468            .unwrap();2469        assert_eq!(sessions.focus_rate_limit_entries(), 1);2470        // The spectator now holds focus, so release it before detaching.2471        let mut release = spectator_request.clone();2472        release.action = FocusCommandActionV2::Release;2473        release.expected_revision = sessions2474            .v2_revision(spectator, spectator_lease.session_id)2475            .unwrap();2476        release.expected_focus_generation = match sessions.v2_state.lock().unwrap().focus.state() {2477            FocusStateV2::Held {2478                focus_generation, ..2479            } => Some(*focus_generation),2480            other => panic!("expected held focus, got {other:?}"),2481        };2482        let effect = sessions2483            .apply_focus_command(spectator, spectator_lease.session_id, &release)2484            .unwrap();2485        if let Some(transition) = effect.neutralization {2486            sessions2487                .complete_v2_focus_transition(transition.transition_id)2488                .unwrap();2489        }2490        assert!(2491            sessions2492                .disconnect_v2_viewer(spectator, spectator_lease.session_id)2493                .unwrap()2494        );2495        assert_eq!(2496            sessions.focus_rate_limit_entries(),2497            0,2498            "adaptive detachment must reclaim focus rate-limit state"2499        );2500        drop(spectator_lease);2501        drop(holder_lease);2502    }25032504    #[test]2505    fn v2_admits_bounded_viewers_and_publishes_personalized_complete_rosters() {2506        let sessions = Arc::new(SessionRegistry::new(3));2507        let telemetry = Arc::new(MediaV3Telemetry::default());2508        let mut leases = Vec::new();2509        for index in 1..=3 {2510            let remote = endpoint(index);2511            let lease = sessions2512                .claim_v2_authorized(2513                    remote,2514                    [index; 16],2515                    authorized_viewer(&format!("viewer-{index}"), InvitationGrants::ALL),2516                )2517                .unwrap();2518            sessions2519                .bind_v2_generation(2520                    remote,2521                    lease.session_id,2522                    77,2523                    SessionClock::start(),2524                    Arc::clone(&telemetry),2525                    None,2526                )2527                .unwrap();2528            leases.push(lease);2529        }25302531        for (index, lease) in leases.iter().enumerate() {2532            let remote = endpoint(u8::try_from(index + 1).unwrap());2533            let snapshots = sessions2534                .subscribe_v2_snapshots(remote, lease.session_id)2535                .unwrap();2536            let snapshot = snapshots.borrow().clone().unwrap();2537            assert_eq!(snapshot.viewers.len(), 3);2538            assert_eq!(snapshot.media.generation_id, 77);2539            assert_eq!(2540                snapshot.viewers.iter().filter(|viewer| viewer.you).count(),2541                12542            );2543            assert_eq!(snapshot.self_viewer().session_id, lease.session_id);2544        }25452546        assert!(2547            sessions2548                .claim_v2_authorized(2549                    endpoint(4),2550                    [4; 16],2551                    authorized_viewer("viewer-4", InvitationGrants::VIEW),2552                )2553                .unwrap_err()2554                .to_string()2555                .contains("capacity")2556        );2557    }25582559    #[test]2560    fn same_peer_replacement_is_immediate_generation_safe_and_neutralizes_focus() {2561        let sessions = Arc::new(SessionRegistry::new(2));2562        let remote = endpoint(1);2563        let old = sessions2564            .claim_v2_authorized(2565                remote,2566                [1; 16],2567                authorized_viewer("viewer-one", InvitationGrants::ALL),2568            )2569            .unwrap();2570        let granted = sessions2571            .apply_focus_command(2572                remote,2573                old.session_id,2574                &FocusCommandV2 {2575                    request_id: 1,2576                    action: FocusCommandActionV2::Request,2577                    slot: ControllerSlot::ZERO,2578                    expected_revision: 1,2579                    expected_focus_generation: None,2580                    expected_proposal_id: None,2581                },2582            )2583            .unwrap();2584        assert!(granted.snapshot.self_focus_generation().is_some());25852586        let mut replacement = sessions2587            .claim_v2_authorized(2588                remote,2589                [2; 16],2590                authorized_viewer("viewer-one", InvitationGrants::ALL),2591            )2592            .unwrap();2593        let transition = replacement2594            .replacement_transition2595            .expect("focused replacement must neutralize");2596        assert!(matches!(2597            replacement.initial_snapshot.focus,2598            FocusStateV2::Neutralizing { .. }2599        ));2600        sessions2601            .complete_v2_focus_transition(transition.transition_id)2602            .unwrap();2603        let replacement_snapshot = sessions2604            .subscribe_v2_snapshots(remote, replacement.session_id)2605            .unwrap()2606            .borrow()2607            .clone()2608            .unwrap();2609        assert!(matches!(2610            replacement_snapshot.focus,2611            FocusStateV2::Vacant { .. }2612        ));2613        replacement.retire_replaced_viewer();2614        drop(old);2615        assert!(sessions.is_active(remote, replacement.session_id));2616        assert!(!sessions.is_active(remote, granted.snapshot.self_viewer().session_id));2617    }26182619    #[test]2620    fn occupied_focus_retains_one_proposal_without_queueing_additional_viewers() {2621        let sessions = Arc::new(SessionRegistry::new(3));2622        let first = sessions2623            .claim_v2_authorized(2624                endpoint(1),2625                [1; 16],2626                authorized_viewer("viewer-one", InvitationGrants::ALL),2627            )2628            .unwrap();2629        let second = sessions2630            .claim_v2_authorized(2631                endpoint(2),2632                [2; 16],2633                authorized_viewer("viewer-two", InvitationGrants::ALL),2634            )2635            .unwrap();2636        let third = sessions2637            .claim_v2_authorized(2638                endpoint(3),2639                [3; 16],2640                authorized_viewer("viewer-three", InvitationGrants::ALL),2641            )2642            .unwrap();2643        let revision = third.initial_snapshot.revision;2644        sessions2645            .apply_focus_command(2646                endpoint(1),2647                first.session_id,2648                &FocusCommandV2 {2649                    request_id: 1,2650                    action: FocusCommandActionV2::Request,2651                    slot: ControllerSlot::ZERO,2652                    expected_revision: revision,2653                    expected_focus_generation: None,2654                    expected_proposal_id: None,2655                },2656            )2657            .unwrap();2658        let occupied_revision = sessions2659            .v2_revision(endpoint(2), second.session_id)2660            .unwrap();2661        let proposal = sessions2662            .apply_focus_command(2663                endpoint(2),2664                second.session_id,2665                &FocusCommandV2 {2666                    request_id: 2,2667                    action: FocusCommandActionV2::Request,2668                    slot: ControllerSlot::ZERO,2669                    expected_revision: occupied_revision,2670                    expected_focus_generation: None,2671                    expected_proposal_id: None,2672                },2673            )2674            .unwrap();2675        assert_eq!(2676            proposal.snapshot.focus_proposal.as_ref().unwrap().requester,2677            second.initial_snapshot.self_presence_id2678        );2679        let proposal_revision = proposal.snapshot.revision;2680        let error = sessions2681            .apply_focus_command(2682                endpoint(3),2683                third.session_id,2684                &FocusCommandV2 {2685                    request_id: 3,2686                    action: FocusCommandActionV2::Request,2687                    slot: ControllerSlot::ZERO,2688                    expected_revision: proposal_revision,2689                    expected_focus_generation: None,2690                    expected_proposal_id: None,2691                },2692            )2693            .unwrap_err();2694        assert!(error.to_string().contains("pending handoff proposal"));2695        assert_eq!(2696            sessions2697                .v2_revision(endpoint(2), second.session_id)2698                .unwrap(),2699            proposal_revision2700        );2701    }27022703    #[test]2704    fn authorization_reduction_blocks_focus_until_neutralization_completes() {2705        let sessions = Arc::new(SessionRegistry::default());2706        let remote = endpoint(1);2707        let handle = "viewer-0000000000000001".to_owned();2708        let media = sessions2709            .claim_v2_authorized(2710                remote,2711                [4; 16],2712                AuthorizedViewer {2713                    handle: handle.clone(),2714                    grants: InvitationGrants::ALL,2715                    authorization_revision: 1,2716                    committed_revision: 1,2717                },2718            )2719            .unwrap();2720        let granted = sessions2721            .apply_focus_command(2722                remote,2723                media.session_id,2724                &FocusCommandV2 {2725                    request_id: 1,2726                    action: FocusCommandActionV2::Request,2727                    slot: ControllerSlot::ZERO,2728                    expected_revision: 1,2729                    expected_focus_generation: None,2730                    expected_proposal_id: None,2731                },2732            )2733            .unwrap();2734        assert_eq!(granted.snapshot.revision, 2);27352736        let effect = sessions2737            .apply_authorization_mutation(&AuthorizationMutation {2738                handle,2739                peer: remote,2740                previous_grants: InvitationGrants::ALL,2741                current_grants: Some(InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD)),2742                authorization_revision: 2,2743                committed_revision: 2,2744            })2745            .unwrap();2746        assert!(effect.neutralize_input);2747        assert!(2748            sessions2749                .apply_focus_command(2750                    remote,2751                    media.session_id,2752                    &FocusCommandV2 {2753                        request_id: 2,2754                        action: FocusCommandActionV2::Request,2755                        slot: ControllerSlot::ZERO,2756                        expected_revision: 3,2757                        expected_focus_generation: None,2758                        expected_proposal_id: None,2759                    },2760                )2761                .unwrap_err()2762                .to_string()2763                .contains("still neutralizing")2764        );2765        sessions2766            .complete_authorization_neutralization(remote, 2, effect.focus_transition_id)2767            .unwrap();2768        assert!(2769            sessions2770                .apply_focus_command(2771                    remote,2772                    media.session_id,2773                    &FocusCommandV2 {2774                        request_id: 3,2775                        action: FocusCommandActionV2::Request,2776                        slot: ControllerSlot::ZERO,2777                        expected_revision: 4,2778                        expected_focus_generation: None,2779                        expected_proposal_id: None,2780                    },2781                )2782                .is_ok()2783        );2784    }27852786    #[test]2787    fn legacy_mode_rejects_v2_coexistence() {2788        let sessions = Arc::new(SessionRegistry::default());2789        let legacy = sessions2790            .claim(endpoint(1), [1; 16], InvitationGrants::ALL)2791            .unwrap();2792        assert!(2793            sessions2794                .claim_v2(endpoint(2), [2; 16], InvitationGrants::ALL)2795                .is_err()2796        );2797        drop(legacy);2798        assert!(2799            sessions2800                .claim_v2(endpoint(2), [2; 16], InvitationGrants::ALL)2801                .is_ok()2802        );2803    }28042805    fn test_moq_broadcast() -> (BroadcastProducer, BroadcastConsumer) {2806        let producer = Broadcast::new().produce();2807        let consumer = producer.consume();2808        (producer, consumer)2809    }28102811    fn test_subscription(2812        host: &iroh::SecretKey,2813        remote: EndpointId,2814        generation_id: u64,2815        issued_at_unix: u64,2816        expires_at_unix: u64,2817    ) -> SignedSubscriptionCapability {2818        SignedSubscriptionCapability::issue(2819            sigil_protocol::SubscriptionClaims::new(2820                *host.public().as_bytes(),2821                generation_id,2822                *remote.as_bytes(),2823                SubscriptionTracks::VIDEO_H264,2824                1,2825                issued_at_unix,2826                expires_at_unix,2827                [7; 32],2828                1,2829            )2830            .unwrap(),2831            &host.to_bytes(),2832        )2833        .unwrap()2834    }28352836    #[test]2837    fn moq_attachment_requires_exact_active_control_remote_and_is_single_use() {2838        let sessions = Arc::new(SessionRegistry::default());2839        assert!(sessions.claim_moq(endpoint(1)).is_err());2840        let lease = sessions2841            .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)2842            .unwrap();2843        let (_producer, consumer) = test_moq_broadcast();2844        let _wait = sessions2845            .expect_moq(2846                endpoint(1),2847                lease.session_id,2848                media_moq_broadcast_name(lease.session_id).unwrap(),2849                consumer,2850            )2851            .unwrap();28522853        // A wrong peer cannot consume the exact pending attachment.2854        assert!(sessions.claim_moq(endpoint(2)).is_err());2855        let attachment = sessions.claim_moq(endpoint(1)).unwrap();2856        assert_eq!(attachment.session_id, lease.session_id);2857        assert_eq!(2858            attachment.broadcast_name,2859            media_moq_broadcast_name(lease.session_id).unwrap()2860        );2861        // The pending token was atomically consumed before the MoQ handshake.2862        assert!(sessions.claim_moq(endpoint(1)).is_err());2863    }28642865    // Regression: `claim_moq` compared the capability's host field against2866    // itself, so the host-binding leg of `verify_binding` was a no-op. It only2867    // matters once a capability can arrive from anywhere but the host's own2868    // issuance path, which is exactly what the relay design contemplates.2869    #[test]2870    fn v2_moq_attachment_rejects_a_capability_minted_by_another_host() {2871        let host = iroh::SecretKey::from_bytes(&[9; 32]);2872        let impostor = iroh::SecretKey::from_bytes(&[31; 32]);2873        let remote = endpoint(1);2874        let sessions = Arc::new(SessionRegistry::default());2875        let lease = sessions2876            .claim_v2(remote, [1; 16], InvitationGrants::VIEW)2877            .unwrap();2878        let forged = test_subscription(&impostor, remote, lease.session_id, 100, 200);2879        let (_producer, consumer) = test_moq_broadcast();2880        assert!(2881            sessions2882                .expect_moq_v2(2883                    remote,2884                    lease.session_id,2885                    media_moq_broadcast_name(lease.session_id).unwrap(),2886                    consumer,2887                    forged,2888                    *host.public().as_bytes(),2889                )2890                .is_err(),2891            "a capability signed by another host must never be attached"2892        );2893    }28942895    #[test]2896    fn v2_moq_attachment_requires_a_fresh_endpoint_bound_subscription() {2897        let host = iroh::SecretKey::from_bytes(&[9; 32]);2898        let remote = endpoint(1);2899        let sessions = Arc::new(SessionRegistry::default());2900        let lease = sessions2901            .claim_v2(remote, [1; 16], InvitationGrants::VIEW)2902            .unwrap();2903        let capability = test_subscription(&host, remote, lease.session_id, 100, 200);2904        let (_producer, consumer) = test_moq_broadcast();2905        let _wait = sessions2906            .expect_moq_v2(2907                remote,2908                lease.session_id,2909                media_moq_broadcast_name(lease.session_id).unwrap(),2910                consumer,2911                capability,2912                *host.public().as_bytes(),2913            )2914            .unwrap();2915        assert!(sessions.claim_moq_at(endpoint(2), 150).is_err());2916        assert!(sessions.claim_moq_at(remote, 150).is_ok());2917        assert!(sessions.claim_moq_at(remote, 150).is_err());29182919        let expired_sessions = Arc::new(SessionRegistry::default());2920        let expired_lease = expired_sessions2921            .claim_v2(remote, [2; 16], InvitationGrants::VIEW)2922            .unwrap();2923        let expired = test_subscription(&host, remote, expired_lease.session_id, 100, 200);2924        let (_producer, consumer) = test_moq_broadcast();2925        let _wait = expired_sessions2926            .expect_moq_v2(2927                remote,2928                expired_lease.session_id,2929                media_moq_broadcast_name(expired_lease.session_id).unwrap(),2930                consumer,2931                expired,2932                *host.public().as_bytes(),2933            )2934            .unwrap();2935        assert!(expired_sessions.claim_moq_at(remote, 201).is_err());2936    }29372938    #[test]2939    fn competing_moq_connections_cannot_both_claim_one_control_attachment() {2940        let sessions = Arc::new(SessionRegistry::default());2941        let lease = sessions2942            .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)2943            .unwrap();2944        let (_producer, consumer) = test_moq_broadcast();2945        let _wait = sessions2946            .expect_moq(2947                endpoint(1),2948                lease.session_id,2949                media_moq_broadcast_name(lease.session_id).unwrap(),2950                consumer,2951            )2952            .unwrap();2953        let barrier = Arc::new(std::sync::Barrier::new(3));2954        let contenders = (0..2)2955            .map(|_| {2956                let sessions = Arc::clone(&sessions);2957                let barrier = Arc::clone(&barrier);2958                std::thread::spawn(move || {2959                    barrier.wait();2960                    sessions.claim_moq(endpoint(1)).is_ok()2961                })2962            })2963            .collect::<Vec<_>>();2964        barrier.wait();2965        let claimed = contenders2966            .into_iter()2967            .map(|thread| usize::from(thread.join().unwrap()))2968            .sum::<usize>();2969        assert_eq!(claimed, 1);2970    }29712972    #[tokio::test]2973    async fn releasing_control_clears_an_unclaimed_moq_attachment() {2974        let sessions = Arc::new(SessionRegistry::default());2975        let lease = sessions2976            .claim(endpoint(1), [1; 16], InvitationGrants::VIEW)2977            .unwrap();2978        let (_producer, consumer) = test_moq_broadcast();2979        let wait = sessions2980            .expect_moq(2981                endpoint(1),2982                lease.session_id,2983                media_moq_broadcast_name(lease.session_id).unwrap(),2984                consumer,2985            )2986            .unwrap();2987        drop(lease);2988        assert!(sessions.claim_moq(endpoint(1)).is_err());2989        assert!(wait.attached.await.is_err());2990        assert!(wait.closed.await.is_err());2991    }29922993    #[test]2994    fn feedback_attaches_only_to_exact_active_view_session() {2995        let sessions = Arc::new(SessionRegistry::default());2996        let nonce = [9; 16];2997        assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());29982999        let no_view = sessions3000            .claim(endpoint(1), nonce, InvitationGrants::GAMEPAD)3001            .unwrap();3002        assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());3003        drop(no_view);30043005        let media = sessions3006            .claim(endpoint(1), nonce, InvitationGrants::VIEW)3007            .unwrap();3008        assert!(sessions.claim_feedback(endpoint(2), nonce).is_err());3009        assert!(sessions.claim_feedback(endpoint(1), [8; 16]).is_err());3010        let feedback = sessions.claim_feedback(endpoint(1), nonce).unwrap();3011        assert_eq!(feedback.session_id, media.session_id);3012        assert!(feedback.encoder_control.is_none());3013        assert!(sessions.claim_feedback(endpoint(1), nonce).is_err());30143015        drop(media);3016        assert!(3017            sessions3018                .claim(endpoint(2), nonce, InvitationGrants::VIEW)3019                .is_err(),3020            "feedback teardown must keep the draining session isolated"3021        );3022        drop(feedback);3023        assert!(3024            sessions3025                .claim(endpoint(2), nonce, InvitationGrants::VIEW)3026                .is_ok()3027        );3028    }30293030    #[test]3031    fn adaptive_encoder_proposals_are_bound_to_the_exact_active_generation() {3032        let sessions = Arc::new(SessionRegistry::default());3033        let remote = endpoint(1);3034        let media = sessions3035            .claim(remote, [3; 16], InvitationGrants::VIEW)3036            .unwrap();3037        let harness = crate::source::EncoderControlTestHarness::new();30383039        assert!(3040            sessions3041                .install_encoder_control(3042                    endpoint(2),3043                    media.session_id,3044                    Some(harness.control.clone())3045                )3046                .is_err()3047        );3048        sessions3049            .install_encoder_control(remote, media.session_id, Some(harness.control.clone()))3050            .unwrap();3051        let feedback = sessions.claim_feedback(remote, [3; 16]).unwrap();3052        assert!(feedback.encoder_control.is_some());3053        let proposal = sessions3054            .propose_adaptive_encoder_update(remote, media.session_id, 8_000, true)3055            .unwrap()3056            .unwrap();3057        assert_eq!(proposal.target_kbps, 8_000);3058        assert!(proposal.force_keyframe_revision > Some(proposal.bitrate_revision));30593060        let old_session_id = media.session_id;3061        drop(media);3062        assert!(3063            sessions3064                .propose_adaptive_encoder_update(remote, old_session_id, 7_000, false)3065                .is_err(),3066            "a draining generation must not issue another encoder proposal"3067        );3068        drop(feedback);3069    }30703071    #[test]3072    fn pending_handshakes_are_bounded() {3073        let sessions = SessionRegistry::default();3074        let capacity = MAX_PENDING_HANDSHAKES * usize::from(crate::config::DEFAULT_MAX_VIEWERS);3075        let permits: Vec<_> = (0..capacity)3076            .map(|_| sessions.pending_handshakes.try_acquire().unwrap())3077            .collect();3078        assert!(sessions.pending_handshakes.try_acquire().is_err());3079        drop(permits);3080        assert!(sessions.pending_handshakes.try_acquire().is_ok());3081    }30823083    #[test]3084    fn per_viewer_reconnect_keyframe_and_feedback_rates_are_bounded() {3085        let sessions = Arc::new(SessionRegistry::new(1));3086        let remote = endpoint(1);3087        let mut lease = sessions3088            .claim_v2_authorized(3089                remote,3090                [1; 16],3091                authorized_viewer("viewer-one", InvitationGrants::ALL),3092            )3093            .unwrap();3094        for nonce in 2..=MAX_ADMISSIONS_PER_WINDOW {3095            lease = sessions3096                .claim_v2_authorized(3097                    remote,3098                    [u8::try_from(nonce).unwrap(); 16],3099                    authorized_viewer("viewer-one", InvitationGrants::ALL),3100                )3101                .unwrap();3102        }3103        assert!(3104            sessions3105                .claim_v2_authorized(3106                    remote,3107                    [99; 16],3108                    authorized_viewer("viewer-one", InvitationGrants::ALL),3109                )3110                .unwrap_err()3111                .to_string()3112                .contains("reconnect rate limit")3113        );31143115        for _ in 0..MAX_KEYFRAME_REQUESTS_PER_WINDOW {3116            sessions3117                .admit_v2_keyframe_request(remote, lease.session_id)3118                .unwrap();3119        }3120        assert!(3121            sessions3122                .admit_v2_keyframe_request(remote, lease.session_id)3123                .unwrap_err()3124                .to_string()3125                .contains("keyframe request rate limit")3126        );31273128        let active_nonce = [12; 16];3129        for _ in 0..MAX_FEEDBACK_CLAIMS_PER_WINDOW {3130            let feedback = sessions.claim_feedback(remote, active_nonce).unwrap();3131            drop(feedback);3132        }3133        assert!(sessions.claim_feedback(remote, active_nonce).is_err());3134    }31353136    #[test]3137    fn runtime_status_reports_bounded_multi_viewer_focus_without_peer_keys() {3138        let sessions = Arc::new(SessionRegistry::new(3));3139        let first = sessions3140            .claim_v2_authorized(3141                endpoint(1),3142                [1; 16],3143                authorized_viewer("viewer-one", InvitationGrants::ALL),3144            )3145            .unwrap();3146        let second = sessions3147            .claim_v2_authorized(3148                endpoint(2),3149                [2; 16],3150                authorized_viewer("viewer-two", InvitationGrants::VIEW),3151            )3152            .unwrap();3153        let status = sessions.runtime_status();3154        assert_eq!(status.mode, "moq_multi_viewer");3155        assert_eq!(status.active_viewers, 2);3156        assert_eq!(status.configured_capacity, 3);3157        assert_eq!(status.focus_occupied, Some(false));3158        sessions3159            .apply_focus_command(3160                endpoint(1),3161                first.session_id,3162                &FocusCommandV2 {3163                    request_id: 1,3164                    action: FocusCommandActionV2::Request,3165                    slot: ControllerSlot::ZERO,3166                    expected_revision: second.initial_snapshot.revision,3167                    expected_focus_generation: None,3168                    expected_proposal_id: None,3169                },3170            )3171            .unwrap();3172        assert_eq!(sessions.runtime_status().focus_occupied, Some(true));3173    }31743175    #[test]3176    fn session_substreams_inherit_the_exact_enrollment_grant() {3177        let sessions = Arc::new(SessionRegistry::default());3178        let grants = InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD);3179        let media = sessions.claim(endpoint(1), [3; 16], grants).unwrap();3180        let input = sessions.claim_input(endpoint(1), [3; 16]).unwrap();3181        let audio = sessions.claim_audio(endpoint(1), [3; 16]).unwrap();3182        assert_eq!(input.grants, grants);3183        assert_eq!(audio.grants, grants);3184        drop(input);3185        drop(audio);3186        drop(media);3187    }31883189    #[test]3190    fn audio_claim_requires_the_active_remote_and_nonce() {3191        let sessions = Arc::new(SessionRegistry::default());3192        let media = sessions3193            .claim(endpoint(1), [9; 16], InvitationGrants::ALL)3194            .unwrap();3195        assert!(sessions.claim_audio(endpoint(2), [9; 16]).is_err());3196        assert!(sessions.claim_audio(endpoint(1), [8; 16]).is_err());3197        let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();3198        drop(media);3199        assert!(3200            sessions3201                .claim(endpoint(2), [0; 16], InvitationGrants::ALL)3202                .is_err()3203        );3204        drop(audio);3205        assert!(3206            sessions3207                .claim(endpoint(2), [0; 16], InvitationGrants::ALL)3208                .is_ok()3209        );3210    }32113212    #[test]3213    fn media_and_audio_leases_share_one_session_clock() {3214        let sessions = Arc::new(SessionRegistry::default());3215        let media = sessions3216            .claim(endpoint(1), [9; 16], InvitationGrants::ALL)3217            .unwrap();3218        let audio = sessions.claim_audio(endpoint(1), [9; 16]).unwrap();3219        assert_eq!(media.session_clock, audio.session_clock);3220    }3221}