- commit
- da393bb1abdff37f3970cb5c5ac4969d1527aad6
- parent
- 19a847649deb04bc1ae156fa5c9b13a3e1c21dc0
- author
- Dami <dami.etoile@gmail.com>
@@ -8,6 +8,7 @@ use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};
88 use anyhow::{Context, Result, ensure};99 use iroh::EndpointId;1010 use serde::{Deserialize, Serialize};11+use sha2::{Digest, Sha256};1112 use sigil_protocol::{1213 INVITATION_CLOCK_SKEW_SECS, InvitationClaims, InvitationGrants, MAX_INVITATION_TTL_SECS,1314 SignedInvitation,@@ -1125,7 +1126,7 @@ impl AuthorizationPolicy {
11251126 "unrestricted development modes do not accept invitations"11261127 );11271128 Ok(AuthorizedViewer {1128- handle: "viewer-0000000000000000".to_owned(),1129+ handle: development_viewer_handle(remote),11291130 grants: InvitationGrants::ALL,11301131 authorization_revision: 1,11311132 committed_revision: 1,@@ -1145,6 +1146,16 @@ impl AuthorizationPolicy {
11451146 }11461147 }11471148 1149+fn development_viewer_handle(remote: EndpointId) -> String {1150+ let digest =1151+ Sha256::digest([b"goq-development-viewer-v1\0".as_slice(), remote.as_bytes()].concat());1152+ let suffix = digest[..VIEWER_HANDLE_RANDOM_BYTES]1153+ .iter()1154+ .map(|byte| format!("{byte:02x}"))1155+ .collect::<String>();1156+ format!("viewer-{suffix}")1157+}1158+11481159 pub fn unix_timestamp_now() -> Result<u64> {11491160 Ok(SystemTime::now()11501161 .duration_since(UNIX_EPOCH)@@ -416,6 +416,7 @@ mod tests {
416416 InputBackend::initialize(&HostConfig {417417 identity_path: state_path.join("unused-identity"),418418 state_path: state_path.to_path_buf(),419+ max_viewers: crate::config::DEFAULT_MAX_VIEWERS,419420 source: VideoSource::TestPattern,420421 width: Some(1280),421422 height: Some(800),@@ -82,6 +82,13 @@ struct Args {
8282 /// Exercise the explicit revisioned control/2 and focus-bound input/2 path.8383 8484 control_v2: bool,85+ /// Join control/2 as a view-only exercise without requesting slot-0 focus.86+ 87+ spectator: bool,88+ /// End focused v2 proof by disconnecting instead of issuing a release89+ /// against a roster revision that may advance while spectators join.90+ 91+ disconnect_with_focus: bool,8592 /// Request a configured recovery keyframe after three accepted frames,8693 /// then prove no delta history is delivered before the recovery barrier.8794 @@ -751,6 +758,8 @@ async fn run_control_v2_smoke(
751758 frames: u32,752759 timeout_seconds: u64,753760 expected_size: Option<(u16, u16)>,761+ spectator: bool,762+ disconnect_with_focus: bool,754763 ) -> Result<()> {755764 let control_connection = endpoint756765 .connect(address.clone(), CONTROL_ALPN_V2)@@ -790,9 +799,15 @@ async fn run_control_v2_smoke(
790799 .snapshot791800 .context("host omitted initial control v2 snapshot")?;792801 ensure!(793- initial.revision > 0 && matches!(initial.focus, FocusStateV2::Vacant { .. }),794- "initial control v2 snapshot was not a revisioned vacant slot"802+ initial.revision > 0,803+ "initial control v2 snapshot was not revisioned"795804 );805+ if !spectator {806+ ensure!(807+ matches!(initial.focus, FocusStateV2::Vacant { .. }),808+ "initial control v2 snapshot did not advertise a vacant slot"809+ );810+ }796811 let generation_id = initial.media.generation_id;797812 let broadcast_name = initial.media.broadcast_name.clone();798813 @@ -813,6 +828,43 @@ async fn run_control_v2_smoke(
813828 }),814829 )815830 .await?;831+ if spectator {832+ let mut accepted = 0_u32;833+ let mut last_sequence = None;834+ while accepted < frames {835+ let outcome = tokio::time::timeout(Duration::from_secs(timeout_seconds), media.next())836+ .await837+ .context("timed out waiting for spectator media")??838+ .context("spectator media ended before requested frames")?;839+ let MoqProbeOutcome::Frame { frame, .. } = outcome else {840+ continue;841+ };842+ if let Some((width, height)) = expected_size {843+ ensure!(844+ frame.header.width == width && frame.header.height == height,845+ "spectator media dimensions did not match {width}x{height}"846+ );847+ }848+ if let Some(previous) = last_sequence {849+ ensure!(850+ frame.header.sequence > previous,851+ "spectator media sequence regressed"852+ );853+ }854+ last_sequence = Some(frame.header.sequence);855+ accepted += 1;856+ }857+ control_connection.close(0_u32.into(), b"spectator probe complete");858+ println!("probe=ok");859+ println!("transport=iroh-moq");860+ println!("control_alpn=sigil/control/2");861+ println!("media_generation_id={generation_id}");862+ println!("media_broadcast_name={broadcast_name}");863+ println!("roster_viewers={}", initial.viewers.len());864+ println!("frames={accepted}");865+ println!("spectator=ok");866+ return Ok(());867+ }816868 let focus_request = FocusCommandV2 {817869 request_id: 1,818870 action: FocusCommandActionV2::Request,@@ -924,26 +976,32 @@ async fn run_control_v2_smoke(
924976 accepted += 1;925977 }926978 927- let release = FocusCommandV2 {928- request_id: 2,929- action: FocusCommandActionV2::Release,930- slot: ControllerSlot::ZERO,931- expected_revision: focused.revision,932- expected_focus_generation: Some(focus_generation),979+ let released = if disconnect_with_focus {980+ None981+ } else {982+ let release = FocusCommandV2 {983+ request_id: 2,984+ action: FocusCommandActionV2::Release,985+ slot: ControllerSlot::ZERO,986+ expected_revision: focused.revision,987+ expected_focus_generation: Some(focus_generation),988+ };989+ write_client_control_v2(990+ &mut control_send,991+ &ClientControlEnvelopeV2::Focus { command: release },992+ )993+ .await994+ .context("releasing control v2 focus")?;995+ Some(996+ read_v2_snapshot_after(997+ &mut control_recv,998+ focused.revision,999+ timeout_seconds,1000+ |snapshot| matches!(snapshot.focus, FocusStateV2::Vacant { .. }),1001+ )1002+ .await?,1003+ )9331004 };934- write_client_control_v2(935- &mut control_send,936- &ClientControlEnvelopeV2::Focus { command: release },937- )938- .await939- .context("releasing control v2 focus")?;940- let released = read_v2_snapshot_after(941- &mut control_recv,942- focused.revision,943- timeout_seconds,944- |snapshot| matches!(snapshot.focus, FocusStateV2::Vacant { .. }),945- )946- .await?;9471005 input_send.finish().context("finishing input v2 stream")?;9481006 input_connection.close(0_u32.into(), b"focus released");9491007 control_connection.close(0_u32.into(), b"probe complete");@@ -963,8 +1021,13 @@ async fn run_control_v2_smoke(
9631021 println!("focus_grant_revision={}", focused.revision);9641022 println!("focus_generation={focus_generation}");9651023 println!("slot_0_input=ok");966- println!("focus_release_revision={}", released.revision);967- println!("focus_release=ok");1024+ if let Some(released) = released {1025+ println!("focus_release_revision={}", released.revision);1026+ println!("focus_release=ok");1027+ } else {1028+ println!("focus_release_revision=disconnect");1029+ println!("focus_release=disconnect");1030+ }9681031 Ok(())9691032 }9701033 @@ -1598,6 +1661,8 @@ async fn main() -> Result<()> {
15981661 args.frames,15991662 args.timeout_seconds,16001663 args.expect_size,1664+ args.spectator,1665+ args.disconnect_with_focus,16011666 )16021667 .await;16031668 }@@ -12,6 +12,8 @@ use sha2::{Digest, Sha256};
1212 use std::os::unix::fs::{MetadataExt, OpenOptionsExt, PermissionsExt};1313 1414 pub const MAX_CONFIG_BYTES: u64 = 64 * 1024;15+pub const DEFAULT_MAX_VIEWERS: u8 = 3;16+pub const MAX_VIEWERS: u8 = 8;1517 const REVISION_PREFIX: &str = "sha256:";1618 1719 @@ -120,6 +122,10 @@ fn default_input_mode() -> InputMode {
120122 InputMode::Disabled121123 }122124 125+const fn default_max_viewers() -> u8 {126+ DEFAULT_MAX_VIEWERS127+}128+123129 fn default_ffmpeg() -> PathBuf {124130 PathBuf::from("ffmpeg")125131 }@@ -363,6 +369,10 @@ fn validate_pipewire_property(name: &str, value: &str) -> Result<()> {
363369 pub struct HostConfig {364370 pub identity_path: PathBuf,365371 pub state_path: PathBuf,372+ /// Concurrent native-MoQ v2 viewers. Durable enrollment has a separate373+ /// fixed bound and legacy transports remain exclusive.374+ 375+ pub max_viewers: u8,366376 367377 pub source: VideoSource,368378 /// Optional encoded-size override. Gamescope uses its advertised native@@ -468,6 +478,10 @@ impl HostConfig {
468478 !self.state_path.as_os_str().is_empty(),469479 "state_path is required"470480 );481+ ensure!(482+ (1..=MAX_VIEWERS).contains(&self.max_viewers),483+ "max_viewers must be between 1 and {MAX_VIEWERS}"484+ );471485 match (self.width, self.height) {472486 (Some(width), Some(height)) => validate_video_dimensions(width, height)?,473487 (None, None) => {}@@ -695,10 +709,33 @@ state_path = "/tmp/state"
695709 )696710 .unwrap();697711 config.validate().unwrap();712+ assert_eq!(config.max_viewers, DEFAULT_MAX_VIEWERS);698713 assert_eq!(config.configured_dimensions(), None);699714 assert_eq!(config.test_pattern_dimensions().unwrap(), (1_280, 800));700715 }701716 717+ 718+ fn max_viewers_defaults_to_three_and_rejects_values_outside_one_through_eight() {719+ let mut config: HostConfig = toml::from_str(720+ r#"721+identity_path = "/tmp/host.key"722+state_path = "/tmp/state"723+"#,724+ )725+ .unwrap();726+ assert_eq!(config.max_viewers, 3);727+ config.validate().unwrap();728+729+ config.max_viewers = 1;730+ config.validate().unwrap();731+ config.max_viewers = 8;732+ config.validate().unwrap();733+ config.max_viewers = 0;734+ assert!(config.validate().is_err());735+ config.max_viewers = 9;736+ assert!(config.validate().is_err());737+ }738+702739 703740 fn validates_strict_gamescope_pipewire_configuration() {704741 let input = r#"@@ -769,6 +769,7 @@ async fn probe_test_pattern(args: CaptureProbeArgs) -> Result<()> {
769769 let config = HostConfig {770770 identity_path: PathBuf::from("unused-by-capture-probe"),771771 state_path: PathBuf::from("."),772+ max_viewers: config::DEFAULT_MAX_VIEWERS,772773 source: VideoSource::TestPattern,773774 width: Some(width),774775 height: Some(height),@@ -979,7 +980,7 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
979980 };980981 let secret = identity::load(&config.identity_path)?;981982 let host_secret = secret.to_bytes();982- let sessions = Arc::new(SessionRegistry::default());983+ let sessions = Arc::new(SessionRegistry::new(config.max_viewers));983984 let publisher = appliance::RuntimePublisher::start(984985 secret.public(),985986 Arc::clone(&sessions),@@ -1331,6 +1332,7 @@ fn load_serve_config(args: &ServeArgs) -> Result<LoadedServeConfig> {
13311332 config: HostConfig {13321333 identity_path,13331334 state_path,1335+ max_viewers: config::DEFAULT_MAX_VIEWERS,13341336 source: source.into(),13351337 width: Some(args.width),13361338 height: Some(args.height),@@ -733,10 +733,10 @@ pub(super) async fn serve_control_moq_v2(
733733 authorized.grants.contains(InvitationGrants::VIEW),734734 "authorized MoQ control v2 peer lacks view permission"735735 );736- let lease = match sessions.claim_v2_authorized(remote, hello.nonce, authorized) {736+ let mut lease = match sessions.claim_v2_authorized(remote, hello.nonce, authorized) {737737 Ok(lease) => lease,738738 Err(error) => {739- send_rejection_v2(&mut send, "host already has an active client").await?;739+ send_rejection_v2(&mut send, error.to_string()).await?;740740 return Err(error);741741 }742742 };@@ -749,7 +749,7 @@ pub(super) async fn serve_control_moq_v2(
749749 }750750 };751751 let shared = Arc::clone(&generation.shared);752- let initial_snapshot = sessions.bind_v2_generation(752+ let mut initial_snapshot = sessions.bind_v2_generation(753753 remote,754754 lease.session_id,755755 shared.generation_id,@@ -757,6 +757,14 @@ pub(super) async fn serve_control_moq_v2(
757757 Arc::clone(&shared.telemetry),758758 shared.encoder_control.clone(),759759 )?;760+ if lease.replacement_neutralize {761+ input_operations.reset()?;762+ initial_snapshot = sessions.complete_v2_replacement(remote, lease.session_id)?;763+ }764+ // Keep the predecessor's generation lease alive until this viewer has765+ // acquired the shared generation, then retire only that replaced control766+ // stream. Survivors and the producer remain untouched.767+ lease.retire_replaced_viewer();760768 let issued_at_unix = unix_timestamp_now()?;761769 let mut subscription_nonce = [0_u8; 32];762770 getrandom::fill(&mut subscription_nonce).context("generating subscription capability nonce")?;@@ -846,6 +854,10 @@ pub(super) async fn serve_control_moq_v2(
846854 FocusTransitionReasonV2::Disconnected,847855 )? {848856 input_operations.reset()?;857+ sessions.complete_v2_focus_neutralization(858+ lease.session_id,859+ FocusTransitionReasonV2::Disconnected,860+ )?;849861 }850862 drop(lease);851863 generation.release().await;@@ -1,3 +1,4 @@
1+use std::collections::HashMap;12 use std::sync::atomic::{AtomicU64, Ordering};23 use std::sync::{Arc, Mutex};34 use std::time::{Duration, Instant};@@ -25,12 +26,12 @@ const MAX_PENDING_HANDSHAKES: usize = 4;
2526 2627 pub struct SessionRegistry {2728 active: Mutex<Option<ActiveSession>>,28- pending_moq: Mutex<Option<PendingMoqAttachment>>,29+ pending_moq: Mutex<HashMap<(EndpointId, u64), PendingMoqAttachment>>,30+ max_viewers: usize,2931 next_session_id: AtomicU64,3032 next_focus_generation: AtomicU64,3133 authorization_committed_revision: AtomicU64,32- v2_state: Mutex<Option<V2SessionState>>,33- v2_snapshot: tokio::sync::watch::Sender<Option<SessionSnapshotV2>>,34+ v2_state: Mutex<V2SessionState>,3435 pub(super) session_changed: tokio::sync::Notify,3536 pub(super) pending_handshakes: tokio::sync::Semaphore,3637 }@@ -98,13 +99,37 @@ enum SessionMode {
9899 V2Single,99100 }100101 101-102+102103 struct V2SessionState {103- presence_id: ViewerPresenceId,104+ viewers: HashMap<EndpointId, V2ViewerSession>,104105 revision: u64,105106 focus: FocusStateV2,106107 transition_reason: FocusTransitionReasonV2,108+ media: Option<MediaGenerationDescriptorV2>,109+ live_control_leases: usize,110+}111+112+impl Default for V2SessionState {113+ fn default() -> Self {114+ Self {115+ viewers: HashMap::new(),116+ revision: 0,117+ focus: FocusStateV2::Vacant {118+ slot: ControllerSlot::ZERO,119+ },120+ transition_reason: FocusTransitionReasonV2::Initial,121+ media: None,122+ live_control_leases: 0,123+ }124+ }125+}126+127+128+struct V2ViewerSession {129+ session: ActiveSession,130+ presence_id: ViewerPresenceId,107131 authorization_neutralizing: bool,132+ snapshots: tokio::sync::watch::Sender<Option<SessionSnapshotV2>>,108133 }109134 110135 @@ -361,27 +386,47 @@ impl MediaV3Telemetry {
361386 362387 impl Default for SessionRegistry {363388 fn default() -> Self {364- let (v2_snapshot, _receiver) = tokio::sync::watch::channel(None);389+ Self::new(crate::config::DEFAULT_MAX_VIEWERS)390+ }391+}392+393+impl SessionRegistry {394+ pub fn new(max_viewers: u8) -> Self {395+ assert!(396+ (1..=crate::config::MAX_VIEWERS).contains(&max_viewers),397+ "validated max_viewers must be between 1 and {}",398+ crate::config::MAX_VIEWERS399+ );365400 Self {366401 active: Mutex::new(None),367- pending_moq: Mutex::new(None),402+ pending_moq: Mutex::new(HashMap::with_capacity(crate::config::MAX_VIEWERS.into())),403+ max_viewers: usize::from(max_viewers),368404 next_session_id: AtomicU64::new(0),369405 next_focus_generation: AtomicU64::new(0),370406 authorization_committed_revision: AtomicU64::new(1),371- v2_state: Mutex::new(None),372- v2_snapshot,407+ v2_state: Mutex::new(V2SessionState::default()),373408 session_changed: tokio::sync::Notify::new(),374- pending_handshakes: tokio::sync::Semaphore::new(MAX_PENDING_HANDSHAKES),409+ pending_handshakes: tokio::sync::Semaphore::new(410+ MAX_PENDING_HANDSHAKES * usize::from(max_viewers),411+ ),375412 }376413 }377-}378414 379-impl SessionRegistry {380415 pub fn has_session(&self) -> bool {381- self.active416+ if self417+ .active382418 .lock()383419 .expect("session registry poisoned")384420 .is_some()421+ {422+ return true;423+ }424+ !self425+ .v2_state426+ .lock()427+ .expect("v2 session state poisoned")428+ .viewers429+ .is_empty()385430 }386431 387432 pub(super) fn claim(@@ -394,6 +439,14 @@ impl SessionRegistry {
394439 if let Some(current) = active.as_ref() {395440 bail!("host already has active client {}", current.remote);396441 }442+ ensure!(443+ self.v2_state444+ .lock()445+ .expect("v2 session state poisoned")446+ .live_control_leases447+ == 0,448+ "host already has an active multi-viewer generation"449+ );397450 let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;398451 let session_clock = SessionClock::start();399452 let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());@@ -465,30 +518,77 @@ impl SessionRegistry {
465518 == authorized.committed_revision,466519 "viewer admission used a stale authorization revision"467520 );468- let mut active = self.active.lock().expect("session registry poisoned");469- if let Some(current) = active.as_ref() {470- bail!("host already has active client {}", current.remote);471- }521+ ensure!(522+ self.active523+ .lock()524+ .expect("session registry poisoned")525+ .is_none(),526+ "host already has an active legacy client"527+ );528+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");529+ let replacing = state.viewers.contains_key(&remote);530+ ensure!(531+ replacing || state.viewers.len() < self.max_viewers,532+ "multi-viewer capacity of {} is full",533+ self.max_viewers534+ );472535 let session_id = self.next_session_id.fetch_add(1, Ordering::Relaxed) + 1;473536 let session_clock = SessionClock::start();474537 let media_v3_telemetry = Arc::new(MediaV3Telemetry::default());475538 let presence_id = ViewerPresenceId::new(authorized.handle.clone())?;476- let v2 = V2SessionState {477- presence_id,478- revision: 1,479- focus: FocusStateV2::Vacant {480- slot: ControllerSlot::ZERO,481- },482- transition_reason: FocusTransitionReasonV2::Initial,483- authorization_neutralizing: false,484- };539+ ensure!(540+ state541+ .viewers542+ .iter()543+ .all(|(peer, viewer)| { *peer == remote || viewer.presence_id != presence_id }),544+ "viewer presence id is already active for another peer"545+ );546+ let replaced = state.viewers.remove(&remote);547+ let replacement_neutralize = replaced.as_ref().is_some_and(|viewer| {548+ matches!(549+ &state.focus,550+ FocusStateV2::Held {551+ holder,552+ session_id: holder_session_id,553+ ..554+ } if holder == &viewer.presence_id555+ && *holder_session_id == viewer.session.session_id556+ )557+ });558+ if replacement_neutralize {559+ let replaced = replaced.as_ref().expect("replacement viewer disappeared");560+ let FocusStateV2::Held {561+ slot,562+ focus_generation,563+ ..564+ } = state.focus.clone()565+ else {566+ unreachable!("replacement focus match was already checked");567+ };568+ let transition_id = self.next_focus_generation.fetch_add(1, Ordering::Relaxed) + 1;569+ state.focus = FocusStateV2::Neutralizing {570+ slot,571+ former_holder: replaced.presence_id.clone(),572+ former_session_id: replaced.session.session_id,573+ former_focus_generation: focus_generation,574+ transition_id,575+ };576+ state.transition_reason = FocusTransitionReasonV2::Replaced;577+ } else if replacing {578+ state.transition_reason = FocusTransitionReasonV2::Replaced;579+ }580+ let media = state.media.clone().unwrap_or(MediaGenerationDescriptorV2 {581+ generation_id: session_id,582+ broadcast_name: media_generation_moq_broadcast_name(session_id)?,583+ });584+ let (snapshots, _receiver) = tokio::sync::watch::channel(None);485585 let session = ActiveSession {486586 remote,487587 session_id,488588 nonce,489589 session_clock,490- media_generation_id: session_id,491- media_broadcast_name: media_moq_broadcast_name(session_id)?,590+ media_generation_id: media.generation_id,591+ media_broadcast_name: media.broadcast_name,492592 grants: authorized.grants,493593 viewer_handle: Some(authorized.handle),494594 authorization_revision: authorized.authorization_revision,@@ -501,13 +601,23 @@ impl SessionRegistry {
501601 encoder_control: None,502602 mode: SessionMode::V2Single,503603 };504- let snapshot = v2_snapshot_for(&session, &v2)?;505- *active = Some(session);506- self.v2_state507- .lock()508- .expect("v2 session state poisoned")509- .replace(v2);510- self.v2_snapshot.send_replace(Some(snapshot.clone()));604+ state.viewers.insert(605+ remote,606+ V2ViewerSession {607+ session,608+ presence_id,609+ authorization_neutralizing: false,610+ snapshots,611+ },612+ );613+ state.live_control_leases = state614+ .live_control_leases615+ .checked_add(1)616+ .context("v2 control lease count overflowed")?;617+ advance_v2_revision(&mut state)?;618+ publish_v2_snapshots(&state)?;619+ let snapshot = v2_snapshot_for(&state, remote)?;620+ let replaced_snapshots = replaced.map(|viewer| viewer.snapshots);511621 Ok(V2SessionLease {512622 registry: Arc::clone(self),513623 remote,@@ -517,6 +627,8 @@ impl SessionRegistry {
517627 initial_snapshot: snapshot,518628 authorization_revision: authorized.authorization_revision,519629 authorization_committed_revision: authorized.committed_revision,630+ replaced_snapshots,631+ replacement_neutralize,520632 })521633 }522634 @@ -529,33 +641,57 @@ impl SessionRegistry {
529641 telemetry: Arc<MediaV3Telemetry>,530642 encoder_control: Option<EncoderControl>,531643 ) -> Result<SessionSnapshotV2> {532- let mut active = self.active.lock().expect("session registry poisoned");533- let session = active534- .as_mut()535- .filter(|session| {536- session.mode == SessionMode::V2Single537- && session.media_active538- && session.remote == remote539- && session.session_id == session_id540- })644+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");645+ let descriptor = MediaGenerationDescriptorV2 {646+ generation_id,647+ broadcast_name: media_generation_moq_broadcast_name(generation_id)?,648+ };649+ if let Some(current) = &state.media {650+ ensure!(651+ current == &descriptor,652+ "viewer attempted to join a different active media generation"653+ );654+ } else {655+ state.media = Some(descriptor.clone());656+ }657+ let viewer = state658+ .viewers659+ .get_mut(&remote)660+ .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)541661 .context("generation binding does not match the active v2 viewer")?;542662 ensure!(generation_id != 0, "media generation id must be non-zero");543- session.session_clock = session_clock;544- session.media_generation_id = generation_id;545- session.media_broadcast_name = media_generation_moq_broadcast_name(generation_id)?;546- session.media_v3_telemetry = telemetry;547- session.encoder_control = encoder_control;548- let v2_state = self.v2_state.lock().expect("v2 session state poisoned");549- let snapshot = v2_snapshot_for(550- session,551- v2_state552- .as_ref()553- .context("active v2 viewer is missing snapshot state")?,554- )?;555- self.v2_snapshot.send_replace(Some(snapshot.clone()));663+ viewer.session.session_clock = session_clock;664+ viewer.session.media_generation_id = generation_id;665+ viewer.session.media_broadcast_name = descriptor.broadcast_name;666+ viewer.session.media_v3_telemetry = telemetry;667+ viewer.session.encoder_control = encoder_control;668+ publish_v2_snapshots(&state)?;669+ let snapshot = v2_snapshot_for(&state, remote)?;556670 Ok(snapshot)557671 }558672 673+ pub(super) fn complete_v2_replacement(674+ &self,675+ remote: EndpointId,676+ session_id: u64,677+ ) -> Result<SessionSnapshotV2> {678+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");679+ state680+ .viewers681+ .get(&remote)682+ .filter(|viewer| viewer.session.session_id == session_id)683+ .context("replacement completion does not match the active viewer")?;684+ if matches!(state.focus, FocusStateV2::Neutralizing { .. }) {685+ state.focus = FocusStateV2::Vacant {686+ slot: ControllerSlot::ZERO,687+ };688+ state.transition_reason = FocusTransitionReasonV2::Replaced;689+ advance_v2_revision(&mut state)?;690+ publish_v2_snapshots(&state)?;691+ }692+ v2_snapshot_for(&state, remote)693+ }694+559695 pub(super) fn claim_input(560696 self: &Arc<Self>,561697 remote: EndpointId,@@ -590,36 +726,23 @@ impl SessionRegistry {
590726 remote: EndpointId,591727 session_id: u64,592728 ) -> Result<tokio::sync::watch::Receiver<Option<SessionSnapshotV2>>> {593- let active = self.active.lock().expect("session registry poisoned");594- active595- .as_ref()596- .filter(|session| {597- session.mode == SessionMode::V2Single598- && session.media_active599- && session.remote == remote600- && session.session_id == session_id601- })729+ let state = self.v2_state.lock().expect("v2 session state poisoned");730+ let viewer = state731+ .viewers732+ .get(&remote)733+ .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)602734 .context("snapshot subscription does not match the active v2 session")?;603- Ok(self.v2_snapshot.subscribe())735+ Ok(viewer.snapshots.subscribe())604736 }605737 606738 pub(super) fn v2_revision(&self, remote: EndpointId, session_id: u64) -> Result<u64> {607- let active = self.active.lock().expect("session registry poisoned");608- active609- .as_ref()610- .filter(|session| {611- session.mode == SessionMode::V2Single612- && session.media_active613- && session.remote == remote614- && session.session_id == session_id615- })739+ let state = self.v2_state.lock().expect("v2 session state poisoned");740+ state741+ .viewers742+ .get(&remote)743+ .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)616744 .context("revision request does not match the active v2 session")?;617- self.v2_state618- .lock()619- .expect("v2 session state poisoned")620- .as_ref()621- .map(|state| state.revision)622- .context("active v2 session is missing revision state")745+ Ok(state.revision)623746 }624747 625748 pub(crate) fn apply_focus_command(@@ -629,51 +752,50 @@ impl SessionRegistry {
629752 command: &FocusCommandV2,630753 ) -> Result<FocusCommandEffect> {631754 command.validate()?;632- let active = self.active.lock().expect("session registry poisoned");633- let session = active634- .as_ref()635- .filter(|session| {636- session.mode == SessionMode::V2Single637- && session.media_active638- && session.remote == remote639- && session.session_id == session_id755+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");756+ let (presence_id, grants, authorization_neutralizing) = state757+ .viewers758+ .get(&remote)759+ .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)760+ .map(|viewer| {761+ (762+ viewer.presence_id.clone(),763+ viewer.session.grants,764+ viewer.authorization_neutralizing,765+ )640766 })641767 .context("focus command does not match the active v2 session")?;642- let mut v2_state = self.v2_state.lock().expect("v2 session state poisoned");643- let v2 = v2_state644- .as_mut()645- .context("active v2 session is missing focus state")?;646768 ensure!(647- command.expected_revision == v2.revision,769+ command.expected_revision == state.revision,648770 "focus command expected stale revision {} instead of {}",649771 command.expected_revision,650- v2.revision772+ state.revision651773 );652774 653775 let neutralize = match command.action {654776 FocusCommandActionV2::Request => {655777 ensure!(656- !v2.authorization_neutralizing,778+ !authorization_neutralizing,657779 "input authorization is still neutralizing"658780 );659781 ensure!(660- session.grants.contains(InvitationGrants::POINTER_KEYBOARD)661- || session.grants.contains(InvitationGrants::GAMEPAD),782+ grants.contains(InvitationGrants::POINTER_KEYBOARD)783+ || grants.contains(InvitationGrants::GAMEPAD),662784 "viewer is not input-capable"663785 );664786 ensure!(665- matches!(v2.focus, FocusStateV2::Vacant { .. }),666- "controller slot 0 is already occupied"787+ matches!(state.focus, FocusStateV2::Vacant { .. }),788+ "controller slot 0 is occupied; handoff is unavailable"667789 );668790 let focus_generation =669791 self.next_focus_generation.fetch_add(1, Ordering::Relaxed) + 1;670- v2.focus = FocusStateV2::Held {792+ state.focus = FocusStateV2::Held {671793 slot: command.slot,672- holder: v2.presence_id.clone(),794+ holder: presence_id.clone(),673795 session_id,674796 focus_generation,675797 };676- v2.transition_reason = FocusTransitionReasonV2::Requested;798+ state.transition_reason = FocusTransitionReasonV2::Requested;677799 false678800 }679801 FocusCommandActionV2::Release => {@@ -682,29 +804,26 @@ impl SessionRegistry {
682804 .expect("validated focus release carries a generation");683805 ensure!(684806 matches!(685- &v2.focus,807+ &state.focus,686808 FocusStateV2::Held {687809 holder,688810 session_id: holder_session_id,689811 focus_generation,690812 ..691- } if holder == &v2.presence_id813+ } if holder == &presence_id692814 && *holder_session_id == session_id693815 && *focus_generation == expected_focus_generation694816 ),695817 "focus release does not match the current holder generation"696818 );697- v2.focus = FocusStateV2::Vacant { slot: command.slot };698- v2.transition_reason = FocusTransitionReasonV2::Released;819+ state.focus = FocusStateV2::Vacant { slot: command.slot };820+ state.transition_reason = FocusTransitionReasonV2::Released;699821 true700822 }701823 };702- v2.revision = v2703- .revision704- .checked_add(1)705- .context("v2 snapshot revision overflowed")?;706- let snapshot = v2_snapshot_for(session, v2)?;707- self.v2_snapshot.send_replace(Some(snapshot.clone()));824+ advance_v2_revision(&mut state)?;825+ publish_v2_snapshots(&state)?;826+ let snapshot = v2_snapshot_for(&state, remote)?;708827 self.session_changed.notify_waiters();709828 Ok(FocusCommandEffect {710829 snapshot,@@ -718,36 +837,67 @@ impl SessionRegistry {
718837 session_id: u64,719838 reason: FocusTransitionReasonV2,720839 ) -> Result<bool> {721- let active = self.active.lock().expect("session registry poisoned");722- let session = active723- .as_ref()724- .filter(|session| {725- session.mode == SessionMode::V2Single726- && session.remote == remote727- && session.session_id == session_id728- })729- .context("focus invalidation does not match the active v2 session")?;730- let mut v2_state = self.v2_state.lock().expect("v2 session state poisoned");731- let v2 = v2_state732- .as_mut()733- .context("active v2 session is missing focus state")?;734- if !matches!(v2.focus, FocusStateV2::Held { .. }) {840+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");841+ let Some(presence_id) = state842+ .viewers843+ .get(&remote)844+ .filter(|viewer| viewer.session.session_id == session_id)845+ .map(|viewer| viewer.presence_id.clone())846+ else {847+ return Ok(false);848+ };849+ let FocusStateV2::Held {850+ slot,851+ holder,852+ session_id: holder_session_id,853+ focus_generation,854+ } = state.focus.clone()855+ else {856+ return Ok(false);857+ };858+ if holder != presence_id || holder_session_id != session_id {735859 return Ok(false);736860 }737- v2.focus = FocusStateV2::Vacant {738- slot: ControllerSlot::ZERO,861+ state.focus = FocusStateV2::Neutralizing {862+ slot,863+ former_holder: holder,864+ former_session_id: session_id,865+ former_focus_generation: focus_generation,866+ transition_id: self.next_focus_generation.fetch_add(1, Ordering::Relaxed) + 1,739867 };740- v2.transition_reason = reason;741- v2.revision = v2742- .revision743- .checked_add(1)744- .context("v2 snapshot revision overflowed")?;745- self.v2_snapshot746- .send_replace(Some(v2_snapshot_for(session, v2)?));868+ state.transition_reason = reason;869+ advance_v2_revision(&mut state)?;870+ publish_v2_snapshots(&state)?;747871 self.session_changed.notify_waiters();748872 Ok(true)749873 }750874 875+ pub(super) fn complete_v2_focus_neutralization(876+ &self,877+ session_id: u64,878+ reason: FocusTransitionReasonV2,879+ ) -> Result<()> {880+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");881+ if !matches!(882+ state.focus,883+ FocusStateV2::Neutralizing {884+ former_session_id,885+ ..886+ } if former_session_id == session_id887+ ) {888+ return Ok(());889+ }890+ state.focus = FocusStateV2::Vacant {891+ slot: ControllerSlot::ZERO,892+ };893+ state.transition_reason = reason;894+ advance_v2_revision(&mut state)?;895+ publish_v2_snapshots(&state)?;896+ drop(state);897+ self.session_changed.notify_waiters();898+ Ok(())899+ }900+751901 pub fn apply_authorization_mutation(752902 &self,753903 mutation: &AuthorizationMutation,@@ -759,21 +909,16 @@ impl SessionRegistry {
759909 self.authorization_committed_revision760910 .fetch_max(mutation.committed_revision, Ordering::SeqCst);761911 762- let mut active = self.active.lock().expect("session registry poisoned");763- let Some(session) = active.as_mut().filter(|session| {764- session.mode == SessionMode::V2Single765- && session.remote == mutation.peer766- && session.viewer_handle.as_deref() == Some(mutation.handle.as_str())912+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");913+ let Some(viewer) = state.viewers.get(&mutation.peer).filter(|viewer| {914+ viewer.session.viewer_handle.as_deref() == Some(mutation.handle.as_str())767915 }) else {768916 return Ok(AuthorizationSessionEffect::default());769917 };770918 ensure!(771- mutation.authorization_revision > session.authorization_revision,919+ mutation.authorization_revision > viewer.session.authorization_revision,772920 "authorization mutation is stale for the active viewer"773921 );774- session.authorization_revision = mutation.authorization_revision;775- session.authorization_committed_revision = mutation.committed_revision;776-777922 let current_grants = mutation.current_grants.unwrap_or(InvitationGrants::VIEW);778923 let input_reduced = mutation779924 .previous_grants@@ -782,52 +927,82 @@ impl SessionRegistry {
782927 || mutation.previous_grants.contains(InvitationGrants::GAMEPAD)783928 && !current_grants.contains(InvitationGrants::GAMEPAD);784929 785- let mut v2_state = self.v2_state.lock().expect("v2 session state poisoned");786- let v2 = v2_state787- .as_mut()788- .context("active v2 session is missing focus state")?;789- let held_focus = matches!(v2.focus, FocusStateV2::Held { .. });930+ let held_focus = matches!(931+ &state.focus,932+ FocusStateV2::Held {933+ holder,934+ session_id,935+ ..936+ } if holder == &viewer.presence_id && *session_id == viewer.session.session_id937+ );790938 let neutralize_input = input_reduced;791939 let disconnected = mutation.current_grants.is_none();792940 793941 if disconnected {794- session.grants = InvitationGrants::VIEW;795- session.media_active = false;796- session.encoder_control = None;942+ let viewer = state943+ .viewers944+ .remove(&mutation.peer)945+ .expect("validated authorization viewer disappeared");946+ viewer.snapshots.send_replace(None);797947 if held_focus {798- v2.focus = FocusStateV2::Vacant {799- slot: ControllerSlot::ZERO,948+ let FocusStateV2::Held {949+ slot,950+ focus_generation,951+ ..952+ } = state.focus.clone()953+ else {954+ unreachable!("held focus was already checked");955+ };956+ state.focus = FocusStateV2::Neutralizing {957+ slot,958+ former_holder: viewer.presence_id,959+ former_session_id: viewer.session.session_id,960+ former_focus_generation: focus_generation,961+ transition_id: self.next_focus_generation.fetch_add(1, Ordering::Relaxed) + 1,800962 };801- v2.transition_reason = FocusTransitionReasonV2::Revoked;802963 }803964 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");804- if pending.as_ref().is_some_and(|attachment| {805- attachment.remote == mutation.peer && attachment.session_id == session.session_id806- }) {807- *pending = None;808- }809- *v2_state = None;810- self.v2_snapshot.send_replace(None);965+ pending.remove(&(mutation.peer, viewer.session.session_id));966+ state.transition_reason = FocusTransitionReasonV2::Revoked;811967 } else if let Some(grants) = mutation.current_grants {812- session.grants = grants;968+ let held_focus_details = held_focus.then(|| {969+ let FocusStateV2::Held {970+ slot,971+ focus_generation,972+ ..973+ } = state.focus.clone()974+ else {975+ unreachable!("held focus was already checked");976+ };977+ (slot, focus_generation)978+ });979+ let viewer = state980+ .viewers981+ .get_mut(&mutation.peer)982+ .expect("validated authorization viewer disappeared");983+ viewer.session.authorization_revision = mutation.authorization_revision;984+ viewer.session.authorization_committed_revision = mutation.committed_revision;985+ viewer.session.grants = grants;813986 if input_reduced {814- v2.authorization_neutralizing = true;815- if held_focus {816- v2.focus = FocusStateV2::Vacant {817- slot: ControllerSlot::ZERO,987+ viewer.authorization_neutralizing = true;988+ let former_holder = viewer.presence_id.clone();989+ let former_session_id = viewer.session.session_id;990+ if let Some((slot, focus_generation)) = held_focus_details {991+ state.focus = FocusStateV2::Neutralizing {992+ slot,993+ former_holder,994+ former_session_id,995+ former_focus_generation: focus_generation,996+ transition_id: self.next_focus_generation.fetch_add(1, Ordering::Relaxed)997+ + 1,818998 };819999 }820- v2.transition_reason = FocusTransitionReasonV2::Revoked;821- v2.revision = v2822- .revision823- .checked_add(1)824- .context("v2 snapshot revision overflowed")?;825- self.v2_snapshot826- .send_replace(Some(v2_snapshot_for(session, v2)?));1000+ state.transition_reason = FocusTransitionReasonV2::Revoked;8271001 }8281002 }829- drop(v2_state);830- drop(active);1003+ advance_v2_revision(&mut state)?;1004+ publish_v2_snapshots(&state)?;1005+ drop(state);8311006 self.session_changed.notify_waiters();8321007 Ok(AuthorizationSessionEffect {8331008 disconnected,@@ -840,22 +1015,20 @@ impl SessionRegistry {
8401015 remote: EndpointId,8411016 authorization_revision: u64,8421017 ) -> Result<()> {843- let active = self.active.lock().expect("session registry poisoned");844- let Some(_session) = active.as_ref().filter(|session| {845- session.mode == SessionMode::V2Single846- && session.media_active847- && session.remote == remote848- && session.authorization_revision == authorization_revision849- }) else {850- return Ok(());851- };852- let mut v2_state = self.v2_state.lock().expect("v2 session state poisoned");853- let v2 = v2_state854- .as_mut()855- .context("active v2 session is missing focus state")?;856- v2.authorization_neutralizing = false;857- drop(v2_state);858- drop(active);1018+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");1019+ if let Some(viewer) = state.viewers.get_mut(&remote)1020+ && viewer.session.authorization_revision == authorization_revision1021+ {1022+ viewer.authorization_neutralizing = false;1023+ }1024+ if matches!(state.focus, FocusStateV2::Neutralizing { .. }) {1025+ state.focus = FocusStateV2::Vacant {1026+ slot: ControllerSlot::ZERO,1027+ };1028+ advance_v2_revision(&mut state)?;1029+ publish_v2_snapshots(&state)?;1030+ }1031+ drop(state);8591032 self.session_changed.notify_waiters();8601033 Ok(())8611034 }@@ -868,49 +1041,55 @@ impl SessionRegistry {
8681041 slot: ControllerSlot,8691042 focus_generation: u64,8701043 ) -> Result<InputV2Lease> {871- let mut active = self.active.lock().expect("session registry poisoned");872- let session = active873- .as_mut()874- .filter(|session| {875- session.mode == SessionMode::V2Single876- && session.media_active877- && session.remote == remote878- && session.nonce == nonce879- && session.session_id == session_id1044+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");1045+ let (presence_id, grants, authorization_revision) = state1046+ .viewers1047+ .get(&remote)1048+ .filter(|viewer| {1049+ viewer.session.media_active1050+ && viewer.session.nonce == nonce1051+ && viewer.session.session_id == session_id1052+ })1053+ .map(|viewer| {1054+ (1055+ viewer.presence_id.clone(),1056+ viewer.session.grants,1057+ viewer.session.authorization_revision,1058+ )8801059 })8811060 .context("input v2 connection does not match the active session")?;882- let v2_state = self.v2_state.lock().expect("v2 session state poisoned");883- let v2 = v2_state884- .as_ref()885- .context("active v2 session is missing focus state")?;8861061 ensure!(8871062 matches!(888- &v2.focus,1063+ &state.focus,8891064 FocusStateV2::Held {8901065 holder,8911066 session_id: holder_session_id,8921067 slot: holder_slot,8931068 focus_generation: holder_generation,894- } if holder == &v2.presence_id1069+ } if holder == &presence_id8951070 && *holder_session_id == session_id8961071 && *holder_slot == slot8971072 && *holder_generation == focus_generation8981073 ),8991074 "input v2 connection does not own the authoritative focus generation"9001075 );1076+ let viewer = state1077+ .viewers1078+ .get_mut(&remote)1079+ .expect("validated v2 viewer disappeared");9011080 ensure!(902- !session.input_claimed,1081+ !viewer.session.input_claimed,9031082 "active viewer already has an input stream"9041083 );905- session.input_claimed = true;1084+ viewer.session.input_claimed = true;9061085 Ok(InputV2Lease {9071086 registry: Arc::clone(self),9081087 remote,9091088 session_id,9101089 slot,9111090 focus_generation,912- grants: session.grants,913- authorization_revision: session.authorization_revision,1091+ grants,1092+ authorization_revision,9141093 })9151094 }9161095 @@ -921,30 +1100,24 @@ impl SessionRegistry {
9211100 slot: ControllerSlot,9221101 focus_generation: u64,9231102 ) -> bool {924- let active = self.active.lock().expect("session registry poisoned");925- let Some(session) = active.as_ref().filter(|session| {926- session.mode == SessionMode::V2Single927- && session.media_active928- && session.remote == remote929- && session.session_id == session_id1103+ let state = self.v2_state.lock().expect("v2 session state poisoned");1104+ let Some(viewer) = state.viewers.get(&remote).filter(|viewer| {1105+ viewer.session.media_active && viewer.session.session_id == session_id9301106 }) else {9311107 return false;9321108 };933- let v2_state = self.v2_state.lock().expect("v2 session state poisoned");934- v2_state.as_ref().is_some_and(|v2| {935- matches!(936- &v2.focus,937- FocusStateV2::Held {938- holder,939- session_id: holder_session_id,940- slot: holder_slot,941- focus_generation: holder_generation,942- } if holder == &v2.presence_id943- && *holder_session_id == session.session_id944- && *holder_slot == slot945- && *holder_generation == focus_generation946- )947- })1109+ matches!(1110+ &state.focus,1111+ FocusStateV2::Held {1112+ holder,1113+ session_id: holder_session_id,1114+ slot: holder_slot,1115+ focus_generation: holder_generation,1116+ } if holder == &viewer.presence_id1117+ && *holder_session_id == viewer.session.session_id1118+ && *holder_slot == slot1119+ && *holder_generation == focus_generation1120+ )9481121 }9491122 9501123 pub(super) fn install_encoder_control(@@ -975,14 +1148,31 @@ impl SessionRegistry {
9751148 target_kbps: u32,9761149 force_keyframe: bool,9771150 ) -> Result<Option<AdaptiveEncoderProposal>> {978- let active = self.active.lock().expect("session registry poisoned");979- let session = active1151+ let legacy_control = self1152+ .active1153+ .lock()1154+ .expect("session registry poisoned")9801155 .as_ref()9811156 .filter(|session| {9821157 session.media_active && session.remote == remote && session.session_id == session_id9831158 })984- .context("adaptive encoder update does not match the active media session")?;985- let Some(control) = session.encoder_control.clone() else {1159+ .and_then(|session| session.encoder_control.clone());1160+ let control = legacy_control.or_else(|| {1161+ self.v2_state1162+ .lock()1163+ .expect("v2 session state poisoned")1164+ .viewers1165+ .get(&remote)1166+ .filter(|viewer| {1167+ viewer.session.media_active && viewer.session.session_id == session_id1168+ })1169+ .and_then(|viewer| viewer.session.encoder_control.clone())1170+ });1171+ let Some(control) = control else {1172+ ensure!(1173+ self.is_active(remote, session_id),1174+ "adaptive encoder update does not match the active media session"1175+ );9861176 return Ok(None);9871177 };9881178 let bitrate_revision = control.request_bitrate_kbps(target_kbps)?;@@ -1003,14 +1193,31 @@ impl SessionRegistry {
10031193 session_id: u64,10041194 target: VideoDimensions,10051195 ) -> Result<Option<ResolutionEncoderProposal>> {1006- let active = self.active.lock().expect("session registry poisoned");1007- let session = active1196+ let legacy_control = self1197+ .active1198+ .lock()1199+ .expect("session registry poisoned")10081200 .as_ref()10091201 .filter(|session| {10101202 session.media_active && session.remote == remote && session.session_id == session_id10111203 })1012- .context("resolution update does not match the active media session")?;1013- let Some(control) = session.encoder_control.clone() else {1204+ .and_then(|session| session.encoder_control.clone());1205+ let control = legacy_control.or_else(|| {1206+ self.v2_state1207+ .lock()1208+ .expect("v2 session state poisoned")1209+ .viewers1210+ .get(&remote)1211+ .filter(|viewer| {1212+ viewer.session.media_active && viewer.session.session_id == session_id1213+ })1214+ .and_then(|viewer| viewer.session.encoder_control.clone())1215+ });1216+ let Some(control) = control else {1217+ ensure!(1218+ self.is_active(remote, session_id),1219+ "resolution update does not match the active media session"1220+ );10141221 return Ok(None);10151222 };10161223 let revision = control.request_resolution(target.width, target.height)?;@@ -1027,12 +1234,35 @@ impl SessionRegistry {
10271234 nonce: [u8; 16],10281235 ) -> Result<FeedbackLease> {10291236 let mut active = self.active.lock().expect("session registry poisoned");1030- let session = active1031- .as_mut()1032- .filter(|session| {1033- session.media_active && session.remote == remote && session.nonce == nonce1034- })1237+ if let Some(session) = active.as_mut().filter(|session| {1238+ session.media_active && session.remote == remote && session.nonce == nonce1239+ }) {1240+ ensure!(1241+ session.grants.contains(InvitationGrants::VIEW),1242+ "active Portal session lacks feedback view permission"1243+ );1244+ ensure!(1245+ !session.feedback_claimed,1246+ "active client already has a feedback connection"1247+ );1248+ session.feedback_claimed = true;1249+ return Ok(FeedbackLease {1250+ registry: Arc::clone(self),1251+ remote,1252+ session_id: session.session_id,1253+ telemetry: Arc::clone(&session.media_v3_telemetry),1254+ encoder_control: session.encoder_control.clone(),1255+ authorization_revision: session.authorization_revision,1256+ });1257+ }1258+ drop(active);1259+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");1260+ let viewer = state1261+ .viewers1262+ .get_mut(&remote)1263+ .filter(|viewer| viewer.session.media_active && viewer.session.nonce == nonce)10351264 .context("feedback connection does not match the active media session")?;1265+ let session = &mut viewer.session;10361266 ensure!(10371267 session.grants.contains(InvitationGrants::VIEW),10381268 "active Portal session lacks feedback view permission"@@ -1099,21 +1329,24 @@ impl SessionRegistry {
10991329 .context("MoQ expectation does not match the active control session")?;11001330 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");11011331 ensure!(1102- pending.is_none(),1332+ !pending.contains_key(&(remote, session_id)),11031333 "active control session already expects MoQ"11041334 );11051335 let (attached, attached_rx) = tokio::sync::oneshot::channel();11061336 let (closed, closed_rx) = tokio::sync::oneshot::channel();1107- *pending = Some(PendingMoqAttachment {1108- remote,1109- session_id,1110- broadcast_name,1111- broadcast,1112- attached,1113- closed,1114- telemetry,1115- subscription_capability: None,1116- });1337+ pending.insert(1338+ (remote, session_id),1339+ PendingMoqAttachment {1340+ remote,1341+ session_id,1342+ broadcast_name,1343+ broadcast,1344+ attached,1345+ closed,1346+ telemetry,1347+ subscription_capability: None,1348+ },1349+ );11171350 Ok(MoqAttachmentWait {11181351 attached: attached_rx,11191352 closed: closed_rx,@@ -1128,15 +1361,12 @@ impl SessionRegistry {
11281361 broadcast: BroadcastConsumer,11291362 subscription_capability: SignedSubscriptionCapability,11301363 ) -> Result<MoqAttachmentWait> {1131- let active = self.active.lock().expect("session registry poisoned");1132- let session = active1133- .as_ref()1134- .filter(|session| {1135- session.mode == SessionMode::V2Single1136- && session.media_active1137- && session.remote == remote1138- && session.session_id == session_id1139- })1364+ let state = self.v2_state.lock().expect("v2 session state poisoned");1365+ let session = state1366+ .viewers1367+ .get(&remote)1368+ .filter(|viewer| viewer.session.media_active && viewer.session.session_id == session_id)1369+ .map(|viewer| &viewer.session)11401370 .context("authenticated MoQ expectation does not match the active v2 session")?;11411371 ensure!(11421372 subscription_capability.claims.media_generation_id == session.media_generation_id@@ -1152,21 +1382,24 @@ impl SessionRegistry {
11521382 let telemetry = Arc::clone(&session.media_v3_telemetry);11531383 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");11541384 ensure!(1155- pending.is_none(),1385+ !pending.contains_key(&(remote, session_id)),11561386 "active control session already expects MoQ"11571387 );11581388 let (attached, attached_rx) = tokio::sync::oneshot::channel();11591389 let (closed, closed_rx) = tokio::sync::oneshot::channel();1160- *pending = Some(PendingMoqAttachment {1161- remote,1162- session_id,1163- broadcast_name,1164- broadcast,1165- attached,1166- closed,1167- telemetry,1168- subscription_capability: Some(subscription_capability),1169- });1390+ pending.insert(1391+ (remote, session_id),1392+ PendingMoqAttachment {1393+ remote,1394+ session_id,1395+ broadcast_name,1396+ broadcast,1397+ attached,1398+ closed,1399+ telemetry,1400+ subscription_capability: Some(subscription_capability),1401+ },1402+ );11701403 Ok(MoqAttachmentWait {11711404 attached: attached_rx,11721405 closed: closed_rx,@@ -1182,17 +1415,27 @@ impl SessionRegistry {
11821415 }11831416 11841417 fn claim_moq_at(&self, remote: EndpointId, now_unix: u64) -> Result<ClaimedMoqAttachment> {1185- let active = self.active.lock().expect("session registry poisoned");1186- let session = active1418+ let legacy = self1419+ .active1420+ .lock()1421+ .expect("session registry poisoned")11871422 .as_ref()11881423 .filter(|session| session.media_active && session.remote == remote)1189- .context("MoQ connection does not match the active control session")?;1424+ .cloned();1425+ let v2 = self1426+ .v2_state1427+ .lock()1428+ .expect("v2 session state poisoned")1429+ .viewers1430+ .get(&remote)1431+ .filter(|viewer| viewer.session.media_active)1432+ .map(|viewer| viewer.session.clone());1433+ let session = legacy1434+ .or(v2)1435+ .context("MoQ connection does not match an active control session")?;11901436 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");11911437 let attachment = pending1192- .as_ref()1193- .filter(|attachment| {1194- attachment.remote == remote && attachment.session_id == session.session_id1195- })1438+ .get(&(remote, session.session_id))11961439 .context("active control session is not expecting a MoQ connection")?;11971440 if session.mode == SessionMode::V2Single {11981441 let capability = attachment@@ -1217,7 +1460,7 @@ impl SessionRegistry {
12171460 }12181461 debug_assert_eq!(attachment.remote, session.remote);12191462 let attachment = pending1220- .take()1463+ .remove(&(remote, session.session_id))12211464 .expect("validated pending MoQ attachment disappeared");12221465 Ok(ClaimedMoqAttachment {12231466 session_id: attachment.session_id,@@ -1236,37 +1479,97 @@ impl SessionRegistry {
12361479 && session.session_id == session_id12371480 {12381481 let mut pending = self.pending_moq.lock().expect("MoQ registry poisoned");1239- if pending.as_ref().is_some_and(|attachment| {1240- attachment.remote == remote && attachment.session_id == session_id1241- }) {1242- *pending = None;1243- }1482+ pending.remove(&(remote, session_id));12441483 // Keep the registry occupied until the input handler has observed12451484 // media shutdown and released all held uinput transitions. This12461485 // prevents a reconnect from sharing the device with a draining12471486 // predecessor session.12481487 session.media_active = false;12491488 session.encoder_control = None;1250- if session.mode == SessionMode::V2Single {1251- *self.v2_state.lock().expect("v2 session state poisoned") = None;1252- self.v2_snapshot.send_replace(None);1253- }12541489 if !session.input_claimed && !session.audio_claimed && !session.feedback_claimed {12551490 *active = None;12561491 }12571492 drop(active);12581493 self.session_changed.notify_waiters();1494+ return;1495+ }1496+ drop(active);1497+1498+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");1499+ state.live_control_leases = state.live_control_leases.saturating_sub(1);1500+ let Some(viewer) = state1501+ .viewers1502+ .get(&remote)1503+ .filter(|viewer| viewer.session.session_id == session_id)1504+ else {1505+ return;1506+ };1507+ let presence_id = viewer.presence_id.clone();1508+ let input_claimed = viewer.session.input_claimed;1509+ let held_focus = match &state.focus {1510+ FocusStateV2::Held {1511+ holder,1512+ session_id: holder_session_id,1513+ focus_generation,1514+ slot,1515+ } if holder == &presence_id && *holder_session_id == session_id => {1516+ Some((*slot, *focus_generation))1517+ }1518+ _ => None,1519+ };1520+ let viewer = state1521+ .viewers1522+ .remove(&remote)1523+ .expect("validated v2 viewer disappeared");1524+ viewer.snapshots.send_replace(None);1525+ self.pending_moq1526+ .lock()1527+ .expect("MoQ registry poisoned")1528+ .remove(&(remote, session_id));1529+ if let Some((slot, focus_generation)) = held_focus {1530+ if input_claimed {1531+ state.focus = FocusStateV2::Neutralizing {1532+ slot,1533+ former_holder: presence_id,1534+ former_session_id: session_id,1535+ former_focus_generation: focus_generation,1536+ transition_id: self.next_focus_generation.fetch_add(1, Ordering::Relaxed) + 1,1537+ };1538+ } else {1539+ state.focus = FocusStateV2::Vacant { slot };1540+ }1541+ state.transition_reason = FocusTransitionReasonV2::Disconnected;1542+ }1543+ if state.viewers.is_empty() {1544+ state.media = None;1545+ }1546+ if advance_v2_revision(&mut state).is_ok() {1547+ let _ = publish_v2_snapshots(&state);12591548 }1549+ drop(state);1550+ self.session_changed.notify_waiters();12601551 }12611552 12621553 pub(crate) fn is_active(&self, remote: EndpointId, session_id: u64) -> bool {1263- self.active1554+ if self1555+ .active12641556 .lock()12651557 .expect("session registry poisoned")12661558 .as_ref()12671559 .is_some_and(|active| {12681560 active.media_active && active.remote == remote && active.session_id == session_id12691561 })1562+ {1563+ return true;1564+ }1565+ self.v2_state1566+ .lock()1567+ .expect("v2 session state poisoned")1568+ .viewers1569+ .get(&remote)1570+ .is_some_and(|viewer| {1571+ viewer.session.media_active && viewer.session.session_id == session_id1572+ })12701573 }12711574 12721575 fn release_input(&self, remote: EndpointId, session_id: u64) {@@ -1279,6 +1582,31 @@ impl SessionRegistry {
12791582 if !session.media_active && !session.audio_claimed && !session.feedback_claimed {12801583 *active = None;12811584 }1585+ return;1586+ }1587+ drop(active);1588+ let mut state = self.v2_state.lock().expect("v2 session state poisoned");1589+ if let Some(viewer) = state.viewers.get_mut(&remote)1590+ && viewer.session.session_id == session_id1591+ {1592+ viewer.session.input_claimed = false;1593+ }1594+ let completed_neutralization = matches!(1595+ &state.focus,1596+ FocusStateV2::Neutralizing {1597+ former_session_id,1598+ ..1599+ } if *former_session_id == session_id1600+ );1601+ if completed_neutralization {1602+ state.focus = FocusStateV2::Vacant {1603+ slot: ControllerSlot::ZERO,1604+ };1605+ if advance_v2_revision(&mut state).is_ok() {1606+ let _ = publish_v2_snapshots(&state);1607+ }1608+ drop(state);1609+ self.session_changed.notify_waiters();12821610 }12831611 }12841612 @@ -1305,29 +1633,69 @@ impl SessionRegistry {
13051633 if !session.media_active && !session.input_claimed && !session.audio_claimed {13061634 *active = None;13071635 }1636+ return;1637+ }1638+ drop(active);1639+ if let Some(viewer) = self1640+ .v2_state1641+ .lock()1642+ .expect("v2 session state poisoned")1643+ .viewers1644+ .get_mut(&remote)1645+ && viewer.session.session_id == session_id1646+ {1647+ viewer.session.feedback_claimed = false;13081648 }13091649 }13101650 }13111651 1312-fn v2_snapshot_for(session: &ActiveSession, v2: &V2SessionState) -> Result<SessionSnapshotV2> {1313- debug_assert_eq!(session.mode, SessionMode::V2Single);1314- let input_capable = session.grants.contains(InvitationGrants::POINTER_KEYBOARD)1315- || session.grants.contains(InvitationGrants::GAMEPAD);1652+fn advance_v2_revision(state: &mut V2SessionState) -> Result<()> {1653+ state.revision = state1654+ .revision1655+ .checked_add(1)1656+ .context("v2 snapshot revision overflowed")?;1657+ Ok(())1658+}1659+1660+fn publish_v2_snapshots(state: &V2SessionState) -> Result<()> {1661+ for (remote, viewer) in &state.viewers {1662+ viewer1663+ .snapshots1664+ .send_replace(Some(v2_snapshot_for(state, *remote)?));1665+ }1666+ Ok(())1667+}1668+1669+fn v2_snapshot_for(state: &V2SessionState, remote: EndpointId) -> Result<SessionSnapshotV2> {1670+ let self_viewer = state1671+ .viewers1672+ .get(&remote)1673+ .context("active v2 viewer is missing from snapshot state")?;1674+ let mut viewers = state1675+ .viewers1676+ .values()1677+ .map(|viewer| ViewerPresenceV2 {1678+ presence_id: viewer.presence_id.clone(),1679+ session_id: viewer.session.session_id,1680+ input_capable: viewer1681+ .session1682+ .grants1683+ .contains(InvitationGrants::POINTER_KEYBOARD)1684+ || viewer.session.grants.contains(InvitationGrants::GAMEPAD),1685+ you: viewer.session.remote == remote,1686+ })1687+ .collect::<Vec<_>>();1688+ viewers.sort_by(|left, right| left.presence_id.cmp(&right.presence_id));13161689 let snapshot = SessionSnapshotV2 {1317- revision: v2.revision,1318- self_presence_id: v2.presence_id.clone(),1319- viewers: vec![ViewerPresenceV2 {1320- presence_id: v2.presence_id.clone(),1321- session_id: session.session_id,1322- input_capable,1323- you: true,1324- }],1325- focus: v2.focus.clone(),1326- transition_reason: v2.transition_reason,1327- media: MediaGenerationDescriptorV2 {1328- generation_id: session.media_generation_id,1329- broadcast_name: session.media_broadcast_name.clone(),1330- },1690+ revision: state.revision,1691+ self_presence_id: self_viewer.presence_id.clone(),1692+ viewers,1693+ focus: state.focus.clone(),1694+ transition_reason: state.transition_reason,1695+ media: state.media.clone().unwrap_or(MediaGenerationDescriptorV2 {1696+ generation_id: self_viewer.session.media_generation_id,1697+ broadcast_name: self_viewer.session.media_broadcast_name.clone(),1698+ }),13311699 };13321700 snapshot.validate()?;13331701 Ok(snapshot)@@ -1352,10 +1720,21 @@ pub(crate) struct V2SessionLease {
13521720 pub(super) initial_snapshot: SessionSnapshotV2,13531721 pub(super) authorization_revision: u64,13541722 pub(super) authorization_committed_revision: u64,1723+ replaced_snapshots: Option<tokio::sync::watch::Sender<Option<SessionSnapshotV2>>>,1724+ pub(super) replacement_neutralize: bool,1725+}1726+1727+impl V2SessionLease {1728+ pub(super) fn retire_replaced_viewer(&mut self) {1729+ if let Some(snapshots) = self.replaced_snapshots.take() {1730+ snapshots.send_replace(None);1731+ }1732+ }13551733 }13561734 13571735 impl Drop for V2SessionLease {13581736 fn drop(&mut self) {1737+ self.retire_replaced_viewer();13591738 self.registry.release(self.remote, self.session_id);13601739 }13611740 }@@ -1665,6 +2044,177 @@ mod tests {
16652044 drop(media);16662045 }16672046 2047+ fn authorized_viewer(handle: &str, grants: InvitationGrants) -> AuthorizedViewer {2048+ AuthorizedViewer {2049+ handle: handle.to_owned(),2050+ grants,2051+ authorization_revision: 1,2052+ committed_revision: 1,2053+ }2054+ }2055+2056+ 2057+ fn v2_admits_bounded_viewers_and_publishes_personalized_complete_rosters() {2058+ let sessions = Arc::new(SessionRegistry::new(3));2059+ let telemetry = Arc::new(MediaV3Telemetry::default());2060+ let mut leases = Vec::new();2061+ for index in 1..=3 {2062+ let remote = endpoint(index);2063+ let lease = sessions2064+ .claim_v2_authorized(2065+ remote,2066+ [index; 16],2067+ authorized_viewer(&format!("viewer-{index}"), InvitationGrants::ALL),2068+ )2069+ .unwrap();2070+ sessions2071+ .bind_v2_generation(2072+ remote,2073+ lease.session_id,2074+ 77,2075+ SessionClock::start(),2076+ Arc::clone(&telemetry),2077+ None,2078+ )2079+ .unwrap();2080+ leases.push(lease);2081+ }2082+2083+ for (index, lease) in leases.iter().enumerate() {2084+ let remote = endpoint(u8::try_from(index + 1).unwrap());2085+ let snapshots = sessions2086+ .subscribe_v2_snapshots(remote, lease.session_id)2087+ .unwrap();2088+ let snapshot = snapshots.borrow().clone().unwrap();2089+ assert_eq!(snapshot.viewers.len(), 3);2090+ assert_eq!(snapshot.media.generation_id, 77);2091+ assert_eq!(2092+ snapshot.viewers.iter().filter(|viewer| viewer.you).count(),2093+ 12094+ );2095+ assert_eq!(snapshot.self_viewer().session_id, lease.session_id);2096+ }2097+2098+ assert!(2099+ sessions2100+ .claim_v2_authorized(2101+ endpoint(4),2102+ [4; 16],2103+ authorized_viewer("viewer-4", InvitationGrants::VIEW),2104+ )2105+ .unwrap_err()2106+ .to_string()2107+ .contains("capacity")2108+ );2109+ }2110+2111+ 2112+ fn same_peer_replacement_is_immediate_generation_safe_and_neutralizes_focus() {2113+ let sessions = Arc::new(SessionRegistry::new(2));2114+ let remote = endpoint(1);2115+ let old = sessions2116+ .claim_v2_authorized(2117+ remote,2118+ [1; 16],2119+ authorized_viewer("viewer-one", InvitationGrants::ALL),2120+ )2121+ .unwrap();2122+ let granted = sessions2123+ .apply_focus_command(2124+ remote,2125+ old.session_id,2126+ &FocusCommandV2 {2127+ request_id: 1,2128+ action: FocusCommandActionV2::Request,2129+ slot: ControllerSlot::ZERO,2130+ expected_revision: 1,2131+ expected_focus_generation: None,2132+ },2133+ )2134+ .unwrap();2135+ assert!(granted.snapshot.self_focus_generation().is_some());2136+2137+ let mut replacement = sessions2138+ .claim_v2_authorized(2139+ remote,2140+ [2; 16],2141+ authorized_viewer("viewer-one", InvitationGrants::ALL),2142+ )2143+ .unwrap();2144+ assert!(replacement.replacement_neutralize);2145+ assert!(matches!(2146+ replacement.initial_snapshot.focus,2147+ FocusStateV2::Neutralizing { .. }2148+ ));2149+ let replacement_snapshot = sessions2150+ .complete_v2_replacement(remote, replacement.session_id)2151+ .unwrap();2152+ assert!(matches!(2153+ replacement_snapshot.focus,2154+ FocusStateV2::Vacant { .. }2155+ ));2156+ replacement.retire_replaced_viewer();2157+ drop(old);2158+ assert!(sessions.is_active(remote, replacement.session_id));2159+ assert!(!sessions.is_active(remote, granted.snapshot.self_viewer().session_id));2160+ }2161+2162+ 2163+ fn occupied_focus_rejects_another_viewer_without_queueing_or_mutating_revision() {2164+ let sessions = Arc::new(SessionRegistry::new(3));2165+ let first = sessions2166+ .claim_v2_authorized(2167+ endpoint(1),2168+ [1; 16],2169+ authorized_viewer("viewer-one", InvitationGrants::ALL),2170+ )2171+ .unwrap();2172+ let second = sessions2173+ .claim_v2_authorized(2174+ endpoint(2),2175+ [2; 16],2176+ authorized_viewer("viewer-two", InvitationGrants::ALL),2177+ )2178+ .unwrap();2179+ let revision = second.initial_snapshot.revision;2180+ sessions2181+ .apply_focus_command(2182+ endpoint(1),2183+ first.session_id,2184+ &FocusCommandV2 {2185+ request_id: 1,2186+ action: FocusCommandActionV2::Request,2187+ slot: ControllerSlot::ZERO,2188+ expected_revision: revision,2189+ expected_focus_generation: None,2190+ },2191+ )2192+ .unwrap();2193+ let occupied_revision = sessions2194+ .v2_revision(endpoint(2), second.session_id)2195+ .unwrap();2196+ let error = sessions2197+ .apply_focus_command(2198+ endpoint(2),2199+ second.session_id,2200+ &FocusCommandV2 {2201+ request_id: 2,2202+ action: FocusCommandActionV2::Request,2203+ slot: ControllerSlot::ZERO,2204+ expected_revision: occupied_revision,2205+ expected_focus_generation: None,2206+ },2207+ )2208+ .unwrap_err();2209+ assert!(error.to_string().contains("handoff is unavailable"));2210+ assert_eq!(2211+ sessions2212+ .v2_revision(endpoint(2), second.session_id)2213+ .unwrap(),2214+ occupied_revision2215+ );2216+ }2217+16682218 16692219 fn authorization_reduction_blocks_focus_until_neutralization_completes() {16702220 let sessions = Arc::new(SessionRegistry::default());@@ -1737,7 +2287,7 @@ mod tests {
17372287 request_id: 3,17382288 action: FocusCommandActionV2::Request,17392289 slot: ControllerSlot::ZERO,1740- expected_revision: 3,2290+ expected_revision: 4,17412291 expected_focus_generation: None,17422292 },17432293 )@@ -2001,7 +2551,8 @@ mod tests {
20012551 20022552 fn pending_handshakes_are_bounded() {20032553 let sessions = SessionRegistry::default();2004- let permits: Vec<_> = (0..MAX_PENDING_HANDSHAKES)2554+ let capacity = MAX_PENDING_HANDSHAKES * usize::from(crate::config::DEFAULT_MAX_VIEWERS);2555+ let permits: Vec<_> = (0..capacity)20052556 .map(|_| sessions.pending_handshakes.try_acquire().unwrap())20062557 .collect();20072558 assert!(sessions.pending_handshakes.try_acquire().is_err());@@ -57,10 +57,28 @@ fields survive an update.
5757 5858 Enrollment capacity is authorization state, not a host configuration field.5959 Authorization v2 durably supports at most 32 enrolled viewers, while this60-implementation phase intentionally keeps the concurrent v2 runtime cap at one61-until shared media-generation ownership is available. A later configuration62-revision adds the independently bounded `max_viewers` setting; it does not63-change the durable enrollment ceiling.60+implementation independently bounds concurrent native-MoQ v2 viewers with the61+strict `max_viewers` host setting. The setting defaults to `3`, accepts only62+`1` through `8`, and does not change the durable enrollment ceiling. It is a63+direct TOML setting rather than part of the appliance config v1 editable64+projection, so changing it follows the same reviewed host-configuration and65+service-restart process as other non-managed fields.66+67+Multi-viewer operation is available only through `sigil/control/2` and the68+authenticated native-MoQ generation. Every admitted viewer gets an independent69+control connection, consume-once media attachment, feedback claim, and bounded70+delivery state while sharing one video source, video encoder, optional Opus71+source/encoder, signed broadcast, and keyframe coordinator. `control/1` and72+grouped-v3 remain strictly one-client compatibility modes. An active legacy73+generation rejects v2 admission, and any live v2 generation rejects legacy74+admission, including while old sidecars are draining.75+76+Concurrent capacity and durable enrollment are intentionally separate:77+78+| Bound | Default | Hard limit | Scope |79+|---|---:|---:|---|80+| `max_viewers` | 3 | 8 | Simultaneously connected native-MoQ v2 viewers |81+| `MAX_ENROLLED_VIEWERS` | fixed | 32 | Durable authorization-v2 viewer records |6482 6583 Requests are strict JSON and contain an optimistic-concurrency revision:6684 @@ -141,6 +141,7 @@
141141 <span id="status-text">idle</span>142142 </div>143143 <div class="topbar-right">144+ <span class="viewer-status hidden" id="viewer-status" aria-live="polite">viewers: unavailable</span>144145 <span class="controller-status" id="controller-status">controller: none</span>145146 <span class="node-id" id="node-id-text"></span>146147 <button class="link-btn" data-controller-focus onclick="togglePanel(true)">info</button>@@ -195,6 +196,11 @@
195196 <button class="link-btn panel-action hidden" id="reset-enrollment" data-controller-focus onclick="openEnrollmentReset()">reset enrollment</button>196197 </div>197198 <div class="panel-section" id="stream-section">199+ <section class="roster-section hidden" id="roster-section" aria-labelledby="roster-heading">200+ <h3 id="roster-heading"><span class="hash" aria-hidden="true">#</span> viewers</h3>201+ <p class="roster-summary" id="roster-summary" aria-live="polite">Roster unavailable</p>202+ <ul class="roster-list" id="roster-list" aria-label="Connected viewers"></ul>203+ </section>198204 <h3><span class="hash" aria-hidden="true">#</span> stream</h3>199205 <div class="info-row"><span class="info-label">codec</span><span class="info-value" id="stream-codec">—</span></div>200206 <div class="info-row"><span class="info-label">transport</span><span class="info-value" id="stream-transport">—</span></div>@@ -11,6 +11,7 @@ import { createInputRuntime } from './input-runtime.mjs';
1111 import { createControlRuntime } from './control-runtime.mjs';1212 import { createSessionState } from './session-state.mjs';1313 import { createFocusRuntime } from './focus-runtime.mjs';14+import { renderRoster, rosterPresentation } from './roster-ui.mjs';1415 import { formatVideoDiscardTelemetry } from './frame-stats.mjs';1516 import { networkDiagnosticsPresentation } from './network-diagnostics.mjs';1617 import {@@ -650,7 +651,7 @@ controlRuntime = createControlRuntime({
650651 });651652 sessionStateRuntime = createSessionState({652653 onFocusLoss: () => controlRuntime.exit({ releaseHostFocus: false, resetEscape: true }),653- onChange: () => updateControlUI(),654+ onChange: () => updateSessionUI(),654655 });655656 focusRuntime = createFocusRuntime({656657 invokeCommand: invoke,@@ -887,14 +888,19 @@ function updateControlUI() {
887888 && capabilities.control888889 && (connectionState.controlProtocol !== 'control-v2' || sessionStateRuntime.eligible);889890 const controlling = controlRuntime.setInactiveIfUnavailable(available);890- const controlAction = connectionState.controlProtocol === 'control-v2' && !sessionStateRuntime.focused891- ? 'request control'892- : controlling ? 'controlling' : 'activate control';891+ const v2 = connectionState.controlProtocol === 'control-v2';892+ const occupiedByAnother = v2893+ && sessionStateRuntime.snapshot().snapshot?.focus?.state === 'held'894+ && !sessionStateRuntime.focused;895+ const controlAction = v2 && !sessionStateRuntime.focused896+ ? occupiedByAnother ? 'spectating · request control' : 'request control'897+ : controlling ? 'release control' : 'activate control';893898 el.textContent = available894899 ? `${controlAction} · ${describeInputCapabilities()}${controlling && capabilities.relativePointer ? ' · Ctrl+Alt+Esc to exit' : controlling && capabilities.gamepad ? ' · hold Back+Start to exit' : ''}`895900 : 'view only · input unavailable';896901 el.classList.toggle('active', controlling);897902 el.classList.toggle('disabled', !available);903+ el.classList.toggle('hidden', v2 && !sessionStateRuntime.eligible);898904 el.setAttribute('aria-disabled', available ? 'false' : 'true');899905 document.body.classList.toggle(900906 'native-pointer-control',@@ -911,6 +917,29 @@ function updateControlUI() {
911917 }912918 }913919 920+function updateSessionUI() {921+ updateControlUI();922+ const viewerStatus = document.getElementById('viewer-status');923+ const rosterSection = document.getElementById('roster-section');924+ const rosterSummary = document.getElementById('roster-summary');925+ const rosterList = document.getElementById('roster-list');926+ const session = sessionStateRuntime.snapshot();927+ const visible = session.mode === 'control-v2' && session.snapshot !== null;928+ viewerStatus.classList.toggle('hidden', !visible);929+ rosterSection.classList.toggle('hidden', !visible);930+ if (!visible) {931+ viewerStatus.textContent = 'viewers: unavailable';932+ rosterSummary.textContent = 'Roster unavailable';933+ rosterList.replaceChildren();934+ return;935+ }936+ const presentation = rosterPresentation(session.snapshot);937+ viewerStatus.textContent = presentation.shell;938+ viewerStatus.setAttribute('aria-label', presentation.shell);939+ rosterSummary.textContent = `${presentation.count} connected · ${presentation.focus}`;940+ renderRoster(rosterList, session.snapshot);941+}942+914943 function scaleCoords(clientX, clientY) {915944 const rect = canvas.getBoundingClientRect();916945 return mapCanvasPointToSurface({@@ -0,0 +1,93 @@
1+const MAX_VIEWERS = 8;2+const MAX_HANDLE_BYTES = 32;3+4+function validatePresenceId(value) {5+ if (typeof value !== 'string'6+ || value.length < 17+ || value.length > MAX_HANDLE_BYTES8+ || !/^[A-Za-z0-9-]+$/.test(value)) {9+ throw new Error('viewer presence id must be a bounded opaque handle');10+ }11+ return value;12+}13+14+function compactHandle(handle) {15+ return handle.length <= 18 ? handle : `${handle.slice(0, 8)}...${handle.slice(-6)}`;16+}17+18+export function rosterPresentation(snapshot) {19+ if (!snapshot || !Array.isArray(snapshot.viewers)20+ || snapshot.viewers.length < 1 || snapshot.viewers.length > MAX_VIEWERS) {21+ throw new Error('roster must contain 1..=8 viewers');22+ }23+ const seen = new Set();24+ const holder = snapshot.focus?.state === 'held' ? snapshot.focus.holder : null;25+ const rows = snapshot.viewers.map((viewer) => {26+ const handle = validatePresenceId(viewer?.presence_id);27+ if (seen.has(handle)) throw new Error('roster contains a duplicate viewer');28+ seen.add(handle);29+ const isHolder = holder === handle && snapshot.focus.session_id === viewer.session_id;30+ const input = viewer.input_capable === true ? 'input eligible' : 'view only';31+ const role = isHolder ? 'holds control' : input;32+ return Object.freeze({33+ handle: compactHandle(handle),34+ fullHandle: handle,35+ you: viewer.you === true,36+ inputCapable: viewer.input_capable === true,37+ focusHolder: isHolder,38+ role,39+ adaptive: 'shared stream',40+ });41+ });42+ const self = rows.filter((row) => row.you);43+ if (self.length !== 1 || self[0].fullHandle !== snapshot.self_presence_id) {44+ throw new Error('roster requires one matching you marker');45+ }46+ if (holder !== null && !rows.some((row) => row.focusHolder)) {47+ throw new Error('focus holder is absent from the roster');48+ }49+ const focus = holder === null50+ ? snapshot.focus?.state === 'neutralizing' ? 'control resetting' : 'control available'51+ : self[0].focusHolder ? 'you hold control' : `${compactHandle(holder)} holds control`;52+ return Object.freeze({53+ count: rows.length,54+ rows: Object.freeze(rows),55+ focus,56+ shell: `${rows.length} ${rows.length === 1 ? 'viewer' : 'viewers'} · ${focus}`,57+ });58+}59+60+function appendText(document, parent, className, text) {61+ const element = document.createElement('span');62+ element.className = className;63+ element.textContent = text;64+ parent.appendChild(element);65+ return element;66+}67+68+export function renderRoster(root, snapshot) {69+ if (!root || typeof root.replaceChildren !== 'function' || !root.ownerDocument) {70+ throw new TypeError('roster root must be a DOM element');71+ }72+ const presentation = rosterPresentation(snapshot);73+ const { ownerDocument: document } = root;74+ const fragment = document.createDocumentFragment();75+ for (const row of presentation.rows) {76+ const item = document.createElement('li');77+ item.className = 'roster-viewer';78+ if (row.you) item.classList.add('you');79+ if (row.focusHolder) item.classList.add('focus-holder');80+ appendText(document, item, 'roster-handle', row.handle);81+ if (row.you) appendText(document, item, 'roster-you', 'you');82+ appendText(document, item, 'roster-role', row.role);83+ appendText(document, item, 'roster-adaptive', row.adaptive);84+ item.setAttribute(85+ 'aria-label',86+ `${row.handle}${row.you ? ', you' : ''}, ${row.role}, ${row.adaptive}`,87+ );88+ fragment.appendChild(item);89+ }90+ root.replaceChildren(fragment);91+ root.setAttribute('aria-label', `${presentation.count} connected viewers. ${presentation.focus}`);92+ return presentation;93+}@@ -0,0 +1,82 @@
1+import test from 'node:test';2+import assert from 'node:assert/strict';3+4+import { renderRoster, rosterPresentation } from './roster-ui.mjs';5+6+function snapshot({ focus = { state: 'vacant', slot: 0 } } = {}) {7+ return {8+ revision: 4,9+ self_presence_id: 'viewer-two',10+ viewers: [11+ { presence_id: 'viewer-one', session_id: 10, input_capable: true, you: false },12+ { presence_id: 'viewer-two', session_id: 11, input_capable: false, you: true },13+ { presence_id: 'viewer-three', session_id: 12, input_capable: true, you: false },14+ ],15+ focus,16+ };17+}18+19+test('presentation exposes only opaque roster, coarse eligibility, and focus state', () => {20+ const presentation = rosterPresentation(snapshot({21+ focus: {22+ state: 'held',23+ slot: 0,24+ holder: 'viewer-one',25+ session_id: 10,26+ focus_generation: 3,27+ },28+ }));29+ assert.equal(presentation.count, 3);30+ assert.equal(presentation.shell, '3 viewers · viewer-one holds control');31+ assert.deepEqual(32+ presentation.rows.map(({ fullHandle, role, you, adaptive }) => ({ fullHandle, role, you, adaptive })),33+ [34+ { fullHandle: 'viewer-one', role: 'holds control', you: false, adaptive: 'shared stream' },35+ { fullHandle: 'viewer-two', role: 'view only', you: true, adaptive: 'shared stream' },36+ { fullHandle: 'viewer-three', role: 'input eligible', you: false, adaptive: 'shared stream' },37+ ],38+ );39+});40+41+test('presentation rejects duplicate, unbounded, and inconsistent identities', () => {42+ const duplicate = snapshot();43+ duplicate.viewers[2].presence_id = 'viewer-one';44+ assert.throws(() => rosterPresentation(duplicate), /duplicate/);45+ const rawPeer = snapshot();46+ rawPeer.viewers[0].presence_id = 'a'.repeat(64);47+ assert.throws(() => rosterPresentation(rawPeer), /bounded opaque/);48+ const missingHolder = snapshot({49+ focus: { state: 'held', slot: 0, holder: 'viewer-gone', session_id: 90, focus_generation: 1 },50+ });51+ assert.throws(() => rosterPresentation(missingHolder), /absent/);52+});53+54+class FakeElement {55+ constructor(tagName, document) {56+ this.tagName = tagName;57+ this.ownerDocument = document;58+ this.children = [];59+ this.attributes = new Map();60+ this.className = '';61+ this.classList = { add: (...names) => { this.className += ` ${names.join(' ')}`; } };62+ this.textContent = '';63+ }64+ appendChild(child) { this.children.push(child); return child; }65+ replaceChildren(...children) { this.children = children; }66+ setAttribute(name, value) { this.attributes.set(name, value); }67+}68+69+test('renderer uses text nodes and emits a non-color-only accessible summary', () => {70+ const document = {71+ createElement: (tag) => new FakeElement(tag, document),72+ createDocumentFragment: () => new FakeElement('#fragment', document),73+ };74+ const root = new FakeElement('ul', document);75+ const presentation = renderRoster(root, snapshot());76+ assert.equal(presentation.shell, '3 viewers · control available');77+ assert.equal(root.attributes.get('aria-label'), '3 connected viewers. control available');78+ const rows = root.children[0].children;79+ assert.equal(rows.length, 3);80+ assert.match(rows[1].attributes.get('aria-label'), /you, view only/);81+ assert.equal('innerHTML' in rows[0], false);82+});@@ -0,0 +1,32 @@
1+import assert from 'node:assert/strict';2+import { readFile } from 'node:fs/promises';3+import test from 'node:test';4+5+const html = await readFile(new URL('./index.html', import.meta.url), 'utf8');6+const css = await readFile(new URL('./style.css', import.meta.url), 'utf8');7+const main = await readFile(new URL('./main.js', import.meta.url), 'utf8');8+const roster = await readFile(new URL('./roster-ui.mjs', import.meta.url), 'utf8');9+10+test('viewer and focus status is textual, live, and not color-only', () => {11+ assert.match(html, /id="viewer-status"[^>]*aria-live="polite"/);12+ assert.match(html, /id="roster-summary"[^>]*aria-live="polite"/);13+ assert.match(html, /id="roster-list"[^>]*aria-label="Connected viewers"/);14+ assert.match(roster, /you hold control/);15+ assert.match(roster, /control available/);16+ assert.match(roster, /view only/);17+ assert.match(roster, /input eligible/);18+});19+20+test('view-only v2 spectators have no focus action but retain audio and drawer access', () => {21+ assert.match(main, /classList\.toggle\('hidden', v2 && !sessionStateRuntime\.eligible\)/);22+ assert.match(html, /id="audio-toggle"[^>]*data-controller-focus/);23+ assert.match(html, /onclick="togglePanel\(true\)"/);24+ assert.match(main, /spectating · request control/);25+});26+27+test('roster rendering is text-safe and responsive controller focus remains visible', () => {28+ assert.doesNotMatch(roster, /innerHTML|insertAdjacentHTML|outerHTML/);29+ assert.match(roster, /textContent = text/);30+ assert.match(css, /\.link-btn:focus-visible,[\s\S]*\.link-btn\.controller-focus[\s\S]*outline:\s*2px solid var\(--moss\)/);31+ assert.match(css, /@media \(max-width: 720px\)[\s\S]*\.topbar-right[\s\S]*min-width:\s*0/);32+});@@ -41,3 +41,16 @@ test('stream diagnostics remain exclusive to the Info panel', () => {
4141 assert.doesNotMatch(html, /🔊|🔇|🔈/);4242 assert.match(css, /#audio-toggle\s*\{[^}]*min-width:\s*80px;[^}]*min-height:\s*36px;/s);4343 });44+45+test('multi-viewer roster stays in the Info drawer and outside the stream surface', () => {46+ const mainStart = html.indexOf('<main class="main">');47+ const mainEnd = html.indexOf('</main>', mainStart);48+ const panelStart = html.indexOf('<aside class="panel" id="panel"');49+ const panelEnd = html.indexOf('</aside>', panelStart);50+ const roster = html.indexOf('id="roster-section"');51+52+ assert.ok(roster > panelStart && roster < panelEnd);53+ assert.ok(roster < mainStart || roster > mainEnd);54+ assert.match(css, /\.roster-section\.hidden\s*\{\s*display:\s*none;/);55+ assert.match(css, /@media \(max-width: 480px\)[\s\S]*\.viewer-status/);56+});@@ -27,6 +27,25 @@ test('accepts revision gaps and ignores stale or native-generation-mismatched sn
2727 assert.equal(runtime.applyNativeSnapshot({ native_generation: 3, snapshot: snapshot(7) }), false);2828 });2929 30+test('mirrors complete multi-viewer roster replacements without changing local identity', () => {31+ const runtime = createSessionState();32+ runtime.acceptConnection({33+ nativeGeneration: 3,34+ controlProtocol: 'control-v2',35+ initialSnapshot: snapshot(1),36+ });37+ const joined = snapshot(5, {38+ state: 'held', slot: 0, holder: 'viewer-2', session_id: 8, focus_generation: 2,39+ });40+ joined.viewers.push({41+ presence_id: 'viewer-2', session_id: 8, input_capable: true, you: false,42+ });43+ assert.equal(runtime.applyNativeSnapshot({ native_generation: 3, snapshot: joined }), true);44+ assert.equal(runtime.snapshot().snapshot.viewers.length, 2);45+ assert.equal(runtime.focused, false);46+ assert.equal(runtime.eligible, true);47+});48+3049 test('publishes focus teardown before exposing self-focus loss', () => {3150 const order = [];3251 const runtime = createSessionState({@@ -139,6 +139,16 @@ body {
139139 font-size: 10px;140140 color: var(--muted-fg);141141 }142+.viewer-status {143+ max-width: 220px;144+ overflow: hidden;145+ color: var(--sand);146+ font-family: var(--mono);147+ font-size: 10px;148+ text-overflow: ellipsis;149+ white-space: nowrap;150+}151+.viewer-status.hidden { display: none; }142152 .controller-status.connected { color: var(--green); }143153 .controller-status.warning { color: var(--yellow); }144154 .link-btn {@@ -372,6 +382,10 @@ body.native-pointer-control * { cursor: none !important; }
372382 .pin-label.hidden { display: none; }373383 374384 @media (max-width: 720px) {385+ .topbar { gap: 8px; padding-inline: 10px; }386+ .topbar-right { gap: 8px; min-width: 0; }387+ .node-id { display: none; }388+ .viewer-status { max-width: 34vw; }375389 .bottombar {376390 grid-template-columns: auto minmax(0, 1fr) auto;377391 padding-inline: 10px;@@ -381,6 +395,8 @@ body.native-pointer-control * { cursor: none !important; }
381395 }382396 383397 @media (max-width: 480px) {398+ .controller-status { display: none; }399+ .viewer-status { max-width: 45vw; }384400 .bottombar { padding-inline: 8px; }385401 .pin-input { width: 60px; }386402 #control-toggle { max-width: 40vw; }@@ -715,6 +731,53 @@ body.native-pointer-control * { cursor: none !important; }
715731 margin-bottom: 8px;716732 }717733 .panel-section h3 .hash { color: var(--ochre); font-weight: 500; }734+.roster-section {735+ margin: 0 -6px 18px;736+ padding: 10px 6px 12px;737+ border-bottom: 1px solid var(--border-strong);738+}739+.roster-section.hidden { display: none; }740+.roster-summary {741+ margin-bottom: 7px;742+ color: var(--sand);743+ font-family: var(--mono);744+ font-size: 10px;745+ line-height: 1.4;746+}747+.roster-list { list-style: none; }748+.roster-viewer {749+ display: grid;750+ grid-template-columns: minmax(0, 1fr) auto;751+ gap: 2px 8px;752+ padding: 7px 6px;753+ border-top: 1px solid var(--border);754+}755+.roster-viewer.focus-holder {756+ border-left: 3px solid var(--moss);757+ background: color-mix(in srgb, var(--moss) 8%, transparent);758+}759+.roster-handle {760+ overflow: hidden;761+ font-family: var(--mono);762+ font-size: 11px;763+ text-overflow: ellipsis;764+ white-space: nowrap;765+}766+.roster-you {767+ align-self: center;768+ padding: 1px 4px;769+ border: 1px solid var(--ochre);770+ color: var(--ochre);771+ font-family: var(--mono);772+ font-size: 9px;773+ text-transform: uppercase;774+}775+.roster-role,776+.roster-adaptive {777+ color: var(--muted-fg);778+ font-size: 10px;779+}780+.roster-adaptive { text-align: right; }718781 .info-row {719782 display: flex;720783 justify-content: space-between;@@ -26,6 +26,9 @@ expected_path_mode=direct
2626 control_v2=false2727 viewers=12828 assert_shared_generation=false29+replace_viewer=030+legacy_exclusive=false31+expect_second_rejected=false2932 3033 usage() {3134 cat <<'EOF'@@ -47,9 +50,12 @@ Options:
4750 --relay-only Remove direct probe transports and require the4851 control, media, and input paths to stay relayed4952 --control-v2 Exercise explicit control/2 and input/2 focus50- --viewers COUNT Viewer count for control-v2 (Phase 1 requires 1)53+ --viewers COUNT Concurrent viewers (control-v2 supports 1..3 here)5154 --assert-shared-generation Require host-scoped generation naming, one5255 producer start, and final-viewer cleanup56+ --replace-viewer INDEX Replace this 1-based control-v2 viewer identity57+ --legacy-exclusive Exercise the unchanged exclusive legacy path58+ --expect-second-rejected Require explicit legacy second-client rejection5359 --help Show this help5460 EOF5561 }@@ -128,6 +134,19 @@ while [[ $# -gt 0 ]]; do
128134 assert_shared_generation=true129135 shift130136 ;;137+ --replace-viewer)138+ [[ $# -ge 2 ]] || die "--replace-viewer requires a value"139+ replace_viewer="$2"140+ shift 2141+ ;;142+ --legacy-exclusive)143+ legacy_exclusive=true144+ shift145+ ;;146+ --expect-second-rejected)147+ expect_second_rejected=true148+ shift149+ ;;131150 --help|-h)132151 usage133152 exit 0@@ -152,12 +171,24 @@ require_positive_integer "--reconnect-cycles" "$reconnect_cycles"
152171 require_positive_integer "--reconnect-frames" "$reconnect_frames"153172 require_positive_integer "--probe-timeout-seconds" "$probe_timeout_seconds"154173 require_positive_integer "--viewers" "$viewers"155-if [[ "$control_v2" == true && "$viewers" -ne 1 ]]; then156- die "Phase 1 control-v2 proof supports exactly --viewers 1"174+if [[ "$control_v2" == true && ( "$viewers" -lt 1 || "$viewers" -gt 3 ) ]]; then175+ die "this bounded control-v2 proof supports --viewers 1 through 3"176+fi177+case "$replace_viewer" in178+ ''|*[!0-9]*) die "--replace-viewer must be a non-negative integer" ;;179+esac180+if [[ "$replace_viewer" -gt "$viewers" ]]; then181+ die "--replace-viewer must identify one of the configured viewers"157182 fi158183 if [[ "$control_v2" == true && "$media_v3" == true ]]; then159184 die "--control-v2 conflicts with --media-v3"160185 fi186+if [[ "$legacy_exclusive" == true && "$control_v2" == true ]]; then187+ die "--legacy-exclusive conflicts with --control-v2"188+fi189+if [[ "$legacy_exclusive" == true && "$viewers" -ne 2 ]]; then190+ die "--legacy-exclusive currently proves exactly --viewers 2"191+fi161192 [[ "$primary_frames" -ge 4 ]] \162193 || die "--primary-frames must be at least 4 for keyframe recovery"163194 [[ "$reconnect_frames" -ge 4 ]] \@@ -200,6 +231,7 @@ primary_watchdog_pid=""
200231 bounded_pid=""201232 bounded_watchdog_pid=""202233 watchdog_pid=""234+viewer_pids=()203235 204236 stop_pid() {205237 local pid="$1"@@ -235,6 +267,10 @@ cleanup() {
235267 [[ -z "$bounded_watchdog_pid" ]] || wait "$bounded_watchdog_pid" 2>/dev/null || true236268 stop_pid "$primary_pid"237269 [[ -z "$primary_watchdog_pid" ]] || wait "$primary_watchdog_pid" 2>/dev/null || true270+ for viewer_pid in "${viewer_pids[@]-}"; do271+ [[ -n "$viewer_pid" ]] || continue272+ stop_pid "$viewer_pid"273+ done238274 stop_pid "$host_pid"239275 [[ -z "$host_watchdog_pid" ]] || wait "$host_watchdog_pid" 2>/dev/null || true240276 @@ -535,6 +571,111 @@ host_node_id="$(sed -n 's/^node_id=//p' "$host_log" | tail -n 1)"
535571 [[ "$host_node_id" == "$node_id" ]] || die "ready host node ID does not match its identity"536572 537573 if [[ "$control_v2" == true ]]; then574+ if [[ "$viewers" -gt 1 ]]; then575+ viewer_logs=()576+ viewer_identities=()577+ viewer_pids=()578+ replacement_log=""579+ viewer=1580+ while [[ "$viewer" -le "$viewers" ]]; do581+ viewer_identity="$tmp_root/viewer-${viewer}.key"582+ "$host_bin" identity init --output "$viewer_identity" \583+ >"$tmp_root/viewer-${viewer}-identity.log" 2>&1584+ viewer_identities+=("$viewer_identity")585+ viewer_log="$tmp_root/control-v2-viewer-${viewer}.log"586+ viewer_logs+=("$viewer_log")587+ if [[ "$viewer" -gt 1 ]]; then588+ "$probe_bin" --control-v2 --spectator \589+ --identity "$viewer_identity" --node-id "$node_id" --frames "$primary_frames" \590+ --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800 \591+ >"$viewer_log" 2>&1 &592+ else593+ "$probe_bin" --control-v2 --disconnect-with-focus \594+ --identity "$viewer_identity" --node-id "$node_id" --frames "$primary_frames" \595+ --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800 \596+ >"$viewer_log" 2>&1 &597+ fi598+ viewer_pids+=("$!")599+ if [[ "$viewer" -eq 1 ]]; then600+ wait_for_log_count "$host_log" 'input v2 client accepted' 1 "$host_pid" 'sigil' \601+ || die "slot-0 holder did not acquire focused input before spectators joined"602+ fi603+ viewer=$((viewer + 1))604+ done605+606+ wait_for_log_count "$host_log" 'MoQ control v2 client accepted' "$viewers" "$host_pid" 'sigil' \607+ || die "not all control-v2 viewers were admitted"608+ wait_for_log_count "$host_log" 'authorized MoQ media attachment accepted' "$viewers" "$host_pid" 'sigil' \609+ || die "not all control-v2 viewers attached to native MoQ"610+611+ if [[ "$replace_viewer" -gt 0 ]]; then612+ replacement_log="$tmp_root/control-v2-replacement-${replace_viewer}.log"613+ if ! run_bounded "$command_timeout_seconds" "$replacement_log" \614+ "$probe_bin" --control-v2 --spectator \615+ --identity "${viewer_identities[$((replace_viewer - 1))]}" \616+ --node-id "$node_id" --frames "$reconnect_frames" \617+ --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800; then618+ sed -n '1,240p' "$replacement_log" >&2 || true619+ die "same-peer replacement probe failed"620+ fi621+ grep -Fxq 'probe=ok' "$replacement_log" || die "replacement viewer did not progress"622+ grep -Fxq "roster_viewers=${viewers}" "$replacement_log" \623+ || die "replacement did not preserve bounded roster cardinality"624+ fi625+626+ viewer=1627+ for viewer_pid in "${viewer_pids[@]}"; do628+ if wait "$viewer_pid"; then viewer_status=0; else viewer_status=$?; fi629+ if [[ "$viewer" -eq "$replace_viewer" ]]; then630+ viewer=$((viewer + 1))631+ continue632+ fi633+ [[ "$viewer_status" -eq 0 ]] || {634+ sed -n '1,240p' "${viewer_logs[$((viewer - 1))]}" >&2 || true635+ die "control-v2 viewer $viewer failed"636+ }637+ grep -Fxq 'probe=ok' "${viewer_logs[$((viewer - 1))]}" \638+ || die "control-v2 viewer $viewer did not complete"639+ viewer=$((viewer + 1))640+ done641+ viewer_pids=()642+ grep -Fxq 'slot_0_input=ok' "${viewer_logs[0]}" || die "slot-0 holder input failed"643+ spectator=2644+ while [[ "$spectator" -le "$viewers" ]]; do645+ if [[ "$spectator" -eq "$replace_viewer" ]]; then646+ spectator=$((spectator + 1))647+ continue648+ fi649+ grep -Fxq 'spectator=ok' "${viewer_logs[$((spectator - 1))]}" \650+ || die "viewer $spectator did not remain a spectator"651+ spectator=$((spectator + 1))652+ done653+ roster_logs=("${viewer_logs[@]}")654+ if [[ -n "$replacement_log" ]]; then roster_logs+=("$replacement_log"); fi655+ grep -Fhq "roster_viewers=${viewers}" "${roster_logs[@]}" 2>/dev/null \656+ || die "no viewer observed the complete roster"657+ wait_for_log_count "$host_log" 'shared media generation stopped with all sources cleaned up' \658+ 1 "$host_pid" 'sigil' || die "shared generation did not stop after the final viewer"659+ [[ "$(grep -Fc -- 'shared media generation started' "$host_log" || true)" -eq 1 ]] \660+ || die "multi-viewer control-v2 started more than one generation"661+ [[ "$(grep -Fc -- 'shared media generation stopped with all sources cleaned up' "$host_log" || true)" -eq 1 ]] \662+ || die "multi-viewer control-v2 stopped the generation more than once"663+664+ kill -TERM "$host_pid" 2>/dev/null || true665+ if wait "$host_pid"; then host_status=0; else host_status=$?; fi666+ host_pid=""667+ wait "$host_watchdog_pid" 2>/dev/null || true668+ host_watchdog_pid=""669+ [[ "$host_status" -eq 0 ]] || die "host exited with status $host_status"670+ printf 'loopback_proof=ok\n'671+ printf 'control_v2=ok\n'672+ printf 'viewers=%s\n' "$viewers"673+ printf 'same_peer_replacement=%s\n' "$([[ "$replace_viewer" -gt 0 ]] && printf ok || printf not-requested)"674+ printf 'shared_generation=ok\n'675+ printf 'slot_0_single_holder=ok\n'676+ printf 'spectator_progress=ok\n'677+ exit 0678+ fi538679 v2_log="$tmp_root/control-v2.log"539680 v1_log="$tmp_root/control-v1-compatibility.log"540681 if ! run_bounded "$command_timeout_seconds" "$v2_log" \@@ -441,7 +441,7 @@ pub(crate) async fn connect_client(
441441 &app,442442 &state.session_snapshot,443443 media_generation,444- initial_snapshot,444+ initial_snapshot.clone(),445445 )446446 .await?;447447 let (control_tx, control_rx) = tokio::sync::mpsc::channel(SESSION_CONTROL_COMMAND_CAPACITY);@@ -449,7 +449,7 @@ pub(crate) async fn connect_client(
449449 tokio::spawn(run_session_control(450450 app.clone(),451451 media_generation,452- media_session_id,452+ initial_snapshot.media.generation_id,453453 media_control_stream,454454 frame_recv,455455 control_rx,@@ -47,7 +47,7 @@ pub(crate) async fn install_initial_snapshot(
4747 pub(crate) async fn run_session_control(4848 app: AppHandle,4949 native_generation: u64,50- session_id: u64,50+ media_generation_id: u64,5151 mut send: iroh::endpoint::SendStream,5252 mut recv: iroh::endpoint::RecvStream,5353 mut commands: tokio::sync::mpsc::Receiver<ClientControlEnvelopeV2>,@@ -65,7 +65,7 @@ pub(crate) async fn run_session_control(
6565 envelope = read_server_control_v2(&mut recv) => {6666 match envelope {6767 Ok(Some(ServerControlEnvelopeV2::Snapshot { snapshot })) => {68- if snapshot.media.generation_id != session_id {68+ if snapshot.media.generation_id != media_generation_id {6969 eprintln!("[client] control v2 snapshot changed media generation");7070 break;7171 }