1use std::{2 collections::BTreeMap,3 fs,4 fs::OpenOptions,5 io::{Read, Write},6 path::{Path, PathBuf},7 time::{Duration, Instant, SystemTime, UNIX_EPOCH},8};910use anyhow::{Context, Result, bail, ensure};11use clap::{Parser, ValueEnum};12use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, endpoint::presets};13use iroh_moq::{Moq, MoqSession};14use moq_net::{BroadcastConsumer, GroupConsumer, TrackConsumer};15use sigil_protocol::{16 AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1,17 AuthenticatedMediaObject, CONTROL_ALPN_V1, CONTROL_ALPN_V2, Capability,18 ClientControlEnvelopeV2, ClientHello, ClientHelloV2, ControllerSlot, FocusCommandActionV2,19 FocusCommandV2, FocusStateV2, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN,20 GAMEPAD_TRIGGER_MAX, GamepadState, INPUT_ALPN_V1, INPUT_ALPN_V2, INVITATION_CLOCK_SKEW_SECS,21 InputAck, InputClientHelloV2, InputEvent, InputEventV2, InvitationGrants,22 KeyframeRequestReasonV3, MAX_INVITATION_TOKEN_LEN, MAX_MEDIA_GROUP_BYTES_V3,23 MAX_MEDIA_OBJECT_ID_V3, MEDIA_ALPN_V3, MEDIA_FEEDBACK_ALPN_V1, MediaCodec,24 MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame,25 MediaObjectCoordinates, MediaObjectV3, MediaTrack, PointerPosition, PointerSurfaceDimensions,26 ProtocolError, ServerControlEnvelopeV2, SignedInvitation, SignedMediaGenerationCertificate,27 SignedSubscriptionCapability, SubscriptionTracks, decode_media_frame_object,28 media_moq_broadcast_name, read_adaptive_bitrate_decision_v1, read_host_hello,29 read_host_hello_v2, read_input_ack, read_input_ack_v2, read_input_host_hello_v2,30 read_media_object_v3, read_server_control_v2, write_client_control_v2, write_client_hello,31 write_client_hello_v2, write_input_client_hello_v2, write_input_event, write_input_event_v2,32 write_media_control_request_v3, write_media_feedback_report_v1,33};343536use std::os::unix::fs::{MetadataExt, OpenOptionsExt};37383940mod identity;4142mod moq_catalog;4344use moq_catalog::{45 MoqCatalogMode, subscribe_authenticated_goq_video_track, subscribe_goq_video_track,46};4748const MEDIA_OBJECT_CAPACITY: usize = 4;495051enum AdaptiveProfile {52 Clean,53 FloorBreach,54 Stale,55}56575859struct Args {60 61 node_id: EndpointId,62 /// Owner-only 32-byte Iroh identity. Omit for the existing ephemeral probe.63 64 identity: Option<PathBuf>,65 /// Remove all direct IP transports from the probe endpoint so control,66 /// media, and input are forced through Iroh relay paths.67 68 relay_only: bool,69 /// Owner-only one-time Sigil invitation file. Sent only on the first media70 /// handshake and requires a persistent identity bound to that invitation.71 72 invitation: Option<PathBuf>,73 74 frames: u32,75 76 timeout_seconds: u64,77 /// Fail unless accepted media frames sustain at least this first-to-last78 /// cadence. Intended for live appliance performance gates.79 80 minimum_fps: Option<f64>,81 /// Optional strict encoded-size assertion. When omitted, accept and report82 /// the host's observed bounded dimensions.83 84 expect_size: Option<(u16, u16)>,85 /// Exercise custom grouped v3 media instead of upstream MoQ. Intended only86 /// for compatibility validation.87 88 media_v3: bool,89 /// Exercise the explicit revisioned control/2 and focus-bound input/2 path.90 91 control_v2: bool,92 /// Join control/2 as a view-only exercise without requesting slot-0 focus.93 94 spectator: bool,95 /// End focused v2 proof by disconnecting instead of issuing a release96 /// against a roster revision that may advance while spectators join.97 98 disconnect_with_focus: bool,99 /// Participate in bounded holder-approved slot-0 handoffs and stop after100 /// observing this many successor generations.101 102 focus_handoffs: u32,103 /// Drive a bounded per-viewer adaptive profile on control/2. Clean viewers104 /// prove survivor recovery; floor-breach viewers prove isolated detach.105 106107108109110111 adaptive_profile: Option<AdaptiveProfile>,112 /// Request a configured recovery keyframe after three accepted frames,113 /// then prove no delta history is delivered before the recovery barrier.114 115 keyframe_smoke: bool,116 /// Correlation identifier for `--keyframe-smoke` host evidence.117 118 keyframe_request_id: u64,119 /// Stop reading upstream MoQ after the first configured IDR, then require120 /// the transport to cancel that stale GOP and resume at a newer configured121 /// IDR within a bounded interval. Intended for latency-growth proofs.122 123124125126127 slow_consumer_ms: Option<u64>,128 /// Require gamepad negotiation and emit one bounded non-neutral snapshot129 /// followed by neutral. Intended for evtest-backed uinput proof.130 131 gamepad_smoke: bool,132 /// Require relative-pointer negotiation and emit bounded motion plus one133 /// complete left-click. Intended for libinput/Gamescope-backed proof.134 135 pointer_smoke: bool,136 /// Require the host to return an immediately available compositor pointer137 /// position and visibility sample. Intended for Gamescope restart proofs.138 139 pointer_feedback_smoke: bool,140 /// Attach the media-feedback stream to the active session, submit one141 /// complete report, and print the host's bounded decision. Intended for142 /// validating shadow-only encoder configurations.143 144145146 )]147 feedback_smoke: bool,148 /// Prove adaptive-resolution liveness across a controlled Gamescope stall.149 /// GATE_DIR coordinates the external SIGSTOP/SIGCONT harness.150 151152153154155156157158159160 )]161 resolution_stall_smoke: Option<PathBuf>,162 /// After the stall is released, submit this many clean, sequence-advancing163 /// feedback windows and require the controller to climb back out of the164 /// pressure trough. The host needs CLEAN_WINDOWS plus COOLDOWN_WINDOWS of165 /// clean evidence, so fewer than about twelve cannot observe an increase.166 /// 0 keeps the historical behaviour of ending at the resume barrier.167 168169170171172173 resolution_recovery_windows: u64,174}175176177enum MediaTransport {178 UpstreamMoq,179 GroupedV3,180}181182impl MediaTransport {183 fn alpn(self) -> &'static [u8] {184 match self {185 Self::UpstreamMoq => CONTROL_ALPN_V1,186 Self::GroupedV3 => MEDIA_ALPN_V3,187 }188 }189190 fn label(self) -> &'static str {191 match self {192 Self::UpstreamMoq => "iroh-moq",193 Self::GroupedV3 => "grouped-v3",194 }195 }196197 fn media_alpn_label(self) -> &'static str {198 match self {199 Self::UpstreamMoq => std::str::from_utf8(iroh_moq::ALPN)200 .expect("pinned iroh-moq ALPN must be printable UTF-8"),201 Self::GroupedV3 => "sigil/media/3",202 }203 }204}205206const KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES: u32 = 3;207const KEYFRAME_SMOKE_MINIMUM_FRAMES: u32 = KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES + 1;208const SLOW_CONSUMER_MINIMUM_FRAMES: u32 = 2;209const SLOW_CONSUMER_RECOVERY_TIMEOUT: Duration = Duration::from_secs(2);210const RESOLUTION_STALL_FEEDBACK_INTERVAL: Duration = Duration::from_secs(1);211const RESOLUTION_STALL_FRESH_REPORTS: u64 = 10;212const RESOLUTION_STALL_FEEDBACK_TIMEOUT: Duration = Duration::from_secs(5);213const RESOLUTION_STALL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(2);214const RESOLUTION_STALL_GATE_POLL_INTERVAL: Duration = Duration::from_millis(50);215const MAX_INVITATION_FILE_BYTES: u64 = (MAX_INVITATION_TOKEN_LEN + 2) as u64;216217218enum MediaObjectOutcomeV3 {219 Object {220 accept_index: u64,221 object: MediaObjectV3,222 },223 Dropped {224 accept_index: u64,225 },226 Malformed {227 accept_index: u64,228 error: ProtocolError,229 },230}231232impl MediaObjectOutcomeV3 {233 fn accept_index(&self) -> u64 {234 match self {235 Self::Object { accept_index, .. }236 | Self::Dropped { accept_index }237 | Self::Malformed { accept_index, .. } => *accept_index,238 }239 }240241 fn is_fast_forward_barrier(&self) -> bool {242 let Self::Object { object, .. } = self else {243 return false;244 };245 object.header.object_id == 0246 && object.header.flags.contains(FrameFlags::KEYFRAME)247 && object.header.flags.contains(FrameFlags::CODEC_CONFIG)248 && object.header.flags.contains(FrameFlags::DISCONTINUITY)249 }250}251252253struct MediaObjectReorderV3 {254 next_accept_index: u64,255 completed: BTreeMap<u64, MediaObjectOutcomeV3>,256}257258impl MediaObjectReorderV3 {259 fn new(first_accept_index: u64) -> Self {260 Self {261 next_accept_index: first_accept_index,262 completed: BTreeMap::new(),263 }264 }265266 fn pending_len(&self) -> usize {267 self.completed.len()268 }269270 fn push(&mut self, outcome: MediaObjectOutcomeV3) -> Result<Option<MediaObjectOutcomeV3>> {271 if matches!(outcome, MediaObjectOutcomeV3::Malformed { .. }) {272 return Ok(Some(outcome));273 }274 let accept_index = outcome.accept_index();275 if accept_index < self.next_accept_index {276 // A discontinuity barrier may advance beyond older in-flight277 // reads. Their eventual timeout/reset outcomes belong to the278 // superseded GOP and must not poison the recovered sequence.279 return Ok(None);280 }281 let fast_forward = outcome.is_fast_forward_barrier();282 ensure!(283 self.completed.insert(accept_index, outcome).is_none(),284 "media v3 accept index {accept_index} completed more than once"285 );286 if fast_forward {287 self.completed.retain(|index, _| *index >= accept_index);288 self.next_accept_index = accept_index;289 }290 self.take_next()291 }292293 fn take_next(&mut self) -> Result<Option<MediaObjectOutcomeV3>> {294 let Some(outcome) = self.completed.remove(&self.next_accept_index) else {295 return Ok(None);296 };297 self.next_accept_index = self298 .next_accept_index299 .checked_add(1)300 .context("media v3 accept index overflowed")?;301 Ok(Some(outcome))302 }303}304305struct MediaObjectReceiverV3 {306 connection: iroh::endpoint::Connection,307 reads: tokio::task::JoinSet<MediaObjectOutcomeV3>,308 reorder: MediaObjectReorderV3,309 next_accept_index: u64,310 accepting: bool,311 read_timeout: Duration,312}313314impl MediaObjectReceiverV3 {315 fn new(connection: iroh::endpoint::Connection, read_timeout: Duration) -> Self {316 Self {317 connection,318 reads: tokio::task::JoinSet::new(),319 reorder: MediaObjectReorderV3::new(0),320 next_accept_index: 0,321 accepting: true,322 read_timeout,323 }324 }325326 async fn next(&mut self) -> Result<Option<MediaObjectOutcomeV3>> {327 loop {328 if let Some(completed) = self.reorder.take_next()? {329 return Ok(Some(completed));330 }331 if !self.accepting && self.reads.is_empty() {332 ensure!(333 self.reorder.pending_len() == 0,334 "media v3 connection closed with an incomplete object order"335 );336 return Ok(None);337 }338339 tokio::select! {340 accepted = self.connection.accept_uni(),341 if self.accepting342 && self.reads.len() + self.reorder.pending_len()343 < MEDIA_OBJECT_CAPACITY => {344 match accepted {345 Ok(mut stream) => {346 let accept_index = self.next_accept_index;347 self.next_accept_index = self348 .next_accept_index349 .checked_add(1)350 .context("media v3 accept index overflowed")?;351 let read_timeout = self.read_timeout;352 self.reads.spawn(async move {353 match tokio::time::timeout(354 read_timeout,355 read_media_object_v3(&mut stream),356 )357 .await358 {359 Ok(Ok(object)) => MediaObjectOutcomeV3::Object {360 accept_index,361 object,362 },363 Ok(Err(ProtocolError::Io(_))) | Err(_) => {364 MediaObjectOutcomeV3::Dropped { accept_index }365 }366 Ok(Err(error)) => MediaObjectOutcomeV3::Malformed {367 accept_index,368 error,369 },370 }371 });372 }373 Err(_) => self.accepting = false,374 }375 }376 completed = self.reads.join_next(), if !self.reads.is_empty() => {377 match completed.expect("guarded by non-empty media v3 object task set") {378 Ok(outcome) => {379 if let Some(completed) = self.reorder.push(outcome)? {380 return Ok(Some(completed));381 }382 }383 Err(error) if error.is_cancelled() => continue,384 Err(error) => return Err(error).context("media v3 object read task failed"),385 }386 }387 }388 }389 }390}391392impl Drop for MediaObjectReceiverV3 {393 fn drop(&mut self) {394 self.reads.abort_all();395 }396}397398399enum MediaObjectDecisionV3 {400 Deliver { discontinuity: bool },401 DropLate,402 DropUntilKeyframe,403}404405406struct MediaObjectSequenceV3 {407 group_id: Option<u64>,408 last_object_id: Option<u32>,409 last_sequence: Option<u64>,410 group_payload_bytes: usize,411 waiting_for_keyframe: bool,412}413414impl MediaObjectSequenceV3 {415 fn new() -> Self {416 Self {417 waiting_for_keyframe: true,418 ..Self::default()419 }420 }421422 fn request_recovery(&mut self) {423 self.waiting_for_keyframe = true;424 }425426 fn note_drop(&mut self) -> bool {427 let entered = !self.waiting_for_keyframe;428 self.waiting_for_keyframe = true;429 entered430 }431432 fn classify(&mut self, object: &MediaObjectV3) -> MediaObjectDecisionV3 {433 let header = &object.header;434 if self435 .group_id436 .is_some_and(|group_id| header.group_id < group_id)437 || self438 .last_sequence439 .is_some_and(|sequence| header.sequence <= sequence)440 {441 return MediaObjectDecisionV3::DropLate;442 }443444 let new_group = self.group_id != Some(header.group_id);445 let configured_group_start = header.object_id == 0446 && header.flags.contains(FrameFlags::KEYFRAME)447 && header.flags.contains(FrameFlags::CODEC_CONFIG);448 if new_group && !configured_group_start {449 self.waiting_for_keyframe = true;450 return MediaObjectDecisionV3::DropUntilKeyframe;451 }452 if !new_group && self.waiting_for_keyframe {453 return MediaObjectDecisionV3::DropUntilKeyframe;454 }455456 let sequence_contiguous = self457 .last_sequence458 .is_none_or(|sequence| sequence.checked_add(1) == Some(header.sequence));459 let object_contiguous = new_group460 || self461 .last_object_id462 .is_some_and(|object_id| object_id.checked_add(1) == Some(header.object_id));463 let Some(next_group_bytes) = (if new_group {464 Some(object.payload.len())465 } else {466 self.group_payload_bytes.checked_add(object.payload.len())467 }) else {468 self.waiting_for_keyframe = true;469 return MediaObjectDecisionV3::DropUntilKeyframe;470 };471 let recovery_requires_discontinuity = new_group472 && self.group_id.is_some()473 && (self.waiting_for_keyframe || !sequence_contiguous)474 && !header.flags.contains(FrameFlags::DISCONTINUITY);475 if (!sequence_contiguous && !new_group)476 || !object_contiguous477 || next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3478 || recovery_requires_discontinuity479 {480 self.waiting_for_keyframe = true;481 return MediaObjectDecisionV3::DropUntilKeyframe;482 }483484 let discontinuity = header.flags.contains(FrameFlags::DISCONTINUITY)485 || self.waiting_for_keyframe486 || !sequence_contiguous;487 self.group_id = Some(header.group_id);488 self.last_object_id = Some(header.object_id);489 self.last_sequence = Some(header.sequence);490 self.group_payload_bytes = next_group_bytes;491 self.waiting_for_keyframe = false;492 MediaObjectDecisionV3::Deliver { discontinuity }493 }494}495496497enum MoqProbeOutcome {498 Frame {499 frame: MediaFrame,500 group_sequence: u64,501 recovery: bool,502 },503 Dropped,504}505506struct MoqProbeGroupCursor {507 group: GroupConsumer,508 sequence: u64,509 object_count: usize,510 object_bytes: usize,511 recovery: bool,512}513514/// Own all upstream handles for the complete proof session. Dropping `Moq`515/// stops its actor, while dropping the consumers cancels their subscriptions.516struct MoqProbeLifetime {517 _moq: Moq,518 session: MoqSession,519 _broadcast: BroadcastConsumer,520}521522struct MoqProbeReceiver {523 lifetime: MoqProbeLifetime,524 track: TrackConsumer,525 catalog_mode: MoqCatalogMode,526 current_group: Option<MoqProbeGroupCursor>,527 last_group_sequence: Option<u64>,528 last_frame_sequence: Option<u64>,529 waiting_for_keyframe: bool,530 pending_recovery: bool,531 cancelled_groups: u64,532 group_gaps: u64,533 maximum_group_objects: usize,534 maximum_group_bytes: usize,535 historical_suffix_frames: u64,536 certificate: Option<SignedMediaGenerationCertificate>,537}538539struct MoqProbeAuthentication {540 expected_host: [u8; 32],541 subscription_capability: String,542 authorization_revision: u64,543 now_unix: u64,544}545546impl MoqProbeReceiver {547 async fn connect(548 endpoint: &Endpoint,549 address: EndpointAddr,550 session_id: u64,551 generation_id: u64,552 advertised_broadcast_name: Option<&str>,553 timeout: Duration,554 authentication: Option<MoqProbeAuthentication>,555 ) -> Result<Self> {556 let broadcast_name = match advertised_broadcast_name {557 Some(name) => name.to_owned(),558 None => media_moq_broadcast_name(session_id)559 .context("deriving session-scoped MoQ broadcast name")?,560 };561 let moq = Moq::new(endpoint.clone());562 let mut session = tokio::time::timeout(timeout, moq.connect(address.clone()))563 .await564 .context("timed out connecting upstream MoQ media session")?565 .context("connecting upstream MoQ media session")?;566 ensure!(567 session.remote_id() == address.id,568 "upstream MoQ connected to unexpected peer {} instead of {}",569 session.remote_id(),570 address.id571 );572 let broadcast = tokio::time::timeout(timeout, session.subscribe(&broadcast_name))573 .await574 .with_context(|| {575 format!("timed out subscribing to upstream MoQ broadcast {broadcast_name}")576 })?577 .with_context(|| format!("subscribing to upstream MoQ broadcast {broadcast_name}"))?;578 let catalog = if let Some(authentication) = authentication {579 SignedSubscriptionCapability::decode(&authentication.subscription_capability)?580 .verify_binding(581 authentication.expected_host,582 generation_id,583 *endpoint.id().as_bytes(),584 SubscriptionTracks::VIDEO_H264,585 authentication.authorization_revision,586 authentication.now_unix,587 )?;588 subscribe_authenticated_goq_video_track(589 &broadcast,590 timeout,591 authentication.expected_host,592 generation_id,593 authentication.now_unix,594 )595 .await596 .context("resolving authenticated Goq MoQ catalog")?597 } else {598 subscribe_goq_video_track(&broadcast, timeout)599 .await600 .context("resolving Goq MoQ catalog")?601 };602 Ok(Self {603 lifetime: MoqProbeLifetime {604 _moq: moq,605 session,606 _broadcast: broadcast,607 },608 track: catalog.track,609 catalog_mode: catalog.mode,610 current_group: None,611 last_group_sequence: None,612 last_frame_sequence: None,613 waiting_for_keyframe: true,614 pending_recovery: false,615 cancelled_groups: 0,616 group_gaps: 0,617 maximum_group_objects: 0,618 maximum_group_bytes: 0,619 historical_suffix_frames: 0,620 certificate: catalog.certificate,621 })622 }623624 fn connection(&self) -> &iroh::endpoint::Connection {625 self.lifetime.session.conn()626 }627628 /// Stop consuming the current group immediately. The track cursor already629 /// advanced when the group was selected, so a subsequent read can only630 /// enter a newer native group; no cancelled delta suffix can be accepted.631 fn request_recovery(&mut self) -> Option<u64> {632 self.waiting_for_keyframe = true;633 self.pending_recovery = true;634 self.current_group.take().map(|cursor| {635 self.cancelled_groups = self.cancelled_groups.saturating_add(1);636 self.last_group_sequence = Some(cursor.sequence);637 cursor.sequence638 })639 }640641 async fn next(&mut self) -> Result<Option<MoqProbeOutcome>> {642 loop {643 if self.current_group.is_none() {644 let group = self645 .track646 .next_group()647 .await648 .context("reading the next sequential upstream MoQ group")?;649 let Some(group) = group else {650 return Ok(None);651 };652 let sequence = group.sequence;653 let group_gap = classify_moq_probe_group(self.last_group_sequence, sequence)?;654 if group_gap {655 self.group_gaps = self.group_gaps.saturating_add(1);656 self.waiting_for_keyframe = true;657 self.pending_recovery = true;658 }659 self.current_group = Some(MoqProbeGroupCursor {660 group,661 sequence,662 object_count: 0,663 object_bytes: 0,664 recovery: self.pending_recovery,665 });666 self.pending_recovery = false;667 }668669 let cursor = self670 .current_group671 .as_mut()672 .expect("MoQ group cursor was initialized");673 let object = match cursor.group.read_frame().await {674 Ok(object) => object,675 Err(error) if moq_probe_error_is_recoverable(&error) => {676 self.cancelled_groups = self.cancelled_groups.saturating_add(1);677 self.last_group_sequence = Some(cursor.sequence);678 self.current_group = None;679 self.waiting_for_keyframe = true;680 self.pending_recovery = true;681 return Ok(Some(MoqProbeOutcome::Dropped));682 }683 Err(error) => {684 bail!(685 "upstream MoQ group {} failed terminally: {error}",686 cursor.sequence687 );688 }689 };690 let Some(object) = object else {691 ensure!(692 cursor.object_count > 0,693 "upstream MoQ group {} was empty",694 cursor.sequence695 );696 self.last_group_sequence = Some(cursor.sequence);697 self.current_group = None;698 continue;699 };700701 let next_bytes = validate_moq_probe_object_bounds(702 cursor.sequence,703 cursor.object_count,704 cursor.object_bytes,705 object.len(),706 )?;707 let authenticated_coordinates = self708 .certificate709 .as_ref()710 .map(|_| AuthenticatedMediaObject::coordinates(&object))711 .transpose()712 .with_context(|| {713 format!(714 "decoding authenticated upstream MoQ group {} object {}",715 cursor.sequence, cursor.object_count716 )717 })?;718 let payload = if let (Some(certificate), Some(coordinates)) =719 (&self.certificate, authenticated_coordinates)720 {721 AuthenticatedMediaObject::verify(722 &object,723 certificate,724 MediaObjectCoordinates {725 generation_id: certificate.claims.generation_id,726 track: MediaTrack::VideoH264,727 group_id: cursor.sequence,728 object_id: u32::try_from(cursor.object_count)729 .context("authenticated MoQ object index overflow")?,730 flags: coordinates.flags,731 },732 )733 .with_context(|| {734 format!(735 "verifying authenticated upstream MoQ group {} object {}",736 cursor.sequence, cursor.object_count737 )738 })?739 } else {740 object.as_ref()741 };742 let frame = decode_media_frame_object(payload).with_context(|| {743 format!(744 "decoding upstream MoQ group {} object {}",745 cursor.sequence, cursor.object_count746 )747 })?;748 ensure!(749 authenticated_coordinates.is_none_or(|coordinates| {750 coordinates.flags == u16::from(frame.header.flags.bits())751 }),752 "authenticated media flags do not match the decoded media envelope"753 );754 let first_object = cursor.object_count == 0;755 let contiguous = validate_moq_probe_frame(756 cursor.sequence,757 first_object,758 self.last_frame_sequence,759 &frame,760 )?;761 let recovery = cursor.recovery762 || self.waiting_for_keyframe763 || (first_object && !contiguous)764 || frame.header.flags.contains(FrameFlags::DISCONTINUITY);765 if self.waiting_for_keyframe && !first_object {766 self.historical_suffix_frames = self.historical_suffix_frames.saturating_add(1);767 bail!(768 "upstream MoQ delivered a historical delta suffix while recovery was pending"769 );770 }771 cursor.object_count += 1;772 cursor.object_bytes = next_bytes;773 self.maximum_group_objects = self.maximum_group_objects.max(cursor.object_count);774 self.maximum_group_bytes = self.maximum_group_bytes.max(next_bytes);775 self.last_frame_sequence = Some(frame.header.sequence);776 self.waiting_for_keyframe = false;777 return Ok(Some(MoqProbeOutcome::Frame {778 frame,779 group_sequence: cursor.sequence,780 recovery,781 }));782 }783 }784}785786787async fn run_control_v2_smoke(788 endpoint: &Endpoint,789 address: EndpointAddr,790 nonce: [u8; 16],791 invitation: Option<&str>,792 frames: u32,793 timeout_seconds: u64,794 expected_size: Option<(u16, u16)>,795 spectator: bool,796 disconnect_with_focus: bool,797 focus_handoffs: u32,798 adaptive_profile: Option<AdaptiveProfile>,799) -> Result<()> {800 let control_connection = endpoint801 .connect(address.clone(), CONTROL_ALPN_V2)802 .await803 .context("connecting control v2 protocol")?;804 let (mut control_send, mut control_recv) = control_connection805 .open_bi()806 .await807 .context("opening control v2 stream")?;808 let mut hello = ClientHelloV2::new("sigil-probe/0.1.0", nonce, vec![Capability::VideoH264]);809 if let Some(invitation) = invitation {810 hello = hello.with_invitation(invitation);811 }812 write_client_hello_v2(&mut control_send, &hello)813 .await814 .context("writing control v2 hello")?;815 let host = tokio::time::timeout(816 Duration::from_secs(timeout_seconds),817 read_host_hello_v2(&mut control_recv),818 )819 .await820 .context("timed out waiting for control v2 hello")??821 .context("host closed during control v2 hello")?;822 if !host.accepted {823 bail!(824 "host rejected control v2 stream: {}",825 host.message.as_deref().unwrap_or("unspecified reason")826 );827 }828 let session_id = host829 .session_id830 .context("host omitted control v2 session id")?;831 let subscription_capability = host832 .media_subscription_capability833 .context("host omitted control v2 media subscription capability")?;834 let media_authorization_revision = host835 .media_authorization_revision836 .context("host omitted control v2 media authorization revision")?;837 let initial = host838 .snapshot839 .context("host omitted initial control v2 snapshot")?;840 ensure!(841 initial.revision > 0,842 "initial control v2 snapshot was not revisioned"843 );844 if !spectator && focus_handoffs == 0 {845 ensure!(846 matches!(initial.focus, FocusStateV2::Vacant { .. }),847 "initial control v2 snapshot did not advertise a vacant slot"848 );849 }850 let generation_id = initial.media.generation_id;851 let broadcast_name = initial.media.broadcast_name.clone();852853 let mut media = MoqProbeReceiver::connect(854 endpoint,855 address.clone(),856 session_id,857 generation_id,858 Some(&broadcast_name),859 Duration::from_secs(timeout_seconds),860 Some(MoqProbeAuthentication {861 expected_host: *address.id.as_bytes(),862 subscription_capability,863 authorization_revision: media_authorization_revision,864 now_unix: SystemTime::now()865 .duration_since(UNIX_EPOCH)866 .context("system clock is before the Unix epoch")?867 .as_secs(),868 }),869 )870 .await?;871 if focus_handoffs > 0 {872 return run_focus_handoff_participant(873 control_connection,874 control_send,875 control_recv,876 media,877 initial,878 focus_handoffs,879 frames,880 timeout_seconds,881 )882 .await;883 }884 if spectator {885 if let Some(profile) = adaptive_profile {886 let mut accepted = 0_u32;887 let mut last_sequence = None;888 let feedback =889 run_v2_adaptive_profile(endpoint, address.clone(), nonce, session_id, profile);890 tokio::pin!(feedback);891 let mut media_open = true;892 let evidence = loop {893 tokio::select! {894 evidence = &mut feedback => break evidence?,895 outcome = media.next(), if media_open => {896 let Some(outcome) = outcome? else {897 if profile == AdaptiveProfile::FloorBreach {898 media_open = false;899 continue;900 }901 bail!("adaptive-profile spectator media ended early");902 };903 let MoqProbeOutcome::Frame { frame, .. } = outcome else {904 continue;905 };906 if let Some((width, height)) = expected_size {907 ensure!(908 frame.header.width == width && frame.header.height == height,909 "adaptive-profile media dimensions did not match {width}x{height}"910 );911 }912 if let Some(previous) = last_sequence {913 ensure!(914 frame.header.sequence > previous,915 "adaptive-profile media sequence regressed"916 );917 }918 last_sequence = Some(frame.header.sequence);919 accepted += 1;920 }921 }922 };923 ensure!(924 accepted >= frames.min(60),925 "adaptive profile did not preserve media progress"926 );927 control_connection.close(0_u32.into(), b"adaptive profile complete");928 println!("probe=ok");929 println!("transport=iroh-moq");930 println!("control_alpn=sigil/control/2");931 println!("media_generation_id={generation_id}");932 println!("media_broadcast_name={broadcast_name}");933 println!("roster_viewers={}", initial.viewers.len());934 println!("frames={accepted}");935 println!("spectator=ok");936 println!("adaptive_profile={}", profile.label());937 println!("adaptive_minimum_kbps={}", evidence.minimum_kbps);938 println!("adaptive_final_kbps={}", evidence.final_kbps);939 println!(940 "adaptive_survivor_recovery={}",941 if evidence.survivor_recovery {942 "ok"943 } else {944 "not-applicable"945 }946 );947 println!(948 "adaptive_floor_detach={}",949 if evidence.floor_detach {950 "ok"951 } else {952 "not-applicable"953 }954 );955 println!("resource_summary_version=1");956 println!("maximum_roster_viewers={}", initial.viewers.len());957 println!("media_progress=monotonic");958 println!("probe_media_object_capacity={MEDIA_OBJECT_CAPACITY}");959 return Ok(());960 }961 let mut accepted = 0_u32;962 let mut last_sequence = None;963 // The roster at hello time is one instantaneous sample: a peer that has964 // not finished attaching yet, or one that has already left, makes it965 // disagree with the configured viewer count for reasons that say966 // nothing about admission. Track the maximum across the snapshot967 // stream, exactly as the focus participant does, so the cardinality968 // assertion is about what the host actually admitted.969 let mut maximum_roster_viewers = initial.viewers.len();970 let mut roster_revision = initial.revision;971 while accepted < frames {972 let outcome = tokio::select! {973 envelope = read_server_control_v2(&mut control_recv) => {974 match envelope?.context("host closed during spectator proof")? {975 ServerControlEnvelopeV2::Snapshot { snapshot: next } => {976 ensure!(977 next.revision >= roster_revision,978 "spectator snapshot revision regressed"979 );980 roster_revision = next.revision;981 maximum_roster_viewers =982 maximum_roster_viewers.max(next.viewers.len());983 continue;984 }985 _ => continue,986 }987 }988 outcome = tokio::time::timeout(Duration::from_secs(timeout_seconds), media.next()) => {989 outcome990 .context("timed out waiting for spectator media")??991 .context("spectator media ended before requested frames")?992 }993 };994 let MoqProbeOutcome::Frame { frame, .. } = outcome else {995 continue;996 };997 if let Some((width, height)) = expected_size {998 ensure!(999 frame.header.width == width && frame.header.height == height,1000 "spectator media dimensions did not match {width}x{height}"1001 );1002 }1003 if let Some(previous) = last_sequence {1004 ensure!(1005 frame.header.sequence > previous,1006 "spectator media sequence regressed"1007 );1008 }1009 last_sequence = Some(frame.header.sequence);1010 accepted += 1;1011 }1012 control_connection.close(0_u32.into(), b"spectator probe complete");1013 println!("probe=ok");1014 println!("transport=iroh-moq");1015 println!("control_alpn=sigil/control/2");1016 println!("media_generation_id={generation_id}");1017 println!("media_broadcast_name={broadcast_name}");1018 println!("roster_viewers={maximum_roster_viewers}");1019 println!("frames={accepted}");1020 println!("spectator=ok");1021 println!("resource_summary_version=1");1022 println!("maximum_roster_viewers={maximum_roster_viewers}");1023 println!("media_progress=monotonic");1024 println!("probe_media_object_capacity={MEDIA_OBJECT_CAPACITY}");1025 return Ok(());1026 }1027 let focus_request = FocusCommandV2 {1028 request_id: 1,1029 action: FocusCommandActionV2::Request,1030 slot: ControllerSlot::ZERO,1031 expected_revision: initial.revision,1032 expected_focus_generation: None,1033 expected_proposal_id: None,1034 };1035 write_client_control_v2(1036 &mut control_send,1037 &ClientControlEnvelopeV2::Focus {1038 command: focus_request,1039 },1040 )1041 .await1042 .context("requesting control v2 focus")?;1043 let focused = read_v2_snapshot_after(1044 &mut control_recv,1045 initial.revision,1046 timeout_seconds,1047 |snapshot| snapshot.self_focus_generation().is_some(),1048 )1049 .await?;1050 let focus_generation = focused1051 .self_focus_generation()1052 .context("focus grant did not identify the requesting viewer")?;10531054 let input_connection = endpoint1055 .connect(address, INPUT_ALPN_V2)1056 .await1057 .context("connecting input v2 protocol")?;1058 let (mut input_send, mut input_recv) = input_connection1059 .open_bi()1060 .await1061 .context("opening input v2 stream")?;1062 let input_hello = InputClientHelloV2::new(1063 "sigil-probe/0.1.0",1064 nonce,1065 session_id,1066 ControllerSlot::ZERO,1067 focus_generation,1068 vec![Capability::InputAck],1069 );1070 write_input_client_hello_v2(&mut input_send, &input_hello)1071 .await1072 .context("writing input v2 hello")?;1073 let input_host = tokio::time::timeout(1074 Duration::from_secs(timeout_seconds),1075 read_input_host_hello_v2(&mut input_recv),1076 )1077 .await1078 .context("timed out waiting for input v2 hello")??1079 .context("host closed during input v2 hello")?;1080 ensure!(input_host.accepted, "host rejected focused input v2 stream");1081 ensure!(1082 input_host.session_id == Some(session_id)1083 && input_host.slot == Some(ControllerSlot::ZERO)1084 && input_host.focus_generation == Some(focus_generation),1085 "input v2 host hello changed its focus binding"1086 );1087 write_input_event_v2(1088 &mut input_send,1089 &InputEventV2 {1090 session_id,1091 slot: ControllerSlot::ZERO,1092 focus_generation,1093 event: InputEvent::Probe,1094 },1095 )1096 .await1097 .context("writing slot-0 input v2 probe")?;1098 let input_ack = tokio::time::timeout(1099 Duration::from_secs(timeout_seconds.min(5)),1100 read_input_ack_v2(&mut input_recv),1101 )1102 .await1103 .context("timed out waiting for input v2 acknowledgement")??1104 .context("host closed before input v2 acknowledgement")?;1105 ensure!(1106 input_ack.session_id == session_id1107 && input_ack.slot == ControllerSlot::ZERO1108 && input_ack.focus_generation == focus_generation1109 && input_ack.ack.sequence == 1,1110 "input v2 acknowledgement changed its focus binding"1111 );11121113 let mut accepted = 0_u32;1114 let mut last_sequence = None;1115 while accepted < frames {1116 let outcome = tokio::time::timeout(Duration::from_secs(timeout_seconds), media.next())1117 .await1118 .context("timed out waiting for control v2 media")??1119 .context("control v2 media ended before requested frames")?;1120 let MoqProbeOutcome::Frame { frame, .. } = outcome else {1121 continue;1122 };1123 if let Some((width, height)) = expected_size {1124 ensure!(1125 frame.header.width == width && frame.header.height == height,1126 "control v2 media dimensions did not match {width}x{height}"1127 );1128 }1129 if let Some(previous) = last_sequence {1130 ensure!(1131 frame.header.sequence > previous,1132 "control v2 media sequence regressed"1133 );1134 }1135 last_sequence = Some(frame.header.sequence);1136 accepted += 1;1137 }11381139 let released = if disconnect_with_focus {1140 None1141 } else {1142 let release = FocusCommandV2 {1143 request_id: 2,1144 action: FocusCommandActionV2::Release,1145 slot: ControllerSlot::ZERO,1146 expected_revision: focused.revision,1147 expected_focus_generation: Some(focus_generation),1148 expected_proposal_id: None,1149 };1150 write_client_control_v2(1151 &mut control_send,1152 &ClientControlEnvelopeV2::Focus { command: release },1153 )1154 .await1155 .context("releasing control v2 focus")?;1156 Some(1157 read_v2_snapshot_after(1158 &mut control_recv,1159 focused.revision,1160 timeout_seconds,1161 |snapshot| matches!(snapshot.focus, FocusStateV2::Vacant { .. }),1162 )1163 .await?,1164 )1165 };1166 input_send.finish().context("finishing input v2 stream")?;1167 input_connection.close(0_u32.into(), b"focus released");1168 control_connection.close(0_u32.into(), b"probe complete");11691170 println!("probe=ok");1171 println!("transport=iroh-moq");1172 println!("control_alpn=sigil/control/2");1173 println!("input_alpn=sigil/input/2");1174 println!("media_authentication=ed25519-v1");1175 println!("media_generation_certified=ok");1176 println!("media_generation_id={generation_id}");1177 println!("media_broadcast_name={broadcast_name}");1178 println!("media_generation_scope=host");1179 println!("subscription_endpoint_binding=ok");1180 println!("roster_viewers={}", initial.viewers.len());1181 println!("frames={accepted}");1182 println!("initial_snapshot_revision={}", initial.revision);1183 println!("focus_grant_revision={}", focused.revision);1184 println!("focus_generation={focus_generation}");1185 println!("slot_0_input=ok");1186 if let Some(released) = released {1187 println!("focus_release_revision={}", released.revision);1188 println!("focus_release=ok");1189 } else {1190 println!("focus_release_revision=disconnect");1191 println!("focus_release=disconnect");1192 }1193 println!("resource_summary_version=1");1194 println!("maximum_roster_viewers={}", initial.viewers.len());1195 println!("media_progress=monotonic");1196 println!("probe_media_object_capacity={MEDIA_OBJECT_CAPACITY}");1197 Ok(())1198}11991200fn focus_holder_identity(1201 snapshot: &sigil_protocol::SessionSnapshotV2,1202) -> Option<(String, u64, u64)> {1203 match &snapshot.focus {1204 FocusStateV2::Held {1205 holder,1206 session_id,1207 focus_generation,1208 ..1209 } => Some((holder.as_str().to_owned(), *session_id, *focus_generation)),1210 FocusStateV2::Vacant { .. } | FocusStateV2::Neutralizing { .. } => None,1211 }1212}12131214fn next_handoff_command(1215 snapshot: &sigil_protocol::SessionSnapshotV2,1216 request_id: u64,1217) -> Option<FocusCommandV2> {1218 let self_viewer = snapshot.self_viewer();1219 match &snapshot.focus {1220 FocusStateV2::Vacant { slot } => Some(FocusCommandV2 {1221 request_id,1222 action: FocusCommandActionV2::Request,1223 slot: *slot,1224 expected_revision: snapshot.revision,1225 expected_focus_generation: None,1226 expected_proposal_id: None,1227 }),1228 FocusStateV2::Held {1229 slot,1230 holder,1231 session_id,1232 focus_generation,1233 } if holder == &snapshot.self_presence_id && *session_id == self_viewer.session_id => {1234 snapshot1235 .focus_proposal1236 .as_ref()1237 .map(|proposal| FocusCommandV2 {1238 request_id,1239 action: FocusCommandActionV2::Approve,1240 slot: *slot,1241 expected_revision: snapshot.revision,1242 expected_focus_generation: Some(*focus_generation),1243 expected_proposal_id: Some(proposal.proposal_id),1244 })1245 }1246 FocusStateV2::Held { slot, .. } if snapshot.focus_proposal.is_none() => {1247 Some(FocusCommandV2 {1248 request_id,1249 action: FocusCommandActionV2::Request,1250 slot: *slot,1251 expected_revision: snapshot.revision,1252 expected_focus_generation: None,1253 expected_proposal_id: None,1254 })1255 }1256 FocusStateV2::Held { .. } | FocusStateV2::Neutralizing { .. } => None,1257 }1258}125912601261async fn run_focus_handoff_participant(1262 control_connection: iroh::endpoint::Connection,1263 mut control_send: iroh::endpoint::SendStream,1264 mut control_recv: iroh::endpoint::RecvStream,1265 mut media: MoqProbeReceiver,1266 mut snapshot: sigil_protocol::SessionSnapshotV2,1267 required_handoffs: u32,1268 frames: u32,1269 timeout_seconds: u64,1270) -> Result<()> {1271 let mut request_id = 1_u64;1272 let mut command_in_flight = false;1273 let mut observed_handoffs = 0_u32;1274 let mut accepted_frames = 0_u32;1275 let mut maximum_roster_viewers = snapshot.viewers.len();1276 let mut last_holder = focus_holder_identity(&snapshot);1277 let mut next_command_at = tokio::time::Instant::now();1278 // Two timeout windows for the handoff exchange, plus the wall time the1279 // frame budget itself needs at a conservative 30fps floor.1280 let deadline = tokio::time::Instant::now()1281 + Duration::from_secs(timeout_seconds.max(1).saturating_mul(2))1282 + Duration::from_secs(u64::from(frames) / 30 + 1);12831284 // Leaving as soon as the handoffs land would shrink the roster underneath1285 // any concurrent same-peer replacement cycle, which then fails on viewer1286 // cardinality for reasons that have nothing to do with admission. Hold the1287 // session for the full frame budget, exactly like a spectator.1288 while observed_handoffs < required_handoffs || accepted_frames < frames {1289 ensure!(1290 tokio::time::Instant::now() < deadline,1291 "timed out after observing {observed_handoffs} of {required_handoffs} focus handoffs \1292 and {accepted_frames} of {frames} frames"1293 );1294 if observed_handoffs < required_handoffs1295 && !command_in_flight1296 && tokio::time::Instant::now() >= next_command_at1297 && let Some(command) = next_handoff_command(&snapshot, request_id)1298 {1299 write_client_control_v2(1300 &mut control_send,1301 &ClientControlEnvelopeV2::Focus { command },1302 )1303 .await1304 .context("writing bounded focus handoff command")?;1305 request_id = request_id1306 .checked_add(1)1307 .context("focus handoff request id exhausted")?;1308 command_in_flight = true;1309 next_command_at = tokio::time::Instant::now() + Duration::from_millis(300);1310 }13111312 tokio::select! {1313 envelope = read_server_control_v2(&mut control_recv) => {1314 let envelope = envelope?1315 .context("host closed during bounded focus handoff proof")?;1316 match envelope {1317 ServerControlEnvelopeV2::Snapshot { snapshot: next } => {1318 ensure!(1319 next.revision >= snapshot.revision,1320 "focus handoff snapshot revision regressed"1321 );1322 if next.revision == snapshot.revision {1323 continue;1324 }1325 if let Some(holder) = focus_holder_identity(&next) {1326 if last_holder.as_ref().is_some_and(|previous| previous != &holder) {1327 observed_handoffs = observed_handoffs.saturating_add(1);1328 }1329 last_holder = Some(holder);1330 }1331 maximum_roster_viewers = maximum_roster_viewers.max(next.viewers.len());1332 snapshot = next;1333 command_in_flight = false;1334 }1335 ServerControlEnvelopeV2::FocusResult { result } => {1336 if !result.accepted {1337 command_in_flight = false;1338 next_command_at = tokio::time::Instant::now() + Duration::from_millis(300);1339 }1340 }1341 }1342 }1343 outcome = media.next() => {1344 if matches!(outcome?, Some(MoqProbeOutcome::Frame { .. })) {1345 accepted_frames = accepted_frames.saturating_add(1);1346 }1347 }1348 _ = tokio::time::sleep_until(deadline) => {1349 bail!(1350 "timed out after observing {observed_handoffs} of {required_handoffs} focus \1351 handoffs and {accepted_frames} of {frames} frames"1352 );1353 }1354 }1355 }13561357 control_connection.close(0_u32.into(), b"focus handoff proof complete");1358 println!("probe=ok");1359 println!("control_alpn=sigil/control/2");1360 println!("focus_handoffs_observed={observed_handoffs}");1361 println!("roster_viewers={maximum_roster_viewers}");1362 println!("focus_handoff_media_frames={accepted_frames}");1363 println!("reset_before_successor=host-ordered");1364 println!("resource_summary_version=1");1365 println!("maximum_roster_viewers={maximum_roster_viewers}");1366 println!("media_progress=monotonic");1367 println!("probe_media_object_capacity={MEDIA_OBJECT_CAPACITY}");1368 Ok(())1369}13701371async fn read_v2_snapshot_after(1372 recv: &mut iroh::endpoint::RecvStream,1373 previous_revision: u64,1374 timeout_seconds: u64,1375 predicate: impl Fn(&sigil_protocol::SessionSnapshotV2) -> bool,1376) -> Result<sigil_protocol::SessionSnapshotV2> {1377 let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_seconds);1378 loop {1379 let envelope = tokio::time::timeout_at(deadline, read_server_control_v2(recv))1380 .await1381 .context("timed out waiting for control v2 state")??1382 .context("host closed before control v2 state arrived")?;1383 match envelope {1384 ServerControlEnvelopeV2::Snapshot { snapshot } => {1385 ensure!(1386 snapshot.revision > previous_revision,1387 "control v2 snapshot revision did not advance"1388 );1389 if predicate(&snapshot) {1390 return Ok(snapshot);1391 }1392 }1393 ServerControlEnvelopeV2::FocusResult { result } => {1394 ensure!(1395 result.accepted,1396 "host rejected control v2 focus command: {}",1397 result.message.as_deref().unwrap_or("unspecified reason")1398 );1399 }1400 }1401 }1402}14031404fn classify_moq_probe_group(previous: Option<u64>, current: u64) -> Result<bool> {1405 let Some(previous) = previous else {1406 return Ok(false);1407 };1408 ensure!(1409 current > previous,1410 "upstream MoQ group sequence did not increase: previous={previous}, current={current}"1411 );1412 Ok(previous.checked_add(1) != Some(current))1413}14141415fn validate_moq_probe_object_bounds(1416 group_sequence: u64,1417 object_count: usize,1418 object_bytes: usize,1419 object_len: usize,1420) -> Result<usize> {1421 let maximum_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;1422 ensure!(1423 object_count < maximum_objects,1424 "upstream MoQ group {group_sequence} exceeded {maximum_objects} objects"1425 );1426 let next_bytes = object_bytes1427 .checked_add(object_len)1428 .context("upstream MoQ group byte count overflowed")?;1429 ensure!(1430 next_bytes <= MAX_MEDIA_GROUP_BYTES_V3,1431 "upstream MoQ group {group_sequence} exceeded {MAX_MEDIA_GROUP_BYTES_V3} bytes"1432 );1433 Ok(next_bytes)1434}14351436fn validate_moq_probe_frame(1437 group_sequence: u64,1438 first_object: bool,1439 last_frame_sequence: Option<u64>,1440 frame: &MediaFrame,1441) -> Result<bool> {1442 if first_object {1443 ensure!(1444 frame.header.codec == MediaCodec::H2641445 && frame.header.flags.contains(FrameFlags::KEYFRAME)1446 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),1447 "upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"1448 );1449 // A native group cancellation is itself the transport discontinuity.1450 // Portal marks the first configured IDR in the replacement group as a1451 // decoder discontinuity even when the embedded compatibility header1452 // does not redundantly carry the v3 DISCONTINUITY flag.1453 }1454 let contiguous = last_frame_sequence1455 .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));1456 ensure!(1457 last_frame_sequence.is_none_or(|previous| frame.header.sequence > previous),1458 "upstream MoQ group {group_sequence} contains a non-monotonic frame sequence"1459 );1460 ensure!(1461 first_object || contiguous,1462 "upstream MoQ group {group_sequence} contains a non-contiguous frame sequence"1463 );1464 Ok(contiguous)1465}14661467fn moq_probe_error_is_recoverable(error: &moq_net::Error) -> bool {1468 matches!(1469 error,1470 moq_net::Error::Cancel1471 | moq_net::Error::Old1472 | moq_net::Error::Timeout1473 | moq_net::Error::Dropped1474 | moq_net::Error::CacheFull1475 | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)1476 )1477}14781479struct AcceptedMedia {1480 flags: FrameFlags,1481 width: u16,1482 height: u16,1483 sequence: u64,1484 capture_timestamp_us: u64,1485 payload_len: usize,1486 v3_group_id: Option<u64>,1487}14881489impl From<MediaObjectV3> for AcceptedMedia {1490 fn from(object: MediaObjectV3) -> Self {1491 Self {1492 flags: object.header.flags,1493 width: object.header.width,1494 height: object.header.height,1495 sequence: object.header.sequence,1496 capture_timestamp_us: object.header.capture_timestamp_us,1497 payload_len: object.payload.len(),1498 v3_group_id: Some(object.header.group_id),1499 }1500 }1501}150215031504struct ResolutionStallEvidence {1505 native_dimensions: (u16, u16),1506 reduced_dimensions: (u16, u16),1507 stall_sequence: u64,1508 fresh_duration: Duration,1509 stall_input_ack_micros: u64,1510 resume_input_ack_micros: u64,1511 resume_media_micros: u64,1512 resume_sequence_advance: u64,1513 recovery: Option<ResolutionRecoveryEvidence>,1514}151515161517struct ResolutionRecoveryEvidence {1518 windows: u64,1519 trough_kbps: u32,1520 final_kbps: u32,1521 increase_observed: bool,1522 final_dimensions: (u16, u16),1523 restored_native: bool,1524}15251526struct ResolutionStallGate {1527 ready: PathBuf,1528 stalled: PathBuf,1529 resume_ready: PathBuf,1530 resumed: PathBuf,1531}15321533impl ResolutionStallGate {1534 fn new(directory: &Path) -> Result<Self> {1535 ensure!(1536 directory.is_dir(),1537 "resolution stall gate is not a directory"1538 );1539 let gate = Self {1540 ready: directory.join("ready"),1541 stalled: directory.join("stalled"),1542 resume_ready: directory.join("resume-ready"),1543 resumed: directory.join("resumed"),1544 };1545 for marker in [1546 &gate.ready,1547 &gate.stalled,1548 &gate.resume_ready,1549 &gate.resumed,1550 ] {1551 ensure!(1552 !marker.exists(),1553 "resolution stall gate marker already exists: {}",1554 marker.display()1555 );1556 }1557 Ok(gate)1558 }1559}15601561fn complete_feedback_report(1562 report_id: u64,1563 flags: MediaFeedbackFlags,1564 last_sequence: u64,1565) -> MediaFeedbackReportV1 {1566 MediaFeedbackReportV1 {1567 report_id,1568 interval_ms: 1_000,1569 flags,1570 last_sequence: Some(last_sequence),1571 transport_dropped_delta: 0,1572 frontend_dropped_delta: 0,1573 decoder_dropped_delta: 0,1574 presenter_dropped_delta: 0,1575 frontend_queue_depth: 0,1576 frontend_queue_capacity: 4,1577 decode_queue_depth: 0,1578 decode_queue_capacity: 4,1579 presenter_queue_depth: 0,1580 presenter_queue_capacity: 2,1581 transport_delivery_p95_ms: Some(10),1582 decode_p95_ms: Some(3),1583 presentation_p95_ms: Some(5),1584 }1585}15861587fn resolution_stall_reduced_dimensions(native: (u16, u16)) -> Result<(u16, u16)> {1588 ensure!(1589 native.0.is_multiple_of(4) && native.1.is_multiple_of(4),1590 "native dimensions must be divisible by four for an exact three-quarter tier"1591 );1592 let reduced = (native.0 / 4 * 3, native.1 / 4 * 3);1593 ensure!(1594 reduced.0 >= 64 && reduced.1 >= 64,1595 "three-quarter dimensions are below the H.264 minimum"1596 );1597 ensure!(1598 reduced.0.is_multiple_of(2) && reduced.1.is_multiple_of(2),1599 "three-quarter H.264 dimensions must be even"1600 );1601 ensure!(1602 u32::from(native.0) * u32::from(reduced.1) == u32::from(native.1) * u32::from(reduced.0),1603 "three-quarter dimensions must preserve the native aspect ratio"1604 );1605 Ok(reduced)1606}16071608async fn exchange_feedback_report(1609 send: &mut iroh::endpoint::SendStream,1610 recv: &mut iroh::endpoint::RecvStream,1611 report: &MediaFeedbackReportV1,1612) -> Result<AdaptiveBitrateDecisionV1> {1613 tokio::time::timeout(1614 RESOLUTION_STALL_FEEDBACK_TIMEOUT,1615 write_media_feedback_report_v1(send, report),1616 )1617 .await1618 .context("timed out writing media feedback")??;1619 let decision = tokio::time::timeout(1620 RESOLUTION_STALL_FEEDBACK_TIMEOUT,1621 read_adaptive_bitrate_decision_v1(recv),1622 )1623 .await1624 .context("timed out waiting for media feedback decision")??1625 .context("host closed the media feedback stream")?;1626 ensure!(1627 decision.report_id == report.report_id,1628 "feedback decision report ID {} does not match report {}",1629 decision.report_id,1630 report.report_id1631 );1632 Ok(decision)1633}16341635async fn run_feedback_smoke(1636 endpoint: &Endpoint,1637 address: EndpointAddr,1638 nonce: [u8; 16],1639 session_id: u64,1640 last_sequence: u64,1641) -> Result<(AdaptiveBitrateDecisionV1, &'static str, Option<f64>)> {1642 let connection = endpoint1643 .connect(address, MEDIA_FEEDBACK_ALPN_V1)1644 .await1645 .context("connecting media feedback protocol")?;1646 let (mut send, mut recv) = connection1647 .open_bi()1648 .await1649 .context("opening media feedback stream")?;1650 let negotiation = negotiate(1651 &mut send,1652 &mut recv,1653 nonce,1654 vec![Capability::VideoH264],1655 Capability::VideoH264,1656 "media feedback",1657 None,1658 )1659 .await?;1660 ensure!(1661 negotiation.session_id == session_id,1662 "media and feedback session IDs differ"1663 );16641665 let report = complete_feedback_report(1, MediaFeedbackFlags::NONE, last_sequence);1666 let decision = exchange_feedback_report(&mut send, &mut recv, &report).await?;1667 let (path_mode, path_rtt_ms) = selected_path_diagnostics(&connection);1668 send.finish().context("finishing media feedback stream")?;1669 connection.close(0_u32.into(), b"feedback probe complete");1670 Ok((decision, path_mode, path_rtt_ms))1671}16721673impl AdaptiveProfile {1674 const fn label(self) -> &'static str {1675 match self {1676 Self::Clean => "clean",1677 Self::FloorBreach => "floor-breach",1678 Self::Stale => "stale",1679 }1680 }1681}16821683struct AdaptiveProfileEvidence {1684 minimum_kbps: u32,1685 final_kbps: u32,1686 survivor_recovery: bool,1687 floor_detach: bool,1688}16891690async fn run_v2_adaptive_profile(1691 endpoint: &Endpoint,1692 address: EndpointAddr,1693 nonce: [u8; 16],1694 session_id: u64,1695 profile: AdaptiveProfile,1696) -> Result<AdaptiveProfileEvidence> {1697 let connection = endpoint1698 .connect(address, MEDIA_FEEDBACK_ALPN_V1)1699 .await1700 .context("connecting control-v2 adaptive feedback profile")?;1701 let (mut send, mut recv) = connection1702 .open_bi()1703 .await1704 .context("opening control-v2 adaptive feedback stream")?;1705 let negotiation = negotiate(1706 &mut send,1707 &mut recv,1708 nonce,1709 vec![Capability::VideoH264],1710 Capability::VideoH264,1711 "control-v2 adaptive feedback",1712 None,1713 )1714 .await?;1715 ensure!(1716 negotiation.session_id == session_id,1717 "control-v2 media and adaptive feedback session IDs differ"1718 );17191720 let report_count = match profile {1721 AdaptiveProfile::Clean => 36,1722 AdaptiveProfile::FloorBreach => 32,1723 AdaptiveProfile::Stale => 5,1724 };1725 let mut minimum_kbps = u32::MAX;1726 let mut maximum_kbps = 0_u32;1727 let mut final_kbps = 0_u32;1728 let mut saw_stale = false;1729 let mut floor_detach = false;1730 let mut final_reason_bits = 0_u16;1731 for report_id in 1..=report_count {1732 let mut report = complete_feedback_report(report_id, MediaFeedbackFlags::NONE, report_id);1733 match profile {1734 AdaptiveProfile::Clean => {}1735 AdaptiveProfile::FloorBreach => {1736 report.flags = MediaFeedbackFlags::RESYNC_ACTIVE;1737 report.transport_dropped_delta = 8;1738 report.frontend_dropped_delta = 8;1739 report.decoder_dropped_delta = 8;1740 report.presenter_dropped_delta = 8;1741 report.frontend_queue_depth = report.frontend_queue_capacity;1742 report.decode_queue_depth = report.decode_queue_capacity;1743 report.presenter_queue_depth = report.presenter_queue_capacity;1744 report.transport_delivery_p95_ms = Some(250);1745 report.decode_p95_ms = Some(125);1746 report.presentation_p95_ms = Some(125);1747 }1748 AdaptiveProfile::Stale => {1749 report.interval_ms = 5_000;1750 }1751 }1752 let decision = match exchange_feedback_report(&mut send, &mut recv, &report).await {1753 Ok(decision) => decision,1754 Err(error) if profile == AdaptiveProfile::FloorBreach && minimum_kbps <= 1_000 => {1755 floor_detach = true;1756 eprintln!("[probe] floor-breach feedback stream detached as expected: {error}");1757 break;1758 }1759 Err(error) => return Err(error).context("running control-v2 adaptive profile"),1760 };1761 minimum_kbps = minimum_kbps.min(decision.target_kbps);1762 maximum_kbps = maximum_kbps.max(decision.target_kbps);1763 final_kbps = decision.target_kbps;1764 final_reason_bits = decision.reasons.bits();1765 saw_stale |= decision1766 .reasons1767 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE);1768 tokio::time::sleep(Duration::from_millis(800)).await;1769 }1770 ensure!(1771 minimum_kbps != u32::MAX,1772 "adaptive profile received no decisions"1773 );1774 let survivor_recovery = profile == AdaptiveProfile::Clean1775 && minimum_kbps < maximum_kbps1776 && final_kbps > minimum_kbps;1777 match profile {1778 AdaptiveProfile::Clean => ensure!(1779 survivor_recovery,1780 "clean survivor did not observe aggregate downshift and recovery: min={minimum_kbps}, max={maximum_kbps}, final={final_kbps}, reasons={final_reason_bits}"1781 ),1782 AdaptiveProfile::FloorBreach => ensure!(1783 floor_detach && minimum_kbps <= 1_000,1784 "floor-breach viewer was not detached after bounded recovery"1785 ),1786 AdaptiveProfile::Stale => ensure!(saw_stale, "stale profile was not classified stale"),1787 }1788 if profile != AdaptiveProfile::FloorBreach {1789 send.finish()1790 .context("finishing control-v2 adaptive feedback profile")?;1791 connection.close(0_u32.into(), b"adaptive profile complete");1792 }1793 Ok(AdaptiveProfileEvidence {1794 minimum_kbps,1795 final_kbps,1796 survivor_recovery,1797 floor_detach,1798 })1799}18001801async fn next_resolution_stall_moq_frame(receiver: &mut MoqProbeReceiver) -> Result<AcceptedMedia> {1802 loop {1803 match receiver.next().await? {1804 Some(MoqProbeOutcome::Frame {1805 frame,1806 group_sequence,1807 ..1808 }) => {1809 return Ok(AcceptedMedia {1810 flags: frame.header.flags,1811 width: frame.header.width,1812 height: frame.header.height,1813 sequence: frame.header.sequence,1814 capture_timestamp_us: frame.header.capture_timestamp_us,1815 payload_len: frame.payload.len(),1816 v3_group_id: Some(group_sequence),1817 });1818 }1819 Some(MoqProbeOutcome::Dropped) => {}1820 None => bail!("host closed upstream MoQ during resolution-stall smoke"),1821 }1822 }1823}18241825fn create_resolution_stall_marker(path: &Path, contents: &str) -> Result<()> {1826 let mut options = OpenOptions::new();1827 options.write(true).create_new(true);1828 1829 options.mode(0o600);1830 let mut marker = options1831 .open(path)1832 .with_context(|| format!("creating resolution stall marker {}", path.display()))?;1833 marker1834 .write_all(contents.as_bytes())1835 .with_context(|| format!("writing resolution stall marker {}", path.display()))?;1836 marker1837 .sync_all()1838 .with_context(|| format!("syncing resolution stall marker {}", path.display()))?;1839 Ok(())1840}18411842async fn wait_for_resolution_stall_marker(path: &Path, timeout: Duration) -> Result<()> {1843 tokio::time::timeout(timeout, async {1844 loop {1845 match fs::metadata(path) {1846 Ok(metadata) => {1847 ensure!(1848 metadata.is_file(),1849 "resolution stall gate marker is not a file: {}",1850 path.display()1851 );1852 return Ok(());1853 }1854 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {1855 tokio::time::sleep(RESOLUTION_STALL_GATE_POLL_INTERVAL).await;1856 }1857 Err(error) => {1858 return Err(error).with_context(|| {1859 format!("reading resolution stall gate marker {}", path.display())1860 });1861 }1862 }1863 }1864 })1865 .await1866 .with_context(|| {1867 format!(1868 "timed out waiting for resolution stall marker {}",1869 path.display()1870 )1871 })?1872}18731874async fn send_resolution_stall_input_probe(1875 send: &mut iroh::endpoint::SendStream,1876 recv: &mut iroh::endpoint::RecvStream,1877 expected_ack: &mut u64,1878) -> Result<u64> {1879 let started = Instant::now();1880 write_input_event(send, &InputEvent::Probe)1881 .await1882 .context("writing input probe during resolution stall")?;1883 *expected_ack = expected_ack.saturating_add(1);1884 tokio::time::timeout(1885 RESOLUTION_STALL_RECOVERY_TIMEOUT,1886 read_expected_input_ack(recv, 2, *expected_ack),1887 )1888 .await1889 .context("input acknowledgement exceeded the resolution-stall bound")??;1890 Ok(u64::try_from(started.elapsed().as_micros()).unwrap_or(u64::MAX))1891}189218931894async fn run_resolution_stall_smoke(1895 gate_directory: &Path,1896 receiver: &mut MoqProbeReceiver,1897 feedback_send: &mut iroh::endpoint::SendStream,1898 feedback_recv: &mut iroh::endpoint::RecvStream,1899 input_send: &mut iroh::endpoint::SendStream,1900 input_recv: &mut iroh::endpoint::RecvStream,1901 expected_ack: &mut u64,1902 expected_native: Option<(u16, u16)>,1903 timeout: Duration,1904 recovery_windows: u64,1905) -> Result<ResolutionStallEvidence> {1906 let gate = ResolutionStallGate::new(gate_directory)?;1907 let initial = tokio::time::timeout(timeout, next_resolution_stall_moq_frame(receiver))1908 .await1909 .context("timed out waiting for initial resolution-stall media")??;1910 ensure!(1911 initial.flags.contains(FrameFlags::KEYFRAME)1912 && initial.flags.contains(FrameFlags::CODEC_CONFIG),1913 "resolution-stall smoke did not begin at a configured IDR"1914 );1915 let native_dimensions = (initial.width, initial.height);1916 if let Some(expected) = expected_native {1917 ensure!(1918 native_dimensions == expected,1919 "expected {}x{} native media but received {}x{}",1920 expected.0,1921 expected.1,1922 native_dimensions.0,1923 native_dimensions.11924 );1925 }1926 let reduced_dimensions = resolution_stall_reduced_dimensions(native_dimensions)?;19271928 let mut report_id = 1_u64;1929 let seed = complete_feedback_report(report_id, MediaFeedbackFlags::NONE, initial.sequence);1930 let _ = exchange_feedback_report(feedback_send, feedback_recv, &seed).await?;1931 tokio::time::sleep(RESOLUTION_STALL_FEEDBACK_INTERVAL).await;19321933 report_id += 1;1934 let pressure = complete_feedback_report(1935 report_id,1936 MediaFeedbackFlags::RESYNC_ACTIVE,1937 initial.sequence,1938 );1939 let pressure_decision =1940 exchange_feedback_report(feedback_send, feedback_recv, &pressure).await?;1941 let mut trough_kbps = pressure_decision.target_kbps;1942 ensure!(1943 pressure_decision1944 .reasons1945 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE),1946 "resolution-stall pressure report did not register receiver pressure"1947 );19481949 let reduced = tokio::time::timeout(timeout, async {1950 loop {1951 let frame = next_resolution_stall_moq_frame(receiver).await?;1952 let dimensions = (frame.width, frame.height);1953 ensure!(1954 dimensions == native_dimensions || dimensions == reduced_dimensions,1955 "resolution-stall media changed to unexpected dimensions {}x{}",1956 dimensions.0,1957 dimensions.11958 );1959 if dimensions == reduced_dimensions1960 && frame.flags.contains(FrameFlags::KEYFRAME)1961 && frame.flags.contains(FrameFlags::CODEC_CONFIG)1962 {1963 break Ok::<_, anyhow::Error>(frame);1964 }1965 }1966 })1967 .await1968 .context("timed out waiting for pressure-triggered reduced configured IDR")??;1969 ensure!(1970 reduced.sequence > initial.sequence,1971 "reduced configured IDR did not advance media sequence"1972 );19731974 create_resolution_stall_marker(1975 &gate.ready,1976 &format!(1977 "native={}x{}\nreduced={}x{}\nsequence={}\n",1978 native_dimensions.0,1979 native_dimensions.1,1980 reduced_dimensions.0,1981 reduced_dimensions.1,1982 reduced.sequence1983 ),1984 )?;1985 println!("resolution_stall_ready=1");1986 println!("resolution_stall_ready_sequence={}", reduced.sequence);1987 std::io::stdout()1988 .flush()1989 .context("flushing resolution-stall readiness marker")?;1990 wait_for_resolution_stall_marker(&gate.stalled, timeout).await?;19911992 report_id += 1;1993 let reseed = complete_feedback_report(report_id, MediaFeedbackFlags::NONE, reduced.sequence);1994 let _ = exchange_feedback_report(feedback_send, feedback_recv, &reseed).await?;1995 let mut last_feedback_at = Instant::now();1996 let mut fresh_started = None;1997 let mut stall_input_ack_micros = None;1998 for index in 0..RESOLUTION_STALL_FRESH_REPORTS {1999 tokio::time::sleep_until(tokio::time::Instant::from_std(2000 last_feedback_at + RESOLUTION_STALL_FEEDBACK_INTERVAL,2001 ))2002 .await;2003 report_id += 1;2004 let report =2005 complete_feedback_report(report_id, MediaFeedbackFlags::NONE, reduced.sequence);2006 let sent_at = Instant::now();2007 fresh_started.get_or_insert(sent_at);2008 let decision = exchange_feedback_report(feedback_send, feedback_recv, &report).await?;2009 trough_kbps = trough_kbps.min(decision.target_kbps);2010 ensure!(2011 decision.state != AdaptiveBitrateStateV1::Increase,2012 "no-progress feedback unexpectedly increased adaptive bitrate"2013 );2014 ensure!(2015 !decision2016 .reasons2017 .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY),2018 "no-progress feedback was incorrectly classified as clean recovery"2019 );2020 last_feedback_at = sent_at;2021 if index + 1 == RESOLUTION_STALL_FRESH_REPORTS / 2 {2022 stall_input_ack_micros = Some(2023 send_resolution_stall_input_probe(input_send, input_recv, expected_ack).await?,2024 );2025 }2026 }2027 let fresh_duration = fresh_started2028 .context("resolution-stall fresh feedback interval did not start")?2029 .elapsed();2030 ensure!(2031 fresh_duration > Duration::from_secs(8),2032 "resolution-stall fresh feedback lasted only {}ms",2033 fresh_duration.as_millis()2034 );20352036 create_resolution_stall_marker(2037 &gate.resume_ready,2038 &format!(2039 "fresh_reports={RESOLUTION_STALL_FRESH_REPORTS}\nfresh_duration_ms={}\n",2040 fresh_duration.as_millis()2041 ),2042 )?;2043 println!("resolution_stall_resume_ready=1");2044 std::io::stdout()2045 .flush()2046 .context("flushing resolution-stall resume marker")?;2047 wait_for_resolution_stall_marker(&gate.resumed, timeout).await?;20482049 let resume_input_ack_micros =2050 send_resolution_stall_input_probe(input_send, input_recv, expected_ack).await?;2051 let media_resumed_at = Instant::now();2052 let resumed = tokio::time::timeout(2053 RESOLUTION_STALL_RECOVERY_TIMEOUT,2054 next_resolution_stall_moq_frame(receiver),2055 )2056 .await2057 .context("media did not resume within the resolution-stall bound")??;2058 ensure!(2059 (resumed.width, resumed.height) == reduced_dimensions,2060 "media resumed at {}x{} instead of reduced {}x{}",2061 resumed.width,2062 resumed.height,2063 reduced_dimensions.0,2064 reduced_dimensions.12065 );2066 let resume_sequence_advance = resumed2067 .sequence2068 .checked_sub(reduced.sequence)2069 .context("resumed media sequence did not advance")?;2070 ensure!(2071 resume_sequence_advance > 0,2072 "resumed media sequence did not advance"2073 );2074 let resume_media_micros =2075 u64::try_from(media_resumed_at.elapsed().as_micros()).unwrap_or(u64::MAX);20762077 // Proving the descent is only half of the contract. Feed clean, genuinely2078 // advancing evidence for long enough to clear the clean-window and2079 // cooldown counters, then require the controller to climb back out.2080 let recovery = if recovery_windows > 0 {2081 let mut latest_sequence = resumed.sequence;2082 let mut latest_dimensions = (resumed.width, resumed.height);2083 let mut final_kbps = trough_kbps;2084 let mut increase_observed = false;2085 let mut last_feedback_at = Instant::now();2086 for _ in 0..recovery_windows {2087 // Drain as fast as frames arrive until the next feedback deadline.2088 // Consuming a single frame per window would starve the receiver and2089 // manufacture exactly the cancellation backpressure this loop2090 // exists to prove has cleared, so the controller would correctly2091 // refuse to climb and the test would be measuring its own harness.2092 let deadline = last_feedback_at + RESOLUTION_STALL_FEEDBACK_INTERVAL;2093 loop {2094 let now = Instant::now();2095 if now >= deadline {2096 break;2097 }2098 match tokio::time::timeout(2099 deadline - now,2100 next_resolution_stall_moq_frame(receiver),2101 )2102 .await2103 {2104 Ok(Ok(frame)) => {2105 latest_sequence = frame.sequence;2106 latest_dimensions = (frame.width, frame.height);2107 }2108 Ok(Err(error)) => return Err(error),2109 Err(_) => break,2110 }2111 }2112 report_id += 1;2113 let report =2114 complete_feedback_report(report_id, MediaFeedbackFlags::NONE, latest_sequence);2115 let sent_at = Instant::now();2116 let decision = exchange_feedback_report(feedback_send, feedback_recv, &report).await?;2117 last_feedback_at = sent_at;2118 final_kbps = decision.target_kbps;2119 if decision.state == AdaptiveBitrateStateV1::Increase {2120 increase_observed = true;2121 }2122 }2123 ensure!(2124 increase_observed,2125 "adaptive bitrate never increased across {recovery_windows} clean recovery windows (trough {trough_kbps} kbps, final {final_kbps} kbps)"2126 );2127 ensure!(2128 final_kbps > trough_kbps,2129 "adaptive bitrate recovered no ground: trough {trough_kbps} kbps, final {final_kbps} kbps"2130 );2131 Some(ResolutionRecoveryEvidence {2132 windows: recovery_windows,2133 trough_kbps,2134 final_kbps,2135 increase_observed,2136 final_dimensions: latest_dimensions,2137 restored_native: latest_dimensions == native_dimensions,2138 })2139 } else {2140 None2141 };21422143 Ok(ResolutionStallEvidence {2144 native_dimensions,2145 reduced_dimensions,2146 stall_sequence: reduced.sequence,2147 fresh_duration,2148 stall_input_ack_micros: stall_input_ack_micros2149 .context("resolution-stall input acknowledgement was not measured")?,2150 resume_input_ack_micros,2151 resume_media_micros,2152 resume_sequence_advance,2153 recovery,2154 })2155}215621572158async fn main() -> Result<()> {2159 let args = Args::parse();2160 ensure!(2161 args.timeout_seconds > 0,2162 "--timeout-seconds must be greater than zero"2163 );2164 ensure!(2165 !args.keyframe_smoke || args.frames >= KEYFRAME_SMOKE_MINIMUM_FRAMES,2166 "--keyframe-smoke requires at least {KEYFRAME_SMOKE_MINIMUM_FRAMES} accepted frames"2167 );2168 ensure!(2169 args.slow_consumer_ms.is_none() || args.frames >= SLOW_CONSUMER_MINIMUM_FRAMES,2170 "--slow-consumer-ms requires at least {SLOW_CONSUMER_MINIMUM_FRAMES} accepted frames"2171 );21722173 let secret = match args.identity.as_deref() {2174 Some(path) => identity::load(path).context("loading persistent probe identity")?,2175 None => SecretKey::generate(),2176 };2177 let mut required_grants = InvitationGrants::VIEW;2178 if args.pointer_smoke {2179 required_grants = required_grants.union(InvitationGrants::POINTER_KEYBOARD);2180 }2181 if args.gamepad_smoke {2182 required_grants = required_grants.union(InvitationGrants::GAMEPAD);2183 }2184 let invitation = args2185 .invitation2186 .as_deref()2187 .map(|path| load_invitation_file(path, args.node_id, secret.public(), required_grants))2188 .transpose()?;2189 let mut nonce = [0_u8; 16];2190 getrandom::fill(&mut nonce).context("generating handshake nonce")?;2191 let mut endpoint_builder = Endpoint::builder(presets::N0).secret_key(secret);2192 if args.relay_only {2193 endpoint_builder = endpoint_builder.clear_ip_transports();2194 }2195 let endpoint = endpoint_builder2196 .bind()2197 .await2198 .context("binding probe endpoint")?;2199 let _ = tokio::time::timeout(Duration::from_secs(10), endpoint.online()).await;2200 let address = EndpointAddr::new(args.node_id);2201 if args.control_v2 {2202 return run_control_v2_smoke(2203 &endpoint,2204 address,2205 nonce,2206 invitation.as_deref(),2207 args.frames,2208 args.timeout_seconds,2209 args.expect_size,2210 args.spectator,2211 args.disconnect_with_focus,2212 args.focus_handoffs,2213 args.adaptive_profile,2214 )2215 .await;2216 }2217 let media_transport = if args.media_v3 {2218 MediaTransport::GroupedV32219 } else {2220 MediaTransport::UpstreamMoq2221 };22222223 let media_connection = endpoint2224 .connect(address.clone(), media_transport.alpn())2225 .await2226 .context("connecting media protocol")?;2227 let (mut media_send, mut media_recv) = media_connection2228 .open_bi()2229 .await2230 .context("opening media stream")?;2231 let handshake_name = if media_transport == MediaTransport::UpstreamMoq {2232 "control"2233 } else {2234 "media"2235 };2236 let media_negotiation = negotiate(2237 &mut media_send,2238 &mut media_recv,2239 nonce,2240 vec![Capability::VideoH264],2241 Capability::VideoH264,2242 handshake_name,2243 invitation.as_deref(),2244 )2245 .await?;2246 let session_id = media_negotiation.session_id;2247 // Sigil finishes its response half after HostHello. Keep our send half2248 // alive as the bounded keyframe-control stream.2249 drop(media_recv);22502251 let mut moq_receiver = if media_transport == MediaTransport::UpstreamMoq {2252 Some(2253 MoqProbeReceiver::connect(2254 &endpoint,2255 address.clone(),2256 session_id,2257 session_id,2258 None,2259 Duration::from_secs(args.timeout_seconds),2260 None,2261 )2262 .await?,2263 )2264 } else {2265 None2266 };22672268 let input_connection = endpoint2269 .connect(address.clone(), INPUT_ALPN_V1)2270 .await2271 .context("connecting input protocol")?;2272 let (mut input_send, mut input_recv) = input_connection2273 .open_bi()2274 .await2275 .context("opening input stream")?;2276 let mut input_offers = vec![Capability::InputAck];2277 if args.pointer_smoke {2278 input_offers.push(Capability::RelativePointer);2279 }2280 if args.pointer_feedback_smoke {2281 input_offers.push(Capability::PointerPositionFeedback);2282 input_offers.push(Capability::PointerVisibilityFeedback);2283 }2284 if args.gamepad_smoke {2285 input_offers.push(Capability::Gamepad);2286 }2287 let input_negotiation = negotiate(2288 &mut input_send,2289 &mut input_recv,2290 nonce,2291 input_offers,2292 Capability::InputAck,2293 "input",2294 None,2295 )2296 .await?;2297 ensure!(2298 session_id == input_negotiation.session_id,2299 "media and input session IDs differ"2300 );2301 if args.pointer_feedback_smoke {2302 ensure!(2303 input_negotiation2304 .capabilities2305 .contains(&Capability::PointerPositionFeedback)2306 && input_negotiation2307 .capabilities2308 .contains(&Capability::PointerVisibilityFeedback),2309 "host did not accept the required pointer feedback capabilities"2310 );2311 let initial_feedback =2312 read_expected_input_ack(&mut input_recv, args.timeout_seconds, 0).await?;2313 ensure!(2314 initial_feedback.pointer_position.is_some(),2315 "host pointer feedback is unavailable"2316 );2317 ensure!(2318 initial_feedback.pointer_visible.is_some(),2319 "host omitted negotiated pointer visibility feedback"2320 );2321 }2322 let input_started = Instant::now();2323 write_input_event(&mut input_send, &InputEvent::Probe)2324 .await2325 .context("writing input probe")?;2326 read_expected_input_ack(&mut input_recv, args.timeout_seconds, 1).await?;2327 let input_ack_micros = input_started.elapsed().as_micros();2328 let mut expected_ack = 1_u64;2329 let mut pointer_sync_position = None;2330 let mut pointer_motion_position = None;23312332 if args.pointer_smoke {2333 ensure!(2334 input_negotiation2335 .capabilities2336 .contains(&Capability::RelativePointer),2337 "host did not accept the required relative pointer capability"2338 );2339 let pointer_dimensions = media_negotiation.pointer_surface_dimensions.context(2340 "host did not advertise pointer surface dimensions required by --pointer-smoke",2341 )?;2342 let [position_sync, relative_motion, click] =2343 pointer_smoke_events(Some(pointer_dimensions))?;2344 let [expected_sync_position, expected_motion_position] =2345 pointer_smoke_expected_positions(pointer_dimensions)?;2346 write_input_event(&mut input_send, &position_sync)2347 .await2348 .context("writing pointer position synchronization")?;2349 expected_ack += 1;2350 let sync_ack =2351 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;2352 if args.pointer_feedback_smoke {2353 pointer_sync_position = Some(2354 confirm_pointer_position(2355 &mut input_send,2356 &mut input_recv,2357 &mut expected_ack,2358 sync_ack,2359 expected_sync_position,2360 Duration::from_secs(args.timeout_seconds.min(5)),2361 )2362 .await2363 .context("verifying compositor position after pointer synchronization")?,2364 );2365 }2366 write_input_event(&mut input_send, &relative_motion)2367 .await2368 .context("writing relative pointer smoke motion")?;2369 expected_ack += 1;2370 let motion_ack =2371 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;2372 if args.pointer_feedback_smoke {2373 pointer_motion_position = Some(2374 confirm_pointer_position(2375 &mut input_send,2376 &mut input_recv,2377 &mut expected_ack,2378 motion_ack,2379 expected_motion_position,2380 Duration::from_secs(args.timeout_seconds.min(5)),2381 )2382 .await2383 .context("verifying compositor position after relative pointer motion")?,2384 );2385 }2386 write_input_event(&mut input_send, &click)2387 .await2388 .context("writing pointer smoke click")?;2389 expected_ack += 1;2390 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;2391 }23922393 if args.gamepad_smoke {2394 ensure!(2395 input_negotiation2396 .capabilities2397 .contains(&Capability::Gamepad),2398 "host did not accept the required gamepad capability"2399 );2400 write_input_event(2401 &mut input_send,2402 &InputEvent::Gamepad {2403 state: GamepadState {2404 a: true,2405 right_shoulder: true,2406 dpad_right: true,2407 left_x: GAMEPAD_AXIS_MAX,2408 right_y: GAMEPAD_AXIS_MIN,2409 left_trigger: GAMEPAD_TRIGGER_MAX,2410 right_trigger: GAMEPAD_TRIGGER_MAX,2411 ..GamepadState::default()2412 },2413 },2414 )2415 .await2416 .context("writing non-neutral gamepad smoke snapshot")?;2417 expected_ack += 1;2418 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;2419 tokio::time::sleep(Duration::from_millis(100)).await;2420 write_input_event(2421 &mut input_send,2422 &InputEvent::Gamepad {2423 state: GamepadState::default(),2424 },2425 )2426 .await2427 .context("writing neutral gamepad smoke snapshot")?;2428 expected_ack += 1;2429 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;2430 }24312432 if let Some(gate_directory) = args.resolution_stall_smoke.as_deref() {2433 let feedback_connection = endpoint2434 .connect(address.clone(), MEDIA_FEEDBACK_ALPN_V1)2435 .await2436 .context("connecting media feedback protocol")?;2437 let (mut feedback_send, mut feedback_recv) = feedback_connection2438 .open_bi()2439 .await2440 .context("opening media feedback stream")?;2441 let feedback_negotiation = negotiate(2442 &mut feedback_send,2443 &mut feedback_recv,2444 nonce,2445 vec![Capability::VideoH264],2446 Capability::VideoH264,2447 "media feedback",2448 None,2449 )2450 .await?;2451 ensure!(2452 session_id == feedback_negotiation.session_id,2453 "media and feedback session IDs differ"2454 );24552456 let evidence = run_resolution_stall_smoke(2457 gate_directory,2458 moq_receiver2459 .as_mut()2460 .expect("resolution-stall smoke requires upstream MoQ"),2461 &mut feedback_send,2462 &mut feedback_recv,2463 &mut input_send,2464 &mut input_recv,2465 &mut expected_ack,2466 args.expect_size,2467 Duration::from_secs(args.timeout_seconds),2468 args.resolution_recovery_windows,2469 )2470 .await?;24712472 feedback_send2473 .finish()2474 .context("finishing media feedback stream")?;2475 input_send.finish().context("finishing input stream")?;2476 media_send2477 .finish()2478 .context("finishing media request stream")?;2479 feedback_connection.close(0_u32.into(), b"probe complete");2480 input_connection.close(0_u32.into(), b"probe complete");2481 moq_receiver2482 .as_ref()2483 .expect("resolution-stall smoke requires upstream MoQ")2484 .connection()2485 .close(0_u32.into(), b"probe complete");2486 media_connection.close(0_u32.into(), b"probe complete");2487 tokio::time::sleep(Duration::from_millis(50)).await;2488 endpoint.close().await;24892490 println!("probe=ok");2491 println!("session_id={session_id}");2492 println!("resolution_stall_smoke=ok");2493 println!("resolution_stall_session_id={session_id}");2494 println!(2495 "resolution_stall_native_dimensions={}x{}",2496 evidence.native_dimensions.0, evidence.native_dimensions.12497 );2498 println!(2499 "resolution_stall_reduced_dimensions={}x{}",2500 evidence.reduced_dimensions.0, evidence.reduced_dimensions.12501 );2502 println!("resolution_stall_sequence={}", evidence.stall_sequence);2503 println!("resolution_stall_fresh_reports={RESOLUTION_STALL_FRESH_REPORTS}");2504 println!(2505 "resolution_stall_fresh_duration_ms={}",2506 evidence.fresh_duration.as_millis()2507 );2508 println!(2509 "resolution_stall_input_ack_micros={}",2510 evidence.stall_input_ack_micros2511 );2512 match evidence.recovery.as_ref() {2513 Some(recovery) => {2514 println!("resolution_recovery_windows={}", recovery.windows);2515 println!("resolution_recovery_trough_kbps={}", recovery.trough_kbps);2516 println!("resolution_recovery_final_kbps={}", recovery.final_kbps);2517 println!(2518 "resolution_recovery_increase_observed={}",2519 recovery.increase_observed2520 );2521 println!(2522 "resolution_recovery_final_dimensions={}x{}",2523 recovery.final_dimensions.0, recovery.final_dimensions.12524 );2525 println!(2526 "resolution_recovery_restored_native={}",2527 recovery.restored_native2528 );2529 }2530 None => {2531 println!("resolution_recovery_windows=not-requested");2532 println!("resolution_recovery_trough_kbps=not-requested");2533 println!("resolution_recovery_final_kbps=not-requested");2534 println!("resolution_recovery_increase_observed=not-requested");2535 println!("resolution_recovery_final_dimensions=not-requested");2536 println!("resolution_recovery_restored_native=not-requested");2537 }2538 }2539 println!(2540 "resolution_resume_input_ack_micros={}",2541 evidence.resume_input_ack_micros2542 );2543 println!(2544 "resolution_resume_media_micros={}",2545 evidence.resume_media_micros2546 );2547 println!(2548 "resolution_resume_sequence_advance={}",2549 evidence.resume_sequence_advance2550 );2551 return Ok(());2552 }25532554 let started = Instant::now();2555 let mut received = 0_u32;2556 let mut bytes = 0_u64;2557 let mut keyframes = 0_u32;2558 let mut gaps = 0_u64;2559 let mut media_objects_dropped = 0_u64;2560 let mut media_objects_late = 0_u64;2561 let mut last_sequence: Option<u64> = None;2562 let mut dimensions = None;2563 let mut first_accepted_at = None;2564 let mut last_accepted_at = None;2565 let mut object_receiver_v3 = (media_transport == MediaTransport::GroupedV3).then(|| {2566 MediaObjectReceiverV3::new(2567 media_connection.clone(),2568 Duration::from_secs(args.timeout_seconds),2569 )2570 });2571 let mut object_sequence_v3 = MediaObjectSequenceV3::new();2572 let mut keyframe_request_sent = false;2573 let mut keyframe_recovery_verified = false;2574 let mut keyframe_request_sent_at: Option<Instant> = None;2575 let mut keyframe_recovery_micros = None;2576 let mut keyframe_request_group_id = None;2577 let mut keyframe_request_last_sequence = None;2578 let mut slow_consumer_stalled = false;2579 let mut slow_consumer_pending_recovery = false;2580 let mut slow_consumer_verified = false;2581 let mut slow_consumer_group_id = None;2582 let mut slow_consumer_last_sequence = None;2583 let mut slow_consumer_capture_timestamp_us = None;2584 let mut slow_consumer_cancelled_groups_before = None;2585 let mut slow_consumer_resumed_at: Option<Instant> = None;2586 let mut slow_consumer_recovery_micros = None;2587 let mut slow_consumer_cancellation_delta = None;2588 let mut slow_consumer_group_advance = None;2589 let mut slow_consumer_sequence_advance = None;2590 let mut slow_consumer_capture_advance_micros = None;2591 let mut slow_consumer_input_ack_micros = None;25922593 while received < args.frames {2594 let (frame, recovery_frame): (AcceptedMedia, bool) = match media_transport {2595 MediaTransport::UpstreamMoq => loop {2596 let media_wait = if slow_consumer_pending_recovery {2597 SLOW_CONSUMER_RECOVERY_TIMEOUT2598 } else {2599 Duration::from_secs(args.timeout_seconds).saturating_add(Duration::from_secs(1))2600 };2601 let outcome = tokio::time::timeout(2602 media_wait,2603 moq_receiver2604 .as_mut()2605 .expect("upstream MoQ receiver is present")2606 .next(),2607 )2608 .await2609 .context("timed out waiting for an upstream MoQ media object")??2610 .context("host closed the upstream MoQ video track")?;2611 match outcome {2612 MoqProbeOutcome::Dropped => {2613 media_objects_dropped = media_objects_dropped.saturating_add(1);2614 }2615 MoqProbeOutcome::Frame {2616 frame,2617 group_sequence,2618 recovery,2619 } => {2620 let requested_keyframe_recovery =2621 keyframe_request_sent && !keyframe_recovery_verified;2622 if requested_keyframe_recovery {2623 ensure!(2624 recovery2625 && frame.header.flags.contains(FrameFlags::KEYFRAME)2626 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),2627 "MoQ keyframe request recovery did not begin at a configured IDR barrier"2628 );2629 ensure!(2630 keyframe_request_group_id2631 .is_some_and(|prior| group_sequence > prior),2632 "MoQ keyframe request recovery did not advance the native group"2633 );2634 ensure!(2635 keyframe_request_last_sequence2636 .is_some_and(|prior| frame.header.sequence > prior),2637 "MoQ keyframe request recovery did not advance media sequence"2638 );2639 keyframe_recovery_micros = keyframe_request_sent_at.map(|sent_at| {2640 u64::try_from(sent_at.elapsed().as_micros()).unwrap_or(u64::MAX)2641 });2642 keyframe_recovery_verified = true;2643 }2644 let slow_consumer_recovery = slow_consumer_pending_recovery;2645 if slow_consumer_recovery {2646 ensure!(2647 recovery2648 && frame.header.flags.contains(FrameFlags::KEYFRAME)2649 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),2650 "MoQ slow-consumer recovery did not begin at a configured IDR barrier"2651 );2652 let prior_group = slow_consumer_group_id2653 .context("slow-consumer group was not recorded")?;2654 let group_advance = group_sequence.checked_sub(prior_group).context(2655 "MoQ slow-consumer recovery did not advance the native group",2656 )?;2657 ensure!(2658 group_advance > 0,2659 "MoQ slow-consumer recovery did not advance the native group"2660 );2661 let prior_sequence = slow_consumer_last_sequence2662 .context("slow-consumer sequence was not recorded")?;2663 let sequence_advance =2664 frame.header.sequence.checked_sub(prior_sequence).context(2665 "MoQ slow-consumer recovery did not advance media sequence",2666 )?;2667 ensure!(2668 sequence_advance > 0,2669 "MoQ slow-consumer recovery did not advance media sequence"2670 );2671 let prior_capture_timestamp = slow_consumer_capture_timestamp_us2672 .context("slow-consumer capture timestamp was not recorded")?;2673 let capture_advance = frame2674 .header2675 .capture_timestamp_us2676 .checked_sub(prior_capture_timestamp)2677 .context(2678 "MoQ slow-consumer recovery capture timestamp moved backward",2679 )?;2680 let minimum_capture_advance = args2681 .slow_consumer_ms2682 .expect("slow-consumer duration is configured")2683 .saturating_mul(1_000)2684 / 2;2685 ensure!(2686 capture_advance >= minimum_capture_advance,2687 "MoQ slow-consumer recovery advanced capture time by only {capture_advance}us; required at least {minimum_capture_advance}us"2688 );2689 let cancelled_before = slow_consumer_cancelled_groups_before2690 .context("slow-consumer cancellation count was not recorded")?;2691 let cancelled_after = moq_receiver2692 .as_ref()2693 .expect("upstream MoQ receiver is present")2694 .cancelled_groups;2695 let cancellation_delta = cancelled_after2696 .checked_sub(cancelled_before)2697 .context("MoQ slow-consumer cancellation count moved backward")?;2698 ensure!(2699 cancellation_delta > 0,2700 "MoQ slow-consumer recovery did not observe a cancelled stale group"2701 );2702 ensure!(2703 moq_receiver2704 .as_ref()2705 .expect("upstream MoQ receiver is present")2706 .historical_suffix_frames2707 == 0,2708 "MoQ slow-consumer recovery accepted historical delta suffix frames"2709 );2710 let recovery_micros = slow_consumer_resumed_at2711 .context("slow-consumer resume instant was not recorded")?2712 .elapsed()2713 .as_micros();2714 ensure!(2715 recovery_micros <= SLOW_CONSUMER_RECOVERY_TIMEOUT.as_micros(),2716 "MoQ slow-consumer recovery exceeded {}us",2717 SLOW_CONSUMER_RECOVERY_TIMEOUT.as_micros()2718 );2719 slow_consumer_recovery_micros =2720 Some(u64::try_from(recovery_micros).unwrap_or(u64::MAX));2721 slow_consumer_cancellation_delta = Some(cancellation_delta);2722 slow_consumer_group_advance = Some(group_advance);2723 slow_consumer_sequence_advance = Some(sequence_advance);2724 slow_consumer_capture_advance_micros = Some(capture_advance);2725 slow_consumer_pending_recovery = false;2726 slow_consumer_verified = true;2727 }2728 break (2729 AcceptedMedia {2730 flags: frame.header.flags,2731 width: frame.header.width,2732 height: frame.header.height,2733 sequence: frame.header.sequence,2734 capture_timestamp_us: frame.header.capture_timestamp_us,2735 payload_len: frame.payload.len(),2736 v3_group_id: Some(group_sequence),2737 },2738 recovery,2739 );2740 }2741 }2742 },2743 MediaTransport::GroupedV3 => loop {2744 let outcome = tokio::time::timeout(2745 Duration::from_secs(args.timeout_seconds)2746 .saturating_add(Duration::from_secs(1)),2747 object_receiver_v32748 .as_mut()2749 .expect("v3 media object receiver is present")2750 .next(),2751 )2752 .await2753 .context("timed out waiting for media v3 object")??2754 .context("host closed the media v3 object connection")?;2755 match outcome {2756 MediaObjectOutcomeV3::Dropped { .. } => {2757 if object_sequence_v3.note_drop() {2758 media_objects_dropped = media_objects_dropped.saturating_add(1);2759 } else {2760 media_objects_late = media_objects_late.saturating_add(1);2761 }2762 }2763 MediaObjectOutcomeV3::Malformed {2764 accept_index,2765 error,2766 } => {2767 bail!("media v3 object {accept_index} is malformed: {error}");2768 }2769 MediaObjectOutcomeV3::Object { object, .. } => {2770 match object_sequence_v3.classify(&object) {2771 MediaObjectDecisionV3::Deliver { discontinuity } => {2772 let recovery = keyframe_request_sent && !keyframe_recovery_verified;2773 if recovery {2774 ensure!(2775 discontinuity2776 && object.header.object_id == 02777 && object.header.flags.contains(FrameFlags::KEYFRAME)2778 && object2779 .header2780 .flags2781 .contains(FrameFlags::CODEC_CONFIG)2782 && object2783 .header2784 .flags2785 .contains(FrameFlags::DISCONTINUITY),2786 "keyframe request recovery did not begin at a configured discontinuity object zero"2787 );2788 ensure!(2789 keyframe_request_group_id2790 .is_some_and(2791 |group_id| object.header.group_id > group_id2792 ),2793 "keyframe request recovery did not advance the media group"2794 );2795 ensure!(2796 keyframe_request_last_sequence.is_some_and(|sequence| {2797 object.header.sequence > sequence2798 }),2799 "keyframe request recovery did not advance media sequence"2800 );2801 keyframe_recovery_micros =2802 keyframe_request_sent_at.map(|sent_at| {2803 u64::try_from(sent_at.elapsed().as_micros())2804 .unwrap_or(u64::MAX)2805 });2806 keyframe_recovery_verified = true;2807 }2808 break (object.into(), recovery);2809 }2810 MediaObjectDecisionV3::DropLate => {2811 media_objects_late = media_objects_late.saturating_add(1);2812 }2813 MediaObjectDecisionV3::DropUntilKeyframe => {2814 media_objects_dropped = media_objects_dropped.saturating_add(1);2815 }2816 }2817 }2818 }2819 },2820 };28212822 let accepted_at = Instant::now();2823 first_accepted_at.get_or_insert(accepted_at);2824 last_accepted_at = Some(accepted_at);28252826 if received == 0 {2827 ensure!(2828 frame.flags.contains(FrameFlags::KEYFRAME)2829 && frame.flags.contains(FrameFlags::CODEC_CONFIG),2830 "first media frame is not a decodable keyframe with codec configuration"2831 );2832 }28332834 let frame_dimensions = (frame.width, frame.height);2835 if let Some(expected) = dimensions {2836 ensure!(2837 frame_dimensions == expected,2838 "media dimensions changed during probe"2839 );2840 } else {2841 dimensions = Some(frame_dimensions);2842 }2843 if let Some(previous) = last_sequence2844 && !recovery_frame2845 {2846 gaps += sequence_gap(previous, frame.sequence)?;2847 }2848 last_sequence = Some(frame.sequence);2849 keyframes += u32::from(frame.flags.contains(FrameFlags::KEYFRAME));2850 bytes = bytes.saturating_add(frame.payload_len as u64);2851 received += 1;28522853 if let Some(stall_ms) = args.slow_consumer_ms2854 && !slow_consumer_stalled2855 {2856 slow_consumer_group_id = frame.v3_group_id;2857 slow_consumer_last_sequence = Some(frame.sequence);2858 slow_consumer_capture_timestamp_us = Some(frame.capture_timestamp_us);2859 slow_consumer_cancelled_groups_before = Some(2860 moq_receiver2861 .as_ref()2862 .expect("upstream MoQ receiver is present")2863 .cancelled_groups,2864 );2865 let first_half = Duration::from_millis(stall_ms / 2);2866 let second_half = Duration::from_millis(stall_ms).saturating_sub(first_half);2867 tokio::time::sleep(first_half).await;2868 let input_probe_started = Instant::now();2869 write_input_event(&mut input_send, &InputEvent::Probe)2870 .await2871 .context("writing input probe during slow-consumer media stall")?;2872 expected_ack = expected_ack.saturating_add(1);2873 tokio::time::timeout(2874 SLOW_CONSUMER_RECOVERY_TIMEOUT,2875 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack),2876 )2877 .await2878 .context("input acknowledgement was blocked by the slow media consumer")??;2879 slow_consumer_input_ack_micros =2880 Some(u64::try_from(input_probe_started.elapsed().as_micros()).unwrap_or(u64::MAX));2881 tokio::time::sleep(second_half).await;2882 slow_consumer_resumed_at = Some(Instant::now());2883 slow_consumer_pending_recovery = true;2884 slow_consumer_stalled = true;2885 }28862887 if args.keyframe_smoke2888 && !keyframe_request_sent2889 && received == KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES2890 {2891 let request = MediaControlRequestV3::request_keyframe(2892 args.keyframe_request_id,2893 Some(frame.sequence),2894 KeyframeRequestReasonV3::DecoderReset,2895 );2896 tokio::time::timeout(2897 Duration::from_secs(1),2898 write_media_control_request_v3(&mut media_send, &request),2899 )2900 .await2901 .context("timed out writing v3 keyframe request")??;2902 keyframe_request_group_id = frame.v3_group_id;2903 keyframe_request_last_sequence = Some(frame.sequence);2904 keyframe_request_sent_at = Some(Instant::now());2905 keyframe_request_sent = true;2906 match media_transport {2907 MediaTransport::UpstreamMoq => {2908 let cancelled_group = moq_receiver2909 .as_mut()2910 .expect("upstream MoQ receiver is present")2911 .request_recovery();2912 ensure!(2913 cancelled_group == keyframe_request_group_id,2914 "MoQ recovery did not cancel the currently accepted native group"2915 );2916 }2917 MediaTransport::GroupedV3 => object_sequence_v3.request_recovery(),2918 }2919 }2920 }29212922 let Some((width, height)) = dimensions else {2923 bail!("probe received no frames");2924 };2925 if let Some((expected_width, expected_height)) = args.expect_size {2926 ensure!(2927 (width, height) == (expected_width, expected_height),2928 "expected {expected_width}x{expected_height} but received {width}x{height}"2929 );2930 }2931 ensure!(keyframes > 0, "probe received no H.264 keyframe");2932 ensure!(gaps == 0, "probe observed {gaps} media sequence gaps");2933 ensure!(2934 !args.keyframe_smoke || keyframe_recovery_verified,2935 "probe did not verify requested keyframe recovery"2936 );2937 ensure!(2938 args.slow_consumer_ms.is_none() || slow_consumer_verified,2939 "probe did not verify bounded slow-consumer recovery"2940 );2941 let accepted_span = first_accepted_at2942 .zip(last_accepted_at)2943 .map_or(Duration::ZERO, |(first, last)| {2944 last.saturating_duration_since(first)2945 });2946 let accepted_fps = accepted_frame_rate(received, accepted_span);2947 if let Some(minimum_fps) = args.minimum_fps {2948 let accepted_fps = accepted_fps.context(2949 "--minimum-fps requires at least two accepted frames with measurable elapsed time",2950 )?;2951 ensure!(2952 accepted_fps >= minimum_fps,2953 "probe sustained only {accepted_fps:.3} accepted fps; required at least {minimum_fps:.3} fps"2954 );2955 }2956 let feedback_evidence = if args.feedback_smoke {2957 Some(2958 run_feedback_smoke(2959 &endpoint,2960 address.clone(),2961 nonce,2962 session_id,2963 last_sequence.context("feedback smoke requires an accepted media sequence")?,2964 )2965 .await?,2966 )2967 } else {2968 None2969 };2970 let media_diagnostics_connection = moq_receiver2971 .as_ref()2972 .map_or(&media_connection, MoqProbeReceiver::connection);2973 let (media_path_mode, media_path_rtt_ms) =2974 selected_path_diagnostics(media_diagnostics_connection);2975 let (control_path_mode, control_path_rtt_ms) = if media_transport == MediaTransport::UpstreamMoq2976 {2977 selected_path_diagnostics(&media_connection)2978 } else {2979 ("not-used", None)2980 };2981 let (input_path_mode, input_path_rtt_ms) = selected_path_diagnostics(&input_connection);2982 let feedback_path_mode = feedback_evidence2983 .as_ref()2984 .map_or("not-used", |(_, path_mode, _)| *path_mode);2985 if args.relay_only {2986 ensure!(2987 control_path_mode == "relay"2988 && media_path_mode == "relay"2989 && input_path_mode == "relay"2990 && (feedback_evidence.is_none() || feedback_path_mode == "relay"),2991 "relay-only probe did not keep every transport on relay: control={control_path_mode}, media={media_path_mode}, input={input_path_mode}, feedback={feedback_path_mode}"2992 );2993 }2994 let (2995 moq_cancelled_groups,2996 moq_group_gaps,2997 moq_maximum_group_objects,2998 moq_maximum_group_bytes,2999 moq_historical_suffix_frames,3000 ) = moq_receiver.as_ref().map_or((0, 0, 0, 0, 0), |receiver| {3001 (3002 receiver.cancelled_groups,3003 receiver.group_gaps,3004 receiver.maximum_group_objects,3005 receiver.maximum_group_bytes,3006 receiver.historical_suffix_frames,3007 )3008 });3009 input_send.finish().context("finishing input stream")?;3010 media_send3011 .finish()3012 .context("finishing media request stream")?;3013 // Close both protocol connections explicitly. Relying only on endpoint3014 // teardown can leave a peer waiting for QUIC idle timeout during repeated3015 // short-lived probes, which obscures whether the host released its3016 // one-client lease deterministically.3017 input_connection.close(0_u32.into(), b"probe complete");3018 if let Some(receiver) = moq_receiver.as_ref() {3019 receiver.connection().close(0_u32.into(), b"probe complete");3020 }3021 media_connection.close(0_u32.into(), b"probe complete");3022 tokio::time::sleep(Duration::from_millis(50)).await;3023 endpoint.close().await;30243025 println!("probe=ok");3026 println!("session_id={session_id}");3027 println!("frames={received}");3028 println!("dimensions={width}x{height}");3029 println!("keyframes={keyframes}");3030 println!("sequence_gaps={gaps}");3031 println!("media_objects_dropped={media_objects_dropped}");3032 println!("media_objects_late={media_objects_late}");3033 println!("transport={}", media_transport.label());3034 println!(3035 "control_alpn={}",3036 if media_transport == MediaTransport::UpstreamMoq {3037 "sigil/control/1"3038 } else {3039 "not-used"3040 }3041 );3042 println!("transport_alpn={}", media_transport.media_alpn_label());3043 println!("first_configured_idr=ok");3044 println!("frame_sequence=monotonic");3045 println!(3046 "group_sequence={}",3047 if media_transport == MediaTransport::UpstreamMoq {3048 "monotonic"3049 } else {3050 "not-applicable"3051 }3052 );3053 if media_transport == MediaTransport::UpstreamMoq {3054 println!(3055 "moq_catalog={}",3056 moq_receiver3057 .as_ref()3058 .expect("upstream MoQ receiver is present")3059 .catalog_mode3060 .label()3061 );3062 println!("moq_group_capacity=1");3063 println!("moq_cancelled_groups={moq_cancelled_groups}");3064 println!("moq_group_gaps={moq_group_gaps}");3065 println!("moq_unrecovered_group_gaps=0");3066 println!("moq_maximum_group_objects={moq_maximum_group_objects}");3067 println!("moq_maximum_group_bytes={moq_maximum_group_bytes}");3068 println!("moq_historical_suffix_frames={moq_historical_suffix_frames}");3069 } else {3070 println!("moq_catalog=not-applicable");3071 for field in [3072 "moq_group_capacity",3073 "moq_cancelled_groups",3074 "moq_group_gaps",3075 "moq_unrecovered_group_gaps",3076 "moq_maximum_group_objects",3077 "moq_maximum_group_bytes",3078 "moq_historical_suffix_frames",3079 ] {3080 println!("{field}=not-applicable");3081 }3082 }3083 println!(3084 "recovery_barrier={}",3085 if args.keyframe_smoke {3086 "configured-idr"3087 } else {3088 "not-requested"3089 }3090 );3091 println!(3092 "keyframe_recovery={}",3093 if args.keyframe_smoke {3094 "ok"3095 } else {3096 "not-requested"3097 }3098 );3099 if args.keyframe_smoke {3100 println!("keyframe_request_id={}", args.keyframe_request_id);3101 println!(3102 "keyframe_recovery_micros={}",3103 keyframe_recovery_micros.context("keyframe recovery latency was not measured")?3104 );3105 } else {3106 println!("keyframe_request_id=not-requested");3107 println!("keyframe_recovery_micros=not-requested");3108 }3109 if let Some(stall_ms) = args.slow_consumer_ms {3110 println!("slow_consumer=ok");3111 println!("slow_consumer_stall_ms={stall_ms}");3112 println!("slow_consumer_first_post_stall=configured-idr");3113 println!("slow_consumer_historical_suffix_frames=0");3114 println!(3115 "slow_consumer_recovery_micros={}",3116 slow_consumer_recovery_micros3117 .context("slow-consumer recovery latency was not measured")?3118 );3119 println!(3120 "slow_consumer_cancellation_delta={}",3121 slow_consumer_cancellation_delta3122 .context("slow-consumer cancellation delta was not measured")?3123 );3124 println!(3125 "slow_consumer_group_advance={}",3126 slow_consumer_group_advance.context("slow-consumer group advance was not measured")?3127 );3128 println!(3129 "slow_consumer_sequence_advance={}",3130 slow_consumer_sequence_advance3131 .context("slow-consumer sequence advance was not measured")?3132 );3133 println!(3134 "slow_consumer_capture_advance_micros={}",3135 slow_consumer_capture_advance_micros3136 .context("slow-consumer capture advance was not measured")?3137 );3138 println!(3139 "slow_consumer_input_ack_micros={}",3140 slow_consumer_input_ack_micros3141 .context("slow-consumer input acknowledgement was not measured")?3142 );3143 } else {3144 for field in [3145 "slow_consumer",3146 "slow_consumer_stall_ms",3147 "slow_consumer_first_post_stall",3148 "slow_consumer_historical_suffix_frames",3149 "slow_consumer_recovery_micros",3150 "slow_consumer_cancellation_delta",3151 "slow_consumer_group_advance",3152 "slow_consumer_sequence_advance",3153 "slow_consumer_capture_advance_micros",3154 "slow_consumer_input_ack_micros",3155 ] {3156 println!("{field}=not-requested");3157 }3158 }3159 match accepted_fps {3160 Some(fps) => println!("accepted_fps={fps:.3}"),3161 None => println!("accepted_fps=unknown"),3162 }3163 println!("input_ack_micros={input_ack_micros}");3164 println!(3165 "pointer_smoke={}",3166 if args.pointer_smoke {3167 "ok"3168 } else {3169 "not-requested"3170 }3171 );3172 println!(3173 "pointer_feedback_smoke={}",3174 if args.pointer_feedback_smoke {3175 "ok"3176 } else {3177 "not-requested"3178 }3179 );3180 match pointer_sync_position {3181 Some(position) => println!("pointer_sync_position={},{}", position.x, position.y),3182 None => println!("pointer_sync_position=not-requested"),3183 }3184 match pointer_motion_position {3185 Some(position) => println!("pointer_motion_position={},{}", position.x, position.y),3186 None => println!("pointer_motion_position=not-requested"),3187 }3188 println!(3189 "gamepad_smoke={}",3190 if args.gamepad_smoke {3191 "ok"3192 } else {3193 "not-requested"3194 }3195 );3196 if let Some((decision, path_mode, path_rtt_ms)) = &feedback_evidence {3197 println!("feedback_smoke=ok");3198 println!("feedback_report_id={}", decision.report_id);3199 println!("feedback_state={:?}", decision.state);3200 println!("feedback_target_kbps={}", decision.target_kbps);3201 println!("feedback_floor_kbps={}", decision.floor_kbps);3202 println!("feedback_ceiling_kbps={}", decision.ceiling_kbps);3203 println!("feedback_applied={}", decision.applied);3204 println!("feedback_path_mode={path_mode}");3205 match path_rtt_ms {3206 Some(rtt) => println!("feedback_path_rtt_ms={rtt:.3}"),3207 None => println!("feedback_path_rtt_ms=unknown"),3208 }3209 } else {3210 for field in [3211 "feedback_smoke",3212 "feedback_report_id",3213 "feedback_state",3214 "feedback_target_kbps",3215 "feedback_floor_kbps",3216 "feedback_ceiling_kbps",3217 "feedback_applied",3218 "feedback_path_mode",3219 "feedback_path_rtt_ms",3220 ] {3221 println!("{field}=not-requested");3222 }3223 }3224 println!("path_mode={media_path_mode}");3225 match media_path_rtt_ms {3226 Some(rtt) => println!("path_rtt_ms={rtt:.3}"),3227 None => println!("path_rtt_ms=unknown"),3228 }3229 println!("control_path_mode={control_path_mode}");3230 match control_path_rtt_ms {3231 Some(rtt) => println!("control_path_rtt_ms={rtt:.3}"),3232 None => println!("control_path_rtt_ms=unknown"),3233 }3234 println!("media_path_mode={media_path_mode}");3235 match media_path_rtt_ms {3236 Some(rtt) => println!("media_path_rtt_ms={rtt:.3}"),3237 None => println!("media_path_rtt_ms=unknown"),3238 }3239 println!("input_path_mode={input_path_mode}");3240 match input_path_rtt_ms {3241 Some(rtt) => println!("input_path_rtt_ms={rtt:.3}"),3242 None => println!("input_path_rtt_ms=unknown"),3243 }3244 println!(3245 "forced_relay={}",3246 if args.relay_only {3247 "ok"3248 } else {3249 "not-requested"3250 }3251 );3252 println!("encoded_bytes={bytes}");3253 println!("elapsed_ms={}", started.elapsed().as_millis());3254 Ok(())3255}32563257fn pointer_smoke_events(3258 pointer_surface_dimensions: Option<PointerSurfaceDimensions>,3259) -> Result<[InputEvent; 3]> {3260 let dimensions = pointer_surface_dimensions3261 .context("host did not advertise pointer surface dimensions required by --pointer-smoke")?;3262 let [sync_position, _motion_position] = pointer_smoke_expected_positions(dimensions)?;3263 Ok([3264 InputEvent::MousePositionSync {3265 x: sync_position.x,3266 y: sync_position.y,3267 },3268 InputEvent::MouseMoveRelative { dx: 32, dy: 16 },3269 InputEvent::MouseClick { b: 1 },3270 ])3271}32723273fn pointer_smoke_expected_positions(3274 dimensions: PointerSurfaceDimensions,3275) -> Result<[PointerPosition; 2]> {3276 let sync_position = PointerPosition::new(3277 i32::from(dimensions.width / 2),3278 i32::from(dimensions.height / 2),3279 )?;3280 let motion_position = PointerPosition::new(3281 sync_position3282 .x3283 .checked_add(32)3284 .context("pointer smoke X position overflowed")?,3285 sync_position3286 .y3287 .checked_add(16)3288 .context("pointer smoke Y position overflowed")?,3289 )?;3290 Ok([sync_position, motion_position])3291}32923293fn sequence_gap(previous: u64, current: u64) -> Result<u64> {3294 let Some(expected) = previous.checked_add(1) else {3295 bail!("media sequence overflowed after {previous}");3296 };3297 ensure!(3298 current >= expected,3299 "non-monotonic media sequence: previous={previous}, current={current}"3300 );3301 Ok(current - expected)3302}33033304fn accepted_frame_rate(frame_count: u32, elapsed: Duration) -> Option<f64> {3305 if frame_count < 2 || elapsed.is_zero() {3306 return None;3307 }3308 Some(f64::from(frame_count - 1) / elapsed.as_secs_f64())3309}33103311fn selected_path_diagnostics(3312 connection: &iroh::endpoint::Connection,3313) -> (&'static str, Option<f64>) {3314 let paths = connection.paths();3315 let Some(path) = paths.iter().find(|path| path.is_selected()) else {3316 return ("unknown", None);3317 };3318 let mode = if path.is_ip() {3319 "direct"3320 } else if path.is_relay() {3321 "relay"3322 } else {3323 "custom"3324 };3325 (mode, Some(path.rtt().as_secs_f64() * 1000.0))3326}33273328fn parse_size(value: &str) -> std::result::Result<(u16, u16), String> {3329 let (width, height) = value3330 .split_once('x')3331 .ok_or_else(|| "size must be WIDTHxHEIGHT".to_owned())?;3332 let width = width.parse().map_err(|_| "invalid width".to_owned())?;3333 let height = height.parse().map_err(|_| "invalid height".to_owned())?;3334 if width == 0 || height == 0 {3335 return Err("dimensions must be non-zero".to_owned());3336 }3337 Ok((width, height))3338}33393340fn parse_minimum_fps(value: &str) -> std::result::Result<f64, String> {3341 let minimum_fps = value3342 .parse::<f64>()3343 .map_err(|_| "minimum fps must be a number".to_owned())?;3344 if !minimum_fps.is_finite() || minimum_fps <= 0.0 || minimum_fps > 240.0 {3345 return Err("minimum fps must be finite and within (0, 240]".to_owned());3346 }3347 Ok(minimum_fps)3348}33493350fn load_invitation_file(3351 path: &Path,3352 expected_host: EndpointId,3353 expected_peer: EndpointId,3354 required_grants: InvitationGrants,3355) -> Result<String> {3356 let now = SystemTime::now()3357 .duration_since(UNIX_EPOCH)3358 .context("system clock is before the Unix epoch")?3359 .as_secs();3360 load_invitation_file_at(path, expected_host, expected_peer, required_grants, now)3361}33623363fn load_invitation_file_at(3364 path: &Path,3365 expected_host: EndpointId,3366 expected_peer: EndpointId,3367 required_grants: InvitationGrants,3368 now: u64,3369) -> Result<String> {3370 let mut options = OpenOptions::new();3371 options.read(true);3372 3373 options.custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);3374 let file = options.open(path).with_context(|| {3375 format!(3376 "opening invitation {} without following symlinks",3377 path.display()3378 )3379 })?;3380 let metadata = file3381 .metadata()3382 .with_context(|| format!("inspecting invitation {}", path.display()))?;3383 ensure!(metadata.is_file(), "invitation must be a regular file");3384 ensure!(3385 metadata.len() <= MAX_INVITATION_FILE_BYTES,3386 "invitation {} exceeds the bounded file size",3387 path.display()3388 );3389 3390 {3391 ensure!(3392 metadata.mode() & 0o077 == 0,3393 "invitation {} must not be readable or writable by group or other users",3394 path.display()3395 );3396 ensure!(3397 metadata.uid() == unsafe { libc::geteuid() },3398 "invitation {} is not owned by the current user",3399 path.display()3400 );3401 }3402 let mut bytes = Vec::with_capacity(metadata.len() as usize);3403 file.take(MAX_INVITATION_FILE_BYTES + 1)3404 .read_to_end(&mut bytes)3405 .with_context(|| format!("reading invitation {}", path.display()))?;3406 ensure!(3407 bytes.len() as u64 <= MAX_INVITATION_FILE_BYTES,3408 "invitation {} grew beyond the bounded file size while being read",3409 path.display()3410 );3411 let token_bytes = bytes3412 .strip_suffix(b"\r\n")3413 .or_else(|| bytes.strip_suffix(b"\n"))3414 .unwrap_or(&bytes);3415 ensure!(3416 !token_bytes.is_empty() && token_bytes.len() <= MAX_INVITATION_TOKEN_LEN,3417 "invitation token has an invalid length"3418 );3419 ensure!(3420 !token_bytes.contains(&b'\n') && !token_bytes.contains(&b'\r'),3421 "invitation file contains unexpected line breaks"3422 );3423 let token = std::str::from_utf8(token_bytes).context("invitation file is not UTF-8")?;3424 let invitation =3425 SignedInvitation::decode(token).context("decoding and verifying invitation file")?;3426 ensure!(3427 invitation.claims.host_node_id == *expected_host.as_bytes(),3428 "invitation belongs to a different Sigil host"3429 );3430 ensure!(3431 invitation.claims.intended_peer_id == *expected_peer.as_bytes(),3432 "invitation is bound to a different probe identity"3433 );3434 ensure!(3435 invitation.claims.grants.contains(required_grants),3436 "invitation does not grant every requested probe capability"3437 );3438 ensure!(3439 invitation3440 .claims3441 .grants3442 .contains(InvitationGrants::POINTER_KEYBOARD)3443 || invitation.claims.grants.contains(InvitationGrants::GAMEPAD),3444 "invitation must grant pointer/keyboard or gamepad for the probe input acknowledgment"3445 );3446 ensure!(3447 now >= invitation3448 .claims3449 .issued_at_unix3450 .saturating_sub(INVITATION_CLOCK_SKEW_SECS),3451 "invitation was issued too far in the future"3452 );3453 ensure!(3454 now <= invitation.claims.expires_at_unix,3455 "invitation has expired"3456 );3457 Ok(token.to_owned())3458}34593460fn probe_client_hello(3461 nonce: [u8; 16],3462 capabilities: Vec<Capability>,3463 invitation: Option<&str>,3464) -> ClientHello {3465 let hello = ClientHello::new("sigil-probe/0.1.0", nonce, capabilities);3466 match invitation {3467 Some(invitation) => hello.with_invitation(invitation),3468 None => hello,3469 }3470}34713472async fn negotiate(3473 send: &mut iroh::endpoint::SendStream,3474 recv: &mut iroh::endpoint::RecvStream,3475 nonce: [u8; 16],3476 capabilities: Vec<Capability>,3477 required: Capability,3478 name: &str,3479 invitation: Option<&str>,3480) -> Result<Negotiated> {3481 write_client_hello(send, &probe_client_hello(nonce, capabilities, invitation))3482 .await3483 .with_context(|| format!("writing {name} hello"))?;3484 let response = tokio::time::timeout(Duration::from_secs(10), read_host_hello(recv))3485 .await3486 .with_context(|| format!("timed out waiting for {name} hello"))??3487 .with_context(|| format!("host closed during {name} hello"))?;3488 if !response.accepted {3489 bail!(3490 "host rejected {name} stream: {}",3491 response.message.as_deref().unwrap_or("unspecified reason")3492 );3493 }3494 ensure!(3495 response.capabilities.contains(&required),3496 "host accepted {name} without required capability {required:?}"3497 );3498 let session_id = response3499 .session_id3500 .with_context(|| format!("host omitted {name} session ID"))?;3501 Ok(Negotiated {3502 session_id,3503 capabilities: response.capabilities,3504 pointer_surface_dimensions: response.pointer_surface_dimensions,3505 })3506}35073508struct Negotiated {3509 session_id: u64,3510 capabilities: Vec<Capability>,3511 pointer_surface_dimensions: Option<PointerSurfaceDimensions>,3512}35133514async fn read_expected_input_ack<R>(3515 recv: &mut R,3516 timeout_seconds: u64,3517 expected_sequence: u64,3518) -> Result<InputAck>3519where3520 R: tokio::io::AsyncRead + Unpin,3521{3522 let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_seconds.min(5));3523 read_expected_input_ack_before(recv, deadline, expected_sequence).await3524}35253526async fn read_expected_input_ack_before<R>(3527 recv: &mut R,3528 deadline: tokio::time::Instant,3529 expected_sequence: u64,3530) -> Result<InputAck>3531where3532 R: tokio::io::AsyncRead + Unpin,3533{3534 loop {3535 let input_ack = tokio::time::timeout_at(deadline, read_input_ack(recv))3536 .await3537 .context("timed out waiting for input acknowledgment")??3538 .context("host closed before input acknowledgment")?;3539 ensure!(3540 input_ack.sequence <= expected_sequence,3541 "unexpected input acknowledgment sequence {}; expected at most {expected_sequence}",3542 input_ack.sequence3543 );3544 if input_ack.sequence == expected_sequence {3545 return Ok(input_ack);3546 }3547 }3548}35493550async fn confirm_pointer_position<S, R>(3551 send: &mut S,3552 recv: &mut R,3553 expected_sequence: &mut u64,3554 mut input_ack: InputAck,3555 expected_position: PointerPosition,3556 timeout: Duration,3557) -> Result<PointerPosition>3558where3559 S: tokio::io::AsyncWrite + Unpin,3560 R: tokio::io::AsyncRead + Unpin,3561{3562 ensure!(3563 !timeout.is_zero(),3564 "pointer feedback confirmation timeout must be nonzero"3565 );3566 let deadline = tokio::time::Instant::now() + timeout.min(Duration::from_secs(5));3567 loop {3568 if input_ack.pointer_position == Some(expected_position) {3569 return Ok(expected_position);3570 }3571 let now = tokio::time::Instant::now();3572 if now >= deadline {3573 bail!(3574 "compositor pointer position did not converge to {},{}; last observed position was {:?}",3575 expected_position.x,3576 expected_position.y,3577 input_ack.pointer_position3578 );3579 }3580 tokio::time::sleep_until(std::cmp::min(now + Duration::from_millis(16), deadline)).await;3581 if tokio::time::Instant::now() >= deadline {3582 bail!(3583 "compositor pointer position did not converge to {},{}; last observed position was {:?}",3584 expected_position.x,3585 expected_position.y,3586 input_ack.pointer_position3587 );3588 }3589 write_input_event(send, &InputEvent::Probe)3590 .await3591 .context("writing pointer feedback convergence probe")?;3592 *expected_sequence = expected_sequence3593 .checked_add(1)3594 .context("input acknowledgment sequence overflowed")?;3595 input_ack = match read_expected_input_ack_before(recv, deadline, *expected_sequence).await {3596 Ok(input_ack) => input_ack,3597 Err(error) if tokio::time::Instant::now() >= deadline => {3598 bail!(3599 "compositor pointer position did not converge to {},{}; last observed position was {:?}: {error:#}",3600 expected_position.x,3601 expected_position.y,3602 input_ack.pointer_position3603 );3604 }3605 Err(error) => return Err(error),3606 };3607 }3608}360936103611mod tests {3612 use super::*;36133614 3615 async fn expected_input_ack_skips_older_feedback_under_one_deadline() {3616 let (mut sender, mut receiver) = tokio::io::duplex(512);3617 for sequence in [0, 0, 1] {3618 sigil_protocol::write_input_ack(3619 &mut sender,3620 &sigil_protocol::InputAck {3621 sequence,3622 pointer_position: None,3623 pointer_visible: Some(false),3624 },3625 )3626 .await3627 .unwrap();3628 }36293630 let input_ack = read_expected_input_ack(&mut receiver, 1, 1).await.unwrap();36313632 assert_eq!(input_ack.sequence, 1);3633 }36343635 3636 async fn expected_input_ack_rejects_future_feedback() {3637 let (mut sender, mut receiver) = tokio::io::duplex(256);3638 sigil_protocol::write_input_ack(3639 &mut sender,3640 &sigil_protocol::InputAck {3641 sequence: 2,3642 pointer_position: None,3643 pointer_visible: Some(false),3644 },3645 )3646 .await3647 .unwrap();36483649 let error = read_expected_input_ack(&mut receiver, 1, 1)3650 .await3651 .unwrap_err();36523653 assert!(error.to_string().contains("expected at most 1"));3654 }36553656 3657 async fn pointer_confirmation_polls_until_the_tracker_reports_the_exact_position() {3658 let target = PointerPosition::new(1_280, 800).unwrap();3659 let initial_ack = InputAck {3660 sequence: 2,3661 pointer_position: Some(PointerPosition::new(10, 20).unwrap()),3662 pointer_visible: Some(false),3663 };3664 let (client, server) = tokio::io::duplex(512);3665 let (mut client_recv, mut client_send) = tokio::io::split(client);3666 let (mut server_recv, mut server_send) = tokio::io::split(server);3667 let server_task = tokio::spawn(async move {3668 let event = sigil_protocol::read_input_event(&mut server_recv)3669 .await3670 .unwrap()3671 .unwrap();3672 assert_eq!(event, InputEvent::Probe);3673 sigil_protocol::write_input_ack(3674 &mut server_send,3675 &InputAck {3676 sequence: 3,3677 pointer_position: Some(target),3678 pointer_visible: Some(false),3679 },3680 )3681 .await3682 .unwrap();3683 });3684 let mut expected_sequence = 2;36853686 let observed = confirm_pointer_position(3687 &mut client_send,3688 &mut client_recv,3689 &mut expected_sequence,3690 initial_ack,3691 target,3692 Duration::from_secs(1),3693 )3694 .await3695 .unwrap();36963697 assert_eq!(observed, target);3698 assert_eq!(expected_sequence, 3);3699 server_task.await.unwrap();3700 }37013702 3703 async fn pointer_confirmation_fails_when_the_tracker_never_converges() {3704 let target = PointerPosition::new(1_280, 800).unwrap();3705 let initial_ack = InputAck {3706 sequence: 2,3707 pointer_position: Some(PointerPosition::new(10, 20).unwrap()),3708 pointer_visible: Some(false),3709 };3710 let (client, server_guard) = tokio::io::duplex(512);3711 let (mut client_recv, mut client_send) = tokio::io::split(client);3712 let mut expected_sequence = 2;37133714 let error = confirm_pointer_position(3715 &mut client_send,3716 &mut client_recv,3717 &mut expected_sequence,3718 initial_ack,3719 target,3720 Duration::from_millis(25),3721 )3722 .await3723 .unwrap_err();37243725 assert!(error.to_string().contains("did not converge to 1280,800"));3726 drop(server_guard);3727 }37283729 3730 fn feedback_report_is_complete_and_keeps_the_observed_sequence() {3731 let report = complete_feedback_report(7, MediaFeedbackFlags::NONE, 42);3732 report.validate().unwrap();3733 assert_eq!(report.report_id, 7);3734 assert_eq!(report.interval_ms, 1_000);3735 assert_eq!(report.last_sequence, Some(42));3736 assert_eq!(report.frontend_queue_depth, 0);3737 assert_eq!(report.decode_queue_depth, 0);3738 assert_eq!(report.presenter_queue_depth, 0);3739 assert_eq!(report.transport_delivery_p95_ms, Some(10));3740 assert_eq!(report.decode_p95_ms, Some(3));3741 assert_eq!(report.presentation_p95_ms, Some(5));37423743 let later = complete_feedback_report(8, MediaFeedbackFlags::NONE, 42);3744 assert_eq!(later.last_sequence, report.last_sequence);3745 }37463747 3748 fn resolution_stall_reduced_dimensions_are_exact_even_and_same_aspect() {3749 assert_eq!(3750 resolution_stall_reduced_dimensions((2_560, 1_600)).unwrap(),3751 (1_920, 1_200)3752 );3753 assert_eq!(3754 resolution_stall_reduced_dimensions((1_280, 800)).unwrap(),3755 (960, 600)3756 );3757 assert!(resolution_stall_reduced_dimensions((1_278, 800)).is_err());3758 assert!(resolution_stall_reduced_dimensions((80, 64)).is_err());3759 }37603761 3762 fn sequence_checks_reject_duplicates_regressions_and_overflow() {3763 assert_eq!(sequence_gap(41, 42).unwrap(), 0);3764 assert_eq!(sequence_gap(41, 45).unwrap(), 3);3765 assert!(sequence_gap(41, 41).is_err());3766 assert!(sequence_gap(41, 40).is_err());3767 assert!(sequence_gap(u64::MAX, 0).is_err());3768 }37693770 3771 fn pointer_smoke_uses_negotiated_native_surface_center_and_order() {3772 let dimensions = PointerSurfaceDimensions::new(2_560, 1_600).unwrap();37733774 assert_eq!(3775 pointer_smoke_events(Some(dimensions)).unwrap(),3776 [3777 InputEvent::MousePositionSync { x: 1_280, y: 800 },3778 InputEvent::MouseMoveRelative { dx: 32, dy: 16 },3779 InputEvent::MouseClick { b: 1 },3780 ]3781 );3782 assert_eq!(3783 pointer_smoke_expected_positions(dimensions).unwrap(),3784 [3785 PointerPosition::new(1_280, 800).unwrap(),3786 PointerPosition::new(1_312, 816).unwrap(),3787 ]3788 );3789 }37903791 3792 fn pointer_smoke_requires_negotiated_native_surface_dimensions() {3793 let error = pointer_smoke_events(None).unwrap_err();37943795 assert!(3796 error3797 .to_string()3798 .contains("host did not advertise pointer surface dimensions")3799 );3800 }38013802 fn media_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {3803 let payload = vec![0x65, 0x88, 0x84];3804 MediaFrame::new(3805 sigil_protocol::MediaFrameHeader::h264(3806 1_280,3807 800,3808 payload.len(),3809 sequence,3810 sequence * 1_000,3811 i64::try_from(sequence * 1_000).unwrap(),3812 flags,3813 )3814 .unwrap(),3815 payload,3816 )3817 .unwrap()3818 }38193820 fn media_object_v3(3821 group_id: u64,3822 object_id: u32,3823 sequence: u64,3824 flags: FrameFlags,3825 ) -> MediaObjectV3 {3826 let payload = vec![0x65, 0x88, 0x84];3827 let header = sigil_protocol::MediaObjectHeaderV3::h264(3828 1_280,3829 800,3830 payload.len(),3831 if object_id == 0 { 0 } else { 128 },3832 flags,3833 object_id,3834 group_id,3835 sequence,3836 sequence * 1_000,3837 i64::try_from(sequence * 1_000).unwrap(),3838 100,3839 )3840 .unwrap();3841 MediaObjectV3::new(header, payload).unwrap()3842 }38433844 fn media_outcome_v3(3845 accept_index: u64,3846 group_id: u64,3847 object_id: u32,3848 sequence: u64,3849 flags: FrameFlags,3850 ) -> MediaObjectOutcomeV3 {3851 MediaObjectOutcomeV3::Object {3852 accept_index,3853 object: media_object_v3(group_id, object_id, sequence, flags),3854 }3855 }38563857 3858 fn media_v3_discontinuity_object_zero_fast_forwards_accept_order() {3859 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3860 let barrier = keyframe.union(FrameFlags::DISCONTINUITY);3861 let mut reorder = MediaObjectReorderV3::new(0);3862 let mut sequence = MediaObjectSequenceV3::new();38633864 assert!(3865 reorder3866 .push(media_outcome_v3(1, 10, 1, 11, FrameFlags::NONE))3867 .unwrap()3868 .is_none()3869 );3870 let recovered = reorder3871 .push(media_outcome_v3(2, 20, 0, 20, barrier))3872 .unwrap()3873 .unwrap();3874 assert_eq!(recovered.accept_index(), 2);3875 let MediaObjectOutcomeV3::Object { object, .. } = recovered else {3876 panic!("recovery barrier must remain an object");3877 };3878 assert_eq!(3879 sequence.classify(&object),3880 MediaObjectDecisionV3::Deliver {3881 discontinuity: true3882 }3883 );3884 assert_eq!(reorder.pending_len(), 0);3885 assert!(3886 reorder3887 .push(MediaObjectOutcomeV3::Dropped { accept_index: 0 })3888 .unwrap()3889 .is_none()3890 );3891 let next = reorder3892 .push(media_outcome_v3(3, 20, 1, 21, FrameFlags::NONE))3893 .unwrap()3894 .unwrap();3895 let MediaObjectOutcomeV3::Object { object, .. } = next else {3896 panic!("recovered-group delta must remain an object");3897 };3898 assert_eq!(3899 sequence.classify(&object),3900 MediaObjectDecisionV3::Deliver {3901 discontinuity: false3902 }3903 );3904 }39053906 3907 fn minimum_fps_parser_and_first_to_last_evaluation_are_bounded() {3908 assert_eq!(parse_minimum_fps("55").unwrap(), 55.0);3909 assert_eq!(parse_minimum_fps("240").unwrap(), 240.0);3910 for invalid in ["0", "-1", "240.1", "NaN", "inf", "not-a-number"] {3911 assert!(parse_minimum_fps(invalid).is_err(), "{invalid}");3912 }39133914 assert_eq!(accepted_frame_rate(56, Duration::from_secs(1)), Some(55.0));3915 assert_eq!(accepted_frame_rate(1, Duration::from_secs(1)), None);3916 assert_eq!(accepted_frame_rate(2, Duration::ZERO), None);3917 assert!(accepted_frame_rate(56, Duration::from_secs(1)).unwrap() >= 55.0);3918 assert!(accepted_frame_rate(55, Duration::from_secs(1)).unwrap() < 55.0);3919 }39203921 3922 fn media_v3_enforces_wire_group_object_and_sequence_continuity() {3923 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3924 let barrier = keyframe.union(FrameFlags::DISCONTINUITY);3925 let mut sequence = MediaObjectSequenceV3::new();39263927 assert_eq!(3928 sequence.classify(&media_object_v3(10, 0, 10, keyframe)),3929 MediaObjectDecisionV3::Deliver {3930 discontinuity: true3931 }3932 );3933 assert_eq!(3934 sequence.classify(&media_object_v3(10, 1, 11, FrameFlags::NONE)),3935 MediaObjectDecisionV3::Deliver {3936 discontinuity: false3937 }3938 );3939 assert_eq!(3940 sequence.classify(&media_object_v3(10, 3, 13, FrameFlags::NONE)),3941 MediaObjectDecisionV3::DropUntilKeyframe3942 );3943 assert_eq!(3944 sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),3945 MediaObjectDecisionV3::DropUntilKeyframe3946 );3947 assert_eq!(3948 sequence.classify(&media_object_v3(20, 0, 20, barrier)),3949 MediaObjectDecisionV3::Deliver {3950 discontinuity: true3951 }3952 );3953 assert_eq!(3954 sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),3955 MediaObjectDecisionV3::DropLate3956 );3957 }39583959 3960 fn explicit_v3_recovery_never_delivers_same_group_delta_history() {3961 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3962 let barrier = keyframe.union(FrameFlags::DISCONTINUITY);3963 let mut sequence = MediaObjectSequenceV3::new();39643965 assert!(matches!(3966 sequence.classify(&media_object_v3(10, 0, 10, keyframe)),3967 MediaObjectDecisionV3::Deliver { .. }3968 ));3969 assert!(matches!(3970 sequence.classify(&media_object_v3(10, 1, 11, FrameFlags::NONE)),3971 MediaObjectDecisionV3::Deliver { .. }3972 ));3973 sequence.request_recovery();3974 assert_eq!(3975 sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),3976 MediaObjectDecisionV3::DropUntilKeyframe3977 );3978 assert_eq!(3979 sequence.classify(&media_object_v3(20, 0, 20, keyframe)),3980 MediaObjectDecisionV3::DropUntilKeyframe,3981 "recovery after skipped history must be explicitly discontinuous"3982 );3983 assert_eq!(3984 sequence.classify(&media_object_v3(30, 0, 30, barrier)),3985 MediaObjectDecisionV3::Deliver {3986 discontinuity: true3987 }3988 );3989 }39903991 3992 fn media_v3_group_payload_is_bounded_by_the_shared_protocol_limit() {3993 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3994 let mut sequence = MediaObjectSequenceV3::new();3995 assert!(matches!(3996 sequence.classify(&media_object_v3(1, 0, 1, keyframe)),3997 MediaObjectDecisionV3::Deliver { .. }3998 ));3999 sequence.group_payload_bytes = MAX_MEDIA_GROUP_BYTES_V3;4000 assert_eq!(4001 sequence.classify(&media_object_v3(1, 1, 2, FrameFlags::NONE)),4002 MediaObjectDecisionV3::DropUntilKeyframe4003 );4004 }40054006 4007 fn upstream_moq_group_and_object_limits_fail_closed() {4008 assert!(!classify_moq_probe_group(None, 8).unwrap());4009 assert!(!classify_moq_probe_group(Some(8), 9).unwrap());4010 assert!(classify_moq_probe_group(Some(8), 10).unwrap());4011 assert!(classify_moq_probe_group(Some(8), 8).is_err());4012 assert!(classify_moq_probe_group(Some(8), 7).is_err());40134014 assert_eq!(validate_moq_probe_object_bounds(4, 0, 0, 42).unwrap(), 42);4015 assert!(4016 validate_moq_probe_object_bounds(4, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,)4017 .is_err()4018 );4019 assert!(validate_moq_probe_object_bounds(4, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());4020 assert!(validate_moq_probe_object_bounds(4, 0, usize::MAX, 1).is_err());4021 }40224023 4024 fn upstream_moq_requires_configured_group_zero_and_idr_recovery() {4025 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);4026 let recovery = keyframe.union(FrameFlags::DISCONTINUITY);4027 let delta = media_frame(11, FrameFlags::NONE);4028 let configured = media_frame(10, keyframe);4029 let configured_next = media_frame(12, keyframe);4030 let recovered = media_frame(20, recovery);40314032 assert!(validate_moq_probe_frame(3, true, None, &delta).is_err());4033 assert!(validate_moq_probe_frame(3, true, None, &configured).is_ok());4034 assert!(validate_moq_probe_frame(4, true, Some(11), &configured).is_err());4035 assert!(validate_moq_probe_frame(4, true, Some(11), &configured_next).is_ok());4036 assert!(validate_moq_probe_frame(4, true, Some(11), &recovered).is_ok());4037 assert!(validate_moq_probe_frame(4, false, Some(10), &delta).is_ok());4038 assert!(validate_moq_probe_frame(4, false, Some(9), &delta).is_err());4039 }40404041 4042 fn upstream_moq_transport_reports_the_pinned_native_alpn() {4043 assert_eq!(MediaTransport::UpstreamMoq.label(), "iroh-moq");4044 assert_eq!(MediaTransport::UpstreamMoq.alpn(), CONTROL_ALPN_V1);4045 assert_eq!(4046 MediaTransport::UpstreamMoq.media_alpn_label(),4047 "moq-lite-04"4048 );4049 assert_eq!(4050 MediaTransport::UpstreamMoq.media_alpn_label().as_bytes(),4051 iroh_moq::ALPN4052 );4053 }40544055 fn test_invitation(4056 host: &SecretKey,4057 peer: &SecretKey,4058 grants: InvitationGrants,4059 issued_at_unix: u64,4060 expires_at_unix: u64,4061 ) -> String {4062 let claims = sigil_protocol::InvitationClaims::new(4063 *host.public().as_bytes(),4064 *peer.public().as_bytes(),4065 issued_at_unix,4066 expires_at_unix,4067 1,4068 [0x55; 32],4069 grants,4070 )4071 .unwrap();4072 SignedInvitation::issue(claims, &host.to_bytes())4073 .unwrap()4074 .encode()4075 }40764077 fn write_private_invitation(path: &Path, token: &str, ending: &[u8]) {4078 let mut bytes = token.as_bytes().to_vec();4079 bytes.extend_from_slice(ending);4080 std::fs::write(path, bytes).unwrap();4081 4082 {4083 use std::os::unix::fs::PermissionsExt as _;4084 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).unwrap();4085 }4086 }40874088 4089 fn persistent_probe_credentials_are_explicit_and_paired() {4090 let node_id = SecretKey::from_bytes(&[0x2a; 32]).public().to_string();4091 assert!(4092 Args::try_parse_from([4093 "sigil-probe",4094 "--node-id",4095 &node_id,4096 "--invitation",4097 "probe.goq-invite",4098 ])4099 .is_err()4100 );4101 let persistent = Args::try_parse_from([4102 "sigil-probe",4103 "--node-id",4104 &node_id,4105 "--identity",4106 "probe.key",4107 "--invitation",4108 "probe.goq-invite",4109 ])4110 .unwrap();4111 assert_eq!(persistent.identity, Some(PathBuf::from("probe.key")));4112 assert_eq!(4113 persistent.invitation,4114 Some(PathBuf::from("probe.goq-invite"))4115 );4116 }41174118 4119 fn private_invitation_preflight_accepts_only_the_bound_current_grants() {4120 let temp = tempfile::tempdir().unwrap();4121 let path = temp.path().join("probe.goq-invite");4122 let host = SecretKey::from_bytes(&[0x31; 32]);4123 let peer = SecretKey::from_bytes(&[0x32; 32]);4124 let other = SecretKey::from_bytes(&[0x33; 32]);4125 let now = SystemTime::now()4126 .duration_since(UNIX_EPOCH)4127 .unwrap()4128 .as_secs();4129 let token = test_invitation(4130 &host,4131 &peer,4132 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),4133 now.saturating_sub(1),4134 now + 600,4135 );41364137 write_private_invitation(&path, &token, b"\n");4138 assert_eq!(4139 load_invitation_file_at(4140 &path,4141 host.public(),4142 peer.public(),4143 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),4144 now,4145 )4146 .unwrap(),4147 token4148 );4149 write_private_invitation(&path, &token, b"\r\n");4150 assert!(4151 load_invitation_file_at(4152 &path,4153 host.public(),4154 peer.public(),4155 InvitationGrants::VIEW,4156 now,4157 )4158 .is_ok()4159 );4160 assert!(4161 load_invitation_file_at(4162 &path,4163 other.public(),4164 peer.public(),4165 InvitationGrants::VIEW,4166 now,4167 )4168 .is_err()4169 );4170 assert!(4171 load_invitation_file_at(4172 &path,4173 host.public(),4174 other.public(),4175 InvitationGrants::VIEW,4176 now,4177 )4178 .is_err()4179 );4180 assert!(4181 load_invitation_file_at(4182 &path,4183 host.public(),4184 peer.public(),4185 InvitationGrants::POINTER_KEYBOARD,4186 now,4187 )4188 .is_err()4189 );41904191 let view_only = test_invitation(4192 &host,4193 &peer,4194 InvitationGrants::VIEW,4195 now.saturating_sub(1),4196 now + 600,4197 );4198 write_private_invitation(&path, &view_only, b"\n");4199 assert!(4200 load_invitation_file_at(4201 &path,4202 host.public(),4203 peer.public(),4204 InvitationGrants::VIEW,4205 now,4206 )4207 .unwrap_err()4208 .to_string()4209 .contains("input acknowledgment")4210 );42114212 let expired = test_invitation(4213 &host,4214 &peer,4215 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),4216 now.saturating_sub(120),4217 now.saturating_sub(60),4218 );4219 write_private_invitation(&path, &expired, b"\n");4220 assert!(4221 load_invitation_file_at(4222 &path,4223 host.public(),4224 peer.public(),4225 InvitationGrants::VIEW,4226 now,4227 )4228 .unwrap_err()4229 .to_string()4230 .contains("expired")4231 );4232 write_private_invitation(&path, &token, b"\n\n");4233 assert!(4234 load_invitation_file_at(4235 &path,4236 host.public(),4237 peer.public(),4238 InvitationGrants::VIEW,4239 now,4240 )4241 .is_err()4242 );4243 let future = test_invitation(4244 &host,4245 &peer,4246 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),4247 now + INVITATION_CLOCK_SKEW_SECS + 1,4248 now + INVITATION_CLOCK_SKEW_SECS + 61,4249 );4250 write_private_invitation(&path, &future, b"\n");4251 assert!(4252 load_invitation_file_at(4253 &path,4254 host.public(),4255 peer.public(),4256 InvitationGrants::VIEW,4257 now,4258 )4259 .unwrap_err()4260 .to_string()4261 .contains("future")4262 );4263 }42644265 4266 4267 fn invitation_reader_rejects_unsafe_files() {4268 use std::os::unix::fs::{PermissionsExt as _, symlink};42694270 let temp = tempfile::tempdir().unwrap();4271 let path = temp.path().join("probe.goq-invite");4272 let link = temp.path().join("linked.goq-invite");4273 let host = SecretKey::from_bytes(&[0x41; 32]);4274 let peer = SecretKey::from_bytes(&[0x42; 32]);4275 let now = SystemTime::now()4276 .duration_since(UNIX_EPOCH)4277 .unwrap()4278 .as_secs();4279 let token = test_invitation(4280 &host,4281 &peer,4282 InvitationGrants::VIEW,4283 now.saturating_sub(1),4284 now + 60,4285 );4286 write_private_invitation(&path, &token, b"\n");4287 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();4288 assert!(4289 load_invitation_file(&path, host.public(), peer.public(), InvitationGrants::VIEW,)4290 .is_err()4291 );4292 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();4293 symlink(&path, &link).unwrap();4294 assert!(4295 load_invitation_file(&link, host.public(), peer.public(), InvitationGrants::VIEW,)4296 .is_err()4297 );4298 std::fs::write(&path, vec![b'x'; MAX_INVITATION_FILE_BYTES as usize + 1]).unwrap();4299 assert!(4300 load_invitation_file(&path, host.public(), peer.public(), InvitationGrants::VIEW,)4301 .is_err()4302 );4303 }43044305 4306 fn invitation_is_attached_only_when_the_caller_selects_the_media_hello() {4307 let host = SecretKey::from_bytes(&[0x51; 32]);4308 let peer = SecretKey::from_bytes(&[0x52; 32]);4309 let now = SystemTime::now()4310 .duration_since(UNIX_EPOCH)4311 .unwrap()4312 .as_secs();4313 let token = test_invitation(4314 &host,4315 &peer,4316 InvitationGrants::VIEW,4317 now.saturating_sub(1),4318 now + 60,4319 );4320 let media = probe_client_hello([7; 16], vec![Capability::VideoH264], Some(&token));4321 let input = probe_client_hello([7; 16], vec![Capability::InputAck], None);4322 assert_eq!(media.invitation.as_deref(), Some(token.as_str()));4323 assert_eq!(input.invitation, None);4324 }43254326 4327 fn media_compatibility_flag_defaults_and_conflicts_are_enforced() {4328 let node_id = SecretKey::from_bytes(&[0x2a; 32]).public().to_string();4329 let default = Args::try_parse_from(["sigil-probe", "--node-id", &node_id]).unwrap();4330 assert!(!default.media_v3);4331 assert_eq!(default.identity, None);4332 assert_eq!(default.invitation, None);4333 assert_eq!(default.expect_size, None);4334 assert!(!default.feedback_smoke);4335 let strict = Args::try_parse_from([4336 "sigil-probe",4337 "--node-id",4338 &node_id,4339 "--expect-size",4340 "2560x1600",4341 ])4342 .unwrap();4343 assert_eq!(strict.expect_size, Some((2_560, 1_600)));4344 let feedback =4345 Args::try_parse_from(["sigil-probe", "--node-id", &node_id, "--feedback-smoke"])4346 .unwrap();4347 assert!(feedback.feedback_smoke);4348 assert!(4349 Args::try_parse_from([4350 "sigil-probe",4351 "--node-id",4352 &node_id,4353 "--feedback-smoke",4354 "--media-v3",4355 ])4356 .is_err()4357 );4358 }4359}