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;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 CONTROL_ALPN_V1, Capability, ClientHello, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN,18 GAMEPAD_TRIGGER_MAX, GamepadState, INPUT_ALPN_V1, INVITATION_CLOCK_SKEW_SECS, InputAck,19 InputEvent, InvitationGrants, KeyframeRequestReasonV3, MAX_INVITATION_TOKEN_LEN,20 MAX_MEDIA_GROUP_BYTES_V3, MAX_MEDIA_OBJECT_ID_V3, MEDIA_ALPN_V3, MEDIA_FEEDBACK_ALPN_V1,21 MediaCodec, MediaControlRequestV3, MediaFeedbackFlags, MediaFeedbackReportV1, MediaFrame,22 MediaObjectV3, PointerPosition, PointerSurfaceDimensions, ProtocolError, SignedInvitation,23 decode_media_frame_object, media_moq_broadcast_name, read_adaptive_bitrate_decision_v1,24 read_host_hello, read_input_ack, read_media_object_v3, write_client_hello, write_input_event,25 write_media_control_request_v3, write_media_feedback_report_v1,26};272829use std::os::unix::fs::{MetadataExt, OpenOptionsExt};30313233mod identity;3435mod moq_catalog;3637use moq_catalog::{MoqCatalogMode, subscribe_goq_video_track};3839const MEDIA_OBJECT_CAPACITY: usize = 4;40414243struct Args {44 45 node_id: EndpointId,46 /// Owner-only 32-byte Iroh identity. Omit for the existing ephemeral probe.47 48 identity: Option<PathBuf>,49 /// Remove all direct IP transports from the probe endpoint so control,50 /// media, and input are forced through Iroh relay paths.51 52 relay_only: bool,53 /// Owner-only one-time Sigil invitation file. Sent only on the first media54 /// handshake and requires a persistent identity bound to that invitation.55 56 invitation: Option<PathBuf>,57 58 frames: u32,59 60 timeout_seconds: u64,61 /// Fail unless accepted media frames sustain at least this first-to-last62 /// cadence. Intended for live appliance performance gates.63 64 minimum_fps: Option<f64>,65 /// Optional strict encoded-size assertion. When omitted, accept and report66 /// the host's observed bounded dimensions.67 68 expect_size: Option<(u16, u16)>,69 /// Exercise custom grouped v3 media instead of upstream MoQ. Intended only70 /// for compatibility validation.71 72 media_v3: bool,73 /// Request a configured recovery keyframe after three accepted frames,74 /// then prove no delta history is delivered before the recovery barrier.75 76 keyframe_smoke: bool,77 /// Correlation identifier for `--keyframe-smoke` host evidence.78 79 keyframe_request_id: u64,80 /// Stop reading upstream MoQ after the first configured IDR, then require81 /// the transport to cancel that stale GOP and resume at a newer configured82 /// IDR within a bounded interval. Intended for latency-growth proofs.83 8485868788 slow_consumer_ms: Option<u64>,89 /// Require gamepad negotiation and emit one bounded non-neutral snapshot90 /// followed by neutral. Intended for evtest-backed uinput proof.91 92 gamepad_smoke: bool,93 /// Require relative-pointer negotiation and emit bounded motion plus one94 /// complete left-click. Intended for libinput/Gamescope-backed proof.95 96 pointer_smoke: bool,97 /// Require the host to return an immediately available compositor pointer98 /// position and visibility sample. Intended for Gamescope restart proofs.99 100 pointer_feedback_smoke: bool,101 /// Attach the media-feedback stream to the active session, submit one102 /// complete report, and print the host's bounded decision. Intended for103 /// validating shadow-only encoder configurations.104 105106107 )]108 feedback_smoke: bool,109 /// Prove adaptive-resolution liveness across a controlled Gamescope stall.110 /// GATE_DIR coordinates the external SIGSTOP/SIGCONT harness.111 112113114115116117118119120121 )]122 resolution_stall_smoke: Option<PathBuf>,123}124125126enum MediaTransport {127 UpstreamMoq,128 GroupedV3,129}130131impl MediaTransport {132 fn alpn(self) -> &'static [u8] {133 match self {134 Self::UpstreamMoq => CONTROL_ALPN_V1,135 Self::GroupedV3 => MEDIA_ALPN_V3,136 }137 }138139 fn label(self) -> &'static str {140 match self {141 Self::UpstreamMoq => "iroh-moq",142 Self::GroupedV3 => "grouped-v3",143 }144 }145146 fn media_alpn_label(self) -> &'static str {147 match self {148 Self::UpstreamMoq => std::str::from_utf8(iroh_moq::ALPN)149 .expect("pinned iroh-moq ALPN must be printable UTF-8"),150 Self::GroupedV3 => "sigil/media/3",151 }152 }153}154155const KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES: u32 = 3;156const KEYFRAME_SMOKE_MINIMUM_FRAMES: u32 = KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES + 1;157const SLOW_CONSUMER_MINIMUM_FRAMES: u32 = 2;158const SLOW_CONSUMER_RECOVERY_TIMEOUT: Duration = Duration::from_secs(2);159const RESOLUTION_STALL_FEEDBACK_INTERVAL: Duration = Duration::from_secs(1);160const RESOLUTION_STALL_FRESH_REPORTS: u64 = 10;161const RESOLUTION_STALL_FEEDBACK_TIMEOUT: Duration = Duration::from_secs(5);162const RESOLUTION_STALL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(2);163const RESOLUTION_STALL_GATE_POLL_INTERVAL: Duration = Duration::from_millis(50);164const MAX_INVITATION_FILE_BYTES: u64 = (MAX_INVITATION_TOKEN_LEN + 2) as u64;165166167enum MediaObjectOutcomeV3 {168 Object {169 accept_index: u64,170 object: MediaObjectV3,171 },172 Dropped {173 accept_index: u64,174 },175 Malformed {176 accept_index: u64,177 error: ProtocolError,178 },179}180181impl MediaObjectOutcomeV3 {182 fn accept_index(&self) -> u64 {183 match self {184 Self::Object { accept_index, .. }185 | Self::Dropped { accept_index }186 | Self::Malformed { accept_index, .. } => *accept_index,187 }188 }189190 fn is_fast_forward_barrier(&self) -> bool {191 let Self::Object { object, .. } = self else {192 return false;193 };194 object.header.object_id == 0195 && object.header.flags.contains(FrameFlags::KEYFRAME)196 && object.header.flags.contains(FrameFlags::CODEC_CONFIG)197 && object.header.flags.contains(FrameFlags::DISCONTINUITY)198 }199}200201202struct MediaObjectReorderV3 {203 next_accept_index: u64,204 completed: BTreeMap<u64, MediaObjectOutcomeV3>,205}206207impl MediaObjectReorderV3 {208 fn new(first_accept_index: u64) -> Self {209 Self {210 next_accept_index: first_accept_index,211 completed: BTreeMap::new(),212 }213 }214215 fn pending_len(&self) -> usize {216 self.completed.len()217 }218219 fn push(&mut self, outcome: MediaObjectOutcomeV3) -> Result<Option<MediaObjectOutcomeV3>> {220 if matches!(outcome, MediaObjectOutcomeV3::Malformed { .. }) {221 return Ok(Some(outcome));222 }223 let accept_index = outcome.accept_index();224 if accept_index < self.next_accept_index {225 // A discontinuity barrier may advance beyond older in-flight226 // reads. Their eventual timeout/reset outcomes belong to the227 // superseded GOP and must not poison the recovered sequence.228 return Ok(None);229 }230 let fast_forward = outcome.is_fast_forward_barrier();231 ensure!(232 self.completed.insert(accept_index, outcome).is_none(),233 "media v3 accept index {accept_index} completed more than once"234 );235 if fast_forward {236 self.completed.retain(|index, _| *index >= accept_index);237 self.next_accept_index = accept_index;238 }239 self.take_next()240 }241242 fn take_next(&mut self) -> Result<Option<MediaObjectOutcomeV3>> {243 let Some(outcome) = self.completed.remove(&self.next_accept_index) else {244 return Ok(None);245 };246 self.next_accept_index = self247 .next_accept_index248 .checked_add(1)249 .context("media v3 accept index overflowed")?;250 Ok(Some(outcome))251 }252}253254struct MediaObjectReceiverV3 {255 connection: iroh::endpoint::Connection,256 reads: tokio::task::JoinSet<MediaObjectOutcomeV3>,257 reorder: MediaObjectReorderV3,258 next_accept_index: u64,259 accepting: bool,260 read_timeout: Duration,261}262263impl MediaObjectReceiverV3 {264 fn new(connection: iroh::endpoint::Connection, read_timeout: Duration) -> Self {265 Self {266 connection,267 reads: tokio::task::JoinSet::new(),268 reorder: MediaObjectReorderV3::new(0),269 next_accept_index: 0,270 accepting: true,271 read_timeout,272 }273 }274275 async fn next(&mut self) -> Result<Option<MediaObjectOutcomeV3>> {276 loop {277 if let Some(completed) = self.reorder.take_next()? {278 return Ok(Some(completed));279 }280 if !self.accepting && self.reads.is_empty() {281 ensure!(282 self.reorder.pending_len() == 0,283 "media v3 connection closed with an incomplete object order"284 );285 return Ok(None);286 }287288 tokio::select! {289 accepted = self.connection.accept_uni(),290 if self.accepting291 && self.reads.len() + self.reorder.pending_len()292 < MEDIA_OBJECT_CAPACITY => {293 match accepted {294 Ok(mut stream) => {295 let accept_index = self.next_accept_index;296 self.next_accept_index = self297 .next_accept_index298 .checked_add(1)299 .context("media v3 accept index overflowed")?;300 let read_timeout = self.read_timeout;301 self.reads.spawn(async move {302 match tokio::time::timeout(303 read_timeout,304 read_media_object_v3(&mut stream),305 )306 .await307 {308 Ok(Ok(object)) => MediaObjectOutcomeV3::Object {309 accept_index,310 object,311 },312 Ok(Err(ProtocolError::Io(_))) | Err(_) => {313 MediaObjectOutcomeV3::Dropped { accept_index }314 }315 Ok(Err(error)) => MediaObjectOutcomeV3::Malformed {316 accept_index,317 error,318 },319 }320 });321 }322 Err(_) => self.accepting = false,323 }324 }325 completed = self.reads.join_next(), if !self.reads.is_empty() => {326 match completed.expect("guarded by non-empty media v3 object task set") {327 Ok(outcome) => {328 if let Some(completed) = self.reorder.push(outcome)? {329 return Ok(Some(completed));330 }331 }332 Err(error) if error.is_cancelled() => continue,333 Err(error) => return Err(error).context("media v3 object read task failed"),334 }335 }336 }337 }338 }339}340341impl Drop for MediaObjectReceiverV3 {342 fn drop(&mut self) {343 self.reads.abort_all();344 }345}346347348enum MediaObjectDecisionV3 {349 Deliver { discontinuity: bool },350 DropLate,351 DropUntilKeyframe,352}353354355struct MediaObjectSequenceV3 {356 group_id: Option<u64>,357 last_object_id: Option<u32>,358 last_sequence: Option<u64>,359 group_payload_bytes: usize,360 waiting_for_keyframe: bool,361}362363impl MediaObjectSequenceV3 {364 fn new() -> Self {365 Self {366 waiting_for_keyframe: true,367 ..Self::default()368 }369 }370371 fn request_recovery(&mut self) {372 self.waiting_for_keyframe = true;373 }374375 fn note_drop(&mut self) -> bool {376 let entered = !self.waiting_for_keyframe;377 self.waiting_for_keyframe = true;378 entered379 }380381 fn classify(&mut self, object: &MediaObjectV3) -> MediaObjectDecisionV3 {382 let header = &object.header;383 if self384 .group_id385 .is_some_and(|group_id| header.group_id < group_id)386 || self387 .last_sequence388 .is_some_and(|sequence| header.sequence <= sequence)389 {390 return MediaObjectDecisionV3::DropLate;391 }392393 let new_group = self.group_id != Some(header.group_id);394 let configured_group_start = header.object_id == 0395 && header.flags.contains(FrameFlags::KEYFRAME)396 && header.flags.contains(FrameFlags::CODEC_CONFIG);397 if new_group && !configured_group_start {398 self.waiting_for_keyframe = true;399 return MediaObjectDecisionV3::DropUntilKeyframe;400 }401 if !new_group && self.waiting_for_keyframe {402 return MediaObjectDecisionV3::DropUntilKeyframe;403 }404405 let sequence_contiguous = self406 .last_sequence407 .is_none_or(|sequence| sequence.checked_add(1) == Some(header.sequence));408 let object_contiguous = new_group409 || self410 .last_object_id411 .is_some_and(|object_id| object_id.checked_add(1) == Some(header.object_id));412 let Some(next_group_bytes) = (if new_group {413 Some(object.payload.len())414 } else {415 self.group_payload_bytes.checked_add(object.payload.len())416 }) else {417 self.waiting_for_keyframe = true;418 return MediaObjectDecisionV3::DropUntilKeyframe;419 };420 let recovery_requires_discontinuity = new_group421 && self.group_id.is_some()422 && (self.waiting_for_keyframe || !sequence_contiguous)423 && !header.flags.contains(FrameFlags::DISCONTINUITY);424 if (!sequence_contiguous && !new_group)425 || !object_contiguous426 || next_group_bytes > MAX_MEDIA_GROUP_BYTES_V3427 || recovery_requires_discontinuity428 {429 self.waiting_for_keyframe = true;430 return MediaObjectDecisionV3::DropUntilKeyframe;431 }432433 let discontinuity = header.flags.contains(FrameFlags::DISCONTINUITY)434 || self.waiting_for_keyframe435 || !sequence_contiguous;436 self.group_id = Some(header.group_id);437 self.last_object_id = Some(header.object_id);438 self.last_sequence = Some(header.sequence);439 self.group_payload_bytes = next_group_bytes;440 self.waiting_for_keyframe = false;441 MediaObjectDecisionV3::Deliver { discontinuity }442 }443}444445446enum MoqProbeOutcome {447 Frame {448 frame: MediaFrame,449 group_sequence: u64,450 recovery: bool,451 },452 Dropped,453}454455struct MoqProbeGroupCursor {456 group: GroupConsumer,457 sequence: u64,458 object_count: usize,459 object_bytes: usize,460 recovery: bool,461}462463/// Own all upstream handles for the complete proof session. Dropping `Moq`464/// stops its actor, while dropping the consumers cancels their subscriptions.465struct MoqProbeLifetime {466 _moq: Moq,467 session: MoqSession,468 _broadcast: BroadcastConsumer,469}470471struct MoqProbeReceiver {472 lifetime: MoqProbeLifetime,473 track: TrackConsumer,474 catalog_mode: MoqCatalogMode,475 current_group: Option<MoqProbeGroupCursor>,476 last_group_sequence: Option<u64>,477 last_frame_sequence: Option<u64>,478 waiting_for_keyframe: bool,479 pending_recovery: bool,480 cancelled_groups: u64,481 group_gaps: u64,482 maximum_group_objects: usize,483 maximum_group_bytes: usize,484 historical_suffix_frames: u64,485}486487impl MoqProbeReceiver {488 async fn connect(489 endpoint: &Endpoint,490 address: EndpointAddr,491 session_id: u64,492 timeout: Duration,493 ) -> Result<Self> {494 let broadcast_name = media_moq_broadcast_name(session_id)495 .context("deriving session-scoped MoQ broadcast name")?;496 let moq = Moq::new(endpoint.clone());497 let mut session = tokio::time::timeout(timeout, moq.connect(address.clone()))498 .await499 .context("timed out connecting upstream MoQ media session")?500 .context("connecting upstream MoQ media session")?;501 ensure!(502 session.remote_id() == address.id,503 "upstream MoQ connected to unexpected peer {} instead of {}",504 session.remote_id(),505 address.id506 );507 let broadcast = tokio::time::timeout(timeout, session.subscribe(&broadcast_name))508 .await509 .with_context(|| {510 format!("timed out subscribing to upstream MoQ broadcast {broadcast_name}")511 })?512 .with_context(|| format!("subscribing to upstream MoQ broadcast {broadcast_name}"))?;513 let catalog = subscribe_goq_video_track(&broadcast, timeout)514 .await515 .context("resolving Goq MoQ catalog")?;516 Ok(Self {517 lifetime: MoqProbeLifetime {518 _moq: moq,519 session,520 _broadcast: broadcast,521 },522 track: catalog.track,523 catalog_mode: catalog.mode,524 current_group: None,525 last_group_sequence: None,526 last_frame_sequence: None,527 waiting_for_keyframe: true,528 pending_recovery: false,529 cancelled_groups: 0,530 group_gaps: 0,531 maximum_group_objects: 0,532 maximum_group_bytes: 0,533 historical_suffix_frames: 0,534 })535 }536537 fn connection(&self) -> &iroh::endpoint::Connection {538 self.lifetime.session.conn()539 }540541 /// Stop consuming the current group immediately. The track cursor already542 /// advanced when the group was selected, so a subsequent read can only543 /// enter a newer native group; no cancelled delta suffix can be accepted.544 fn request_recovery(&mut self) -> Option<u64> {545 self.waiting_for_keyframe = true;546 self.pending_recovery = true;547 self.current_group.take().map(|cursor| {548 self.cancelled_groups = self.cancelled_groups.saturating_add(1);549 self.last_group_sequence = Some(cursor.sequence);550 cursor.sequence551 })552 }553554 async fn next(&mut self) -> Result<Option<MoqProbeOutcome>> {555 loop {556 if self.current_group.is_none() {557 let group = self558 .track559 .next_group()560 .await561 .context("reading the next sequential upstream MoQ group")?;562 let Some(group) = group else {563 return Ok(None);564 };565 let sequence = group.sequence;566 let group_gap = classify_moq_probe_group(self.last_group_sequence, sequence)?;567 if group_gap {568 self.group_gaps = self.group_gaps.saturating_add(1);569 self.waiting_for_keyframe = true;570 self.pending_recovery = true;571 }572 self.current_group = Some(MoqProbeGroupCursor {573 group,574 sequence,575 object_count: 0,576 object_bytes: 0,577 recovery: self.pending_recovery,578 });579 self.pending_recovery = false;580 }581582 let cursor = self583 .current_group584 .as_mut()585 .expect("MoQ group cursor was initialized");586 let object = match cursor.group.read_frame().await {587 Ok(object) => object,588 Err(error) if moq_probe_error_is_recoverable(&error) => {589 self.cancelled_groups = self.cancelled_groups.saturating_add(1);590 self.last_group_sequence = Some(cursor.sequence);591 self.current_group = None;592 self.waiting_for_keyframe = true;593 self.pending_recovery = true;594 return Ok(Some(MoqProbeOutcome::Dropped));595 }596 Err(error) => {597 bail!(598 "upstream MoQ group {} failed terminally: {error}",599 cursor.sequence600 );601 }602 };603 let Some(object) = object else {604 ensure!(605 cursor.object_count > 0,606 "upstream MoQ group {} was empty",607 cursor.sequence608 );609 self.last_group_sequence = Some(cursor.sequence);610 self.current_group = None;611 continue;612 };613614 let next_bytes = validate_moq_probe_object_bounds(615 cursor.sequence,616 cursor.object_count,617 cursor.object_bytes,618 object.len(),619 )?;620 let frame = decode_media_frame_object(&object).with_context(|| {621 format!(622 "decoding upstream MoQ group {} object {}",623 cursor.sequence, cursor.object_count624 )625 })?;626 let first_object = cursor.object_count == 0;627 let contiguous = validate_moq_probe_frame(628 cursor.sequence,629 first_object,630 self.last_frame_sequence,631 &frame,632 )?;633 let recovery = cursor.recovery634 || self.waiting_for_keyframe635 || (first_object && !contiguous)636 || frame.header.flags.contains(FrameFlags::DISCONTINUITY);637 if self.waiting_for_keyframe && !first_object {638 self.historical_suffix_frames = self.historical_suffix_frames.saturating_add(1);639 bail!(640 "upstream MoQ delivered a historical delta suffix while recovery was pending"641 );642 }643 cursor.object_count += 1;644 cursor.object_bytes = next_bytes;645 self.maximum_group_objects = self.maximum_group_objects.max(cursor.object_count);646 self.maximum_group_bytes = self.maximum_group_bytes.max(next_bytes);647 self.last_frame_sequence = Some(frame.header.sequence);648 self.waiting_for_keyframe = false;649 return Ok(Some(MoqProbeOutcome::Frame {650 frame,651 group_sequence: cursor.sequence,652 recovery,653 }));654 }655 }656}657658fn classify_moq_probe_group(previous: Option<u64>, current: u64) -> Result<bool> {659 let Some(previous) = previous else {660 return Ok(false);661 };662 ensure!(663 current > previous,664 "upstream MoQ group sequence did not increase: previous={previous}, current={current}"665 );666 Ok(previous.checked_add(1) != Some(current))667}668669fn validate_moq_probe_object_bounds(670 group_sequence: u64,671 object_count: usize,672 object_bytes: usize,673 object_len: usize,674) -> Result<usize> {675 let maximum_objects = MAX_MEDIA_OBJECT_ID_V3 as usize + 1;676 ensure!(677 object_count < maximum_objects,678 "upstream MoQ group {group_sequence} exceeded {maximum_objects} objects"679 );680 let next_bytes = object_bytes681 .checked_add(object_len)682 .context("upstream MoQ group byte count overflowed")?;683 ensure!(684 next_bytes <= MAX_MEDIA_GROUP_BYTES_V3,685 "upstream MoQ group {group_sequence} exceeded {MAX_MEDIA_GROUP_BYTES_V3} bytes"686 );687 Ok(next_bytes)688}689690fn validate_moq_probe_frame(691 group_sequence: u64,692 first_object: bool,693 last_frame_sequence: Option<u64>,694 frame: &MediaFrame,695) -> Result<bool> {696 if first_object {697 ensure!(698 frame.header.codec == MediaCodec::H264699 && frame.header.flags.contains(FrameFlags::KEYFRAME)700 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),701 "upstream MoQ group {group_sequence} did not begin with a configured H.264 keyframe"702 );703 // A native group cancellation is itself the transport discontinuity.704 // Portal marks the first configured IDR in the replacement group as a705 // decoder discontinuity even when the embedded compatibility header706 // does not redundantly carry the v3 DISCONTINUITY flag.707 }708 let contiguous = last_frame_sequence709 .is_none_or(|previous| previous.checked_add(1) == Some(frame.header.sequence));710 ensure!(711 last_frame_sequence.is_none_or(|previous| frame.header.sequence > previous),712 "upstream MoQ group {group_sequence} contains a non-monotonic frame sequence"713 );714 ensure!(715 first_object || contiguous,716 "upstream MoQ group {group_sequence} contains a non-contiguous frame sequence"717 );718 Ok(contiguous)719}720721fn moq_probe_error_is_recoverable(error: &moq_net::Error) -> bool {722 matches!(723 error,724 moq_net::Error::Cancel725 | moq_net::Error::Old726 | moq_net::Error::Timeout727 | moq_net::Error::Dropped728 | moq_net::Error::CacheFull729 | moq_net::Error::Remote(0 | 2 | 3 | 24 | 26)730 )731}732733struct AcceptedMedia {734 flags: FrameFlags,735 width: u16,736 height: u16,737 sequence: u64,738 capture_timestamp_us: u64,739 payload_len: usize,740 v3_group_id: Option<u64>,741}742743impl From<MediaObjectV3> for AcceptedMedia {744 fn from(object: MediaObjectV3) -> Self {745 Self {746 flags: object.header.flags,747 width: object.header.width,748 height: object.header.height,749 sequence: object.header.sequence,750 capture_timestamp_us: object.header.capture_timestamp_us,751 payload_len: object.payload.len(),752 v3_group_id: Some(object.header.group_id),753 }754 }755}756757758struct ResolutionStallEvidence {759 native_dimensions: (u16, u16),760 reduced_dimensions: (u16, u16),761 stall_sequence: u64,762 fresh_duration: Duration,763 stall_input_ack_micros: u64,764 resume_input_ack_micros: u64,765 resume_media_micros: u64,766 resume_sequence_advance: u64,767}768769struct ResolutionStallGate {770 ready: PathBuf,771 stalled: PathBuf,772 resume_ready: PathBuf,773 resumed: PathBuf,774}775776impl ResolutionStallGate {777 fn new(directory: &Path) -> Result<Self> {778 ensure!(779 directory.is_dir(),780 "resolution stall gate is not a directory"781 );782 let gate = Self {783 ready: directory.join("ready"),784 stalled: directory.join("stalled"),785 resume_ready: directory.join("resume-ready"),786 resumed: directory.join("resumed"),787 };788 for marker in [789 &gate.ready,790 &gate.stalled,791 &gate.resume_ready,792 &gate.resumed,793 ] {794 ensure!(795 !marker.exists(),796 "resolution stall gate marker already exists: {}",797 marker.display()798 );799 }800 Ok(gate)801 }802}803804fn complete_feedback_report(805 report_id: u64,806 flags: MediaFeedbackFlags,807 last_sequence: u64,808) -> MediaFeedbackReportV1 {809 MediaFeedbackReportV1 {810 report_id,811 interval_ms: 1_000,812 flags,813 last_sequence: Some(last_sequence),814 transport_dropped_delta: 0,815 frontend_dropped_delta: 0,816 decoder_dropped_delta: 0,817 presenter_dropped_delta: 0,818 frontend_queue_depth: 0,819 frontend_queue_capacity: 4,820 decode_queue_depth: 0,821 decode_queue_capacity: 4,822 presenter_queue_depth: 0,823 presenter_queue_capacity: 2,824 transport_delivery_p95_ms: Some(10),825 decode_p95_ms: Some(3),826 presentation_p95_ms: Some(5),827 }828}829830fn resolution_stall_reduced_dimensions(native: (u16, u16)) -> Result<(u16, u16)> {831 ensure!(832 native.0.is_multiple_of(4) && native.1.is_multiple_of(4),833 "native dimensions must be divisible by four for an exact three-quarter tier"834 );835 let reduced = (native.0 / 4 * 3, native.1 / 4 * 3);836 ensure!(837 reduced.0 >= 64 && reduced.1 >= 64,838 "three-quarter dimensions are below the H.264 minimum"839 );840 ensure!(841 reduced.0.is_multiple_of(2) && reduced.1.is_multiple_of(2),842 "three-quarter H.264 dimensions must be even"843 );844 ensure!(845 u32::from(native.0) * u32::from(reduced.1) == u32::from(native.1) * u32::from(reduced.0),846 "three-quarter dimensions must preserve the native aspect ratio"847 );848 Ok(reduced)849}850851async fn exchange_feedback_report(852 send: &mut iroh::endpoint::SendStream,853 recv: &mut iroh::endpoint::RecvStream,854 report: &MediaFeedbackReportV1,855) -> Result<AdaptiveBitrateDecisionV1> {856 tokio::time::timeout(857 RESOLUTION_STALL_FEEDBACK_TIMEOUT,858 write_media_feedback_report_v1(send, report),859 )860 .await861 .context("timed out writing media feedback")??;862 let decision = tokio::time::timeout(863 RESOLUTION_STALL_FEEDBACK_TIMEOUT,864 read_adaptive_bitrate_decision_v1(recv),865 )866 .await867 .context("timed out waiting for media feedback decision")??868 .context("host closed the media feedback stream")?;869 ensure!(870 decision.report_id == report.report_id,871 "feedback decision report ID {} does not match report {}",872 decision.report_id,873 report.report_id874 );875 Ok(decision)876}877878async fn run_feedback_smoke(879 endpoint: &Endpoint,880 address: EndpointAddr,881 nonce: [u8; 16],882 session_id: u64,883 last_sequence: u64,884) -> Result<(AdaptiveBitrateDecisionV1, &'static str, Option<f64>)> {885 let connection = endpoint886 .connect(address, MEDIA_FEEDBACK_ALPN_V1)887 .await888 .context("connecting media feedback protocol")?;889 let (mut send, mut recv) = connection890 .open_bi()891 .await892 .context("opening media feedback stream")?;893 let negotiation = negotiate(894 &mut send,895 &mut recv,896 nonce,897 vec![Capability::VideoH264],898 Capability::VideoH264,899 "media feedback",900 None,901 )902 .await?;903 ensure!(904 negotiation.session_id == session_id,905 "media and feedback session IDs differ"906 );907908 let report = complete_feedback_report(1, MediaFeedbackFlags::NONE, last_sequence);909 let decision = exchange_feedback_report(&mut send, &mut recv, &report).await?;910 let (path_mode, path_rtt_ms) = selected_path_diagnostics(&connection);911 send.finish().context("finishing media feedback stream")?;912 connection.close(0_u32.into(), b"feedback probe complete");913 Ok((decision, path_mode, path_rtt_ms))914}915916async fn next_resolution_stall_moq_frame(receiver: &mut MoqProbeReceiver) -> Result<AcceptedMedia> {917 loop {918 match receiver.next().await? {919 Some(MoqProbeOutcome::Frame {920 frame,921 group_sequence,922 ..923 }) => {924 return Ok(AcceptedMedia {925 flags: frame.header.flags,926 width: frame.header.width,927 height: frame.header.height,928 sequence: frame.header.sequence,929 capture_timestamp_us: frame.header.capture_timestamp_us,930 payload_len: frame.payload.len(),931 v3_group_id: Some(group_sequence),932 });933 }934 Some(MoqProbeOutcome::Dropped) => {}935 None => bail!("host closed upstream MoQ during resolution-stall smoke"),936 }937 }938}939940fn create_resolution_stall_marker(path: &Path, contents: &str) -> Result<()> {941 let mut options = OpenOptions::new();942 options.write(true).create_new(true);943 944 options.mode(0o600);945 let mut marker = options946 .open(path)947 .with_context(|| format!("creating resolution stall marker {}", path.display()))?;948 marker949 .write_all(contents.as_bytes())950 .with_context(|| format!("writing resolution stall marker {}", path.display()))?;951 marker952 .sync_all()953 .with_context(|| format!("syncing resolution stall marker {}", path.display()))?;954 Ok(())955}956957async fn wait_for_resolution_stall_marker(path: &Path, timeout: Duration) -> Result<()> {958 tokio::time::timeout(timeout, async {959 loop {960 match fs::metadata(path) {961 Ok(metadata) => {962 ensure!(963 metadata.is_file(),964 "resolution stall gate marker is not a file: {}",965 path.display()966 );967 return Ok(());968 }969 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {970 tokio::time::sleep(RESOLUTION_STALL_GATE_POLL_INTERVAL).await;971 }972 Err(error) => {973 return Err(error).with_context(|| {974 format!("reading resolution stall gate marker {}", path.display())975 });976 }977 }978 }979 })980 .await981 .with_context(|| {982 format!(983 "timed out waiting for resolution stall marker {}",984 path.display()985 )986 })?987}988989async fn send_resolution_stall_input_probe(990 send: &mut iroh::endpoint::SendStream,991 recv: &mut iroh::endpoint::RecvStream,992 expected_ack: &mut u64,993) -> Result<u64> {994 let started = Instant::now();995 write_input_event(send, &InputEvent::Probe)996 .await997 .context("writing input probe during resolution stall")?;998 *expected_ack = expected_ack.saturating_add(1);999 tokio::time::timeout(1000 RESOLUTION_STALL_RECOVERY_TIMEOUT,1001 read_expected_input_ack(recv, 2, *expected_ack),1002 )1003 .await1004 .context("input acknowledgement exceeded the resolution-stall bound")??;1005 Ok(u64::try_from(started.elapsed().as_micros()).unwrap_or(u64::MAX))1006}100710081009async fn run_resolution_stall_smoke(1010 gate_directory: &Path,1011 receiver: &mut MoqProbeReceiver,1012 feedback_send: &mut iroh::endpoint::SendStream,1013 feedback_recv: &mut iroh::endpoint::RecvStream,1014 input_send: &mut iroh::endpoint::SendStream,1015 input_recv: &mut iroh::endpoint::RecvStream,1016 expected_ack: &mut u64,1017 expected_native: Option<(u16, u16)>,1018 timeout: Duration,1019) -> Result<ResolutionStallEvidence> {1020 let gate = ResolutionStallGate::new(gate_directory)?;1021 let initial = tokio::time::timeout(timeout, next_resolution_stall_moq_frame(receiver))1022 .await1023 .context("timed out waiting for initial resolution-stall media")??;1024 ensure!(1025 initial.flags.contains(FrameFlags::KEYFRAME)1026 && initial.flags.contains(FrameFlags::CODEC_CONFIG),1027 "resolution-stall smoke did not begin at a configured IDR"1028 );1029 let native_dimensions = (initial.width, initial.height);1030 if let Some(expected) = expected_native {1031 ensure!(1032 native_dimensions == expected,1033 "expected {}x{} native media but received {}x{}",1034 expected.0,1035 expected.1,1036 native_dimensions.0,1037 native_dimensions.11038 );1039 }1040 let reduced_dimensions = resolution_stall_reduced_dimensions(native_dimensions)?;10411042 let mut report_id = 1_u64;1043 let seed = complete_feedback_report(report_id, MediaFeedbackFlags::NONE, initial.sequence);1044 let _ = exchange_feedback_report(feedback_send, feedback_recv, &seed).await?;1045 tokio::time::sleep(RESOLUTION_STALL_FEEDBACK_INTERVAL).await;10461047 report_id += 1;1048 let pressure = complete_feedback_report(1049 report_id,1050 MediaFeedbackFlags::RESYNC_ACTIVE,1051 initial.sequence,1052 );1053 let pressure_decision =1054 exchange_feedback_report(feedback_send, feedback_recv, &pressure).await?;1055 ensure!(1056 pressure_decision1057 .reasons1058 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE),1059 "resolution-stall pressure report did not register receiver pressure"1060 );10611062 let reduced = tokio::time::timeout(timeout, async {1063 loop {1064 let frame = next_resolution_stall_moq_frame(receiver).await?;1065 let dimensions = (frame.width, frame.height);1066 ensure!(1067 dimensions == native_dimensions || dimensions == reduced_dimensions,1068 "resolution-stall media changed to unexpected dimensions {}x{}",1069 dimensions.0,1070 dimensions.11071 );1072 if dimensions == reduced_dimensions1073 && frame.flags.contains(FrameFlags::KEYFRAME)1074 && frame.flags.contains(FrameFlags::CODEC_CONFIG)1075 {1076 break Ok::<_, anyhow::Error>(frame);1077 }1078 }1079 })1080 .await1081 .context("timed out waiting for pressure-triggered reduced configured IDR")??;1082 ensure!(1083 reduced.sequence > initial.sequence,1084 "reduced configured IDR did not advance media sequence"1085 );10861087 create_resolution_stall_marker(1088 &gate.ready,1089 &format!(1090 "native={}x{}\nreduced={}x{}\nsequence={}\n",1091 native_dimensions.0,1092 native_dimensions.1,1093 reduced_dimensions.0,1094 reduced_dimensions.1,1095 reduced.sequence1096 ),1097 )?;1098 println!("resolution_stall_ready=1");1099 println!("resolution_stall_ready_sequence={}", reduced.sequence);1100 std::io::stdout()1101 .flush()1102 .context("flushing resolution-stall readiness marker")?;1103 wait_for_resolution_stall_marker(&gate.stalled, timeout).await?;11041105 report_id += 1;1106 let reseed = complete_feedback_report(report_id, MediaFeedbackFlags::NONE, reduced.sequence);1107 let _ = exchange_feedback_report(feedback_send, feedback_recv, &reseed).await?;1108 let mut last_feedback_at = Instant::now();1109 let mut fresh_started = None;1110 let mut stall_input_ack_micros = None;1111 for index in 0..RESOLUTION_STALL_FRESH_REPORTS {1112 tokio::time::sleep_until(tokio::time::Instant::from_std(1113 last_feedback_at + RESOLUTION_STALL_FEEDBACK_INTERVAL,1114 ))1115 .await;1116 report_id += 1;1117 let report =1118 complete_feedback_report(report_id, MediaFeedbackFlags::NONE, reduced.sequence);1119 let sent_at = Instant::now();1120 fresh_started.get_or_insert(sent_at);1121 let decision = exchange_feedback_report(feedback_send, feedback_recv, &report).await?;1122 ensure!(1123 decision.state != AdaptiveBitrateStateV1::Increase,1124 "no-progress feedback unexpectedly increased adaptive bitrate"1125 );1126 ensure!(1127 !decision1128 .reasons1129 .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY),1130 "no-progress feedback was incorrectly classified as clean recovery"1131 );1132 last_feedback_at = sent_at;1133 if index + 1 == RESOLUTION_STALL_FRESH_REPORTS / 2 {1134 stall_input_ack_micros = Some(1135 send_resolution_stall_input_probe(input_send, input_recv, expected_ack).await?,1136 );1137 }1138 }1139 let fresh_duration = fresh_started1140 .context("resolution-stall fresh feedback interval did not start")?1141 .elapsed();1142 ensure!(1143 fresh_duration > Duration::from_secs(8),1144 "resolution-stall fresh feedback lasted only {}ms",1145 fresh_duration.as_millis()1146 );11471148 create_resolution_stall_marker(1149 &gate.resume_ready,1150 &format!(1151 "fresh_reports={RESOLUTION_STALL_FRESH_REPORTS}\nfresh_duration_ms={}\n",1152 fresh_duration.as_millis()1153 ),1154 )?;1155 println!("resolution_stall_resume_ready=1");1156 std::io::stdout()1157 .flush()1158 .context("flushing resolution-stall resume marker")?;1159 wait_for_resolution_stall_marker(&gate.resumed, timeout).await?;11601161 let resume_input_ack_micros =1162 send_resolution_stall_input_probe(input_send, input_recv, expected_ack).await?;1163 let media_resumed_at = Instant::now();1164 let resumed = tokio::time::timeout(1165 RESOLUTION_STALL_RECOVERY_TIMEOUT,1166 next_resolution_stall_moq_frame(receiver),1167 )1168 .await1169 .context("media did not resume within the resolution-stall bound")??;1170 ensure!(1171 (resumed.width, resumed.height) == reduced_dimensions,1172 "media resumed at {}x{} instead of reduced {}x{}",1173 resumed.width,1174 resumed.height,1175 reduced_dimensions.0,1176 reduced_dimensions.11177 );1178 let resume_sequence_advance = resumed1179 .sequence1180 .checked_sub(reduced.sequence)1181 .context("resumed media sequence did not advance")?;1182 ensure!(1183 resume_sequence_advance > 0,1184 "resumed media sequence did not advance"1185 );1186 let resume_media_micros =1187 u64::try_from(media_resumed_at.elapsed().as_micros()).unwrap_or(u64::MAX);11881189 Ok(ResolutionStallEvidence {1190 native_dimensions,1191 reduced_dimensions,1192 stall_sequence: reduced.sequence,1193 fresh_duration,1194 stall_input_ack_micros: stall_input_ack_micros1195 .context("resolution-stall input acknowledgement was not measured")?,1196 resume_input_ack_micros,1197 resume_media_micros,1198 resume_sequence_advance,1199 })1200}120112021203async fn main() -> Result<()> {1204 let args = Args::parse();1205 ensure!(1206 args.timeout_seconds > 0,1207 "--timeout-seconds must be greater than zero"1208 );1209 ensure!(1210 !args.keyframe_smoke || args.frames >= KEYFRAME_SMOKE_MINIMUM_FRAMES,1211 "--keyframe-smoke requires at least {KEYFRAME_SMOKE_MINIMUM_FRAMES} accepted frames"1212 );1213 ensure!(1214 args.slow_consumer_ms.is_none() || args.frames >= SLOW_CONSUMER_MINIMUM_FRAMES,1215 "--slow-consumer-ms requires at least {SLOW_CONSUMER_MINIMUM_FRAMES} accepted frames"1216 );12171218 let secret = match args.identity.as_deref() {1219 Some(path) => identity::load(path).context("loading persistent probe identity")?,1220 None => SecretKey::generate(),1221 };1222 let mut required_grants = InvitationGrants::VIEW;1223 if args.pointer_smoke {1224 required_grants = required_grants.union(InvitationGrants::POINTER_KEYBOARD);1225 }1226 if args.gamepad_smoke {1227 required_grants = required_grants.union(InvitationGrants::GAMEPAD);1228 }1229 let invitation = args1230 .invitation1231 .as_deref()1232 .map(|path| load_invitation_file(path, args.node_id, secret.public(), required_grants))1233 .transpose()?;1234 let mut nonce = [0_u8; 16];1235 getrandom::fill(&mut nonce).context("generating handshake nonce")?;1236 let mut endpoint_builder = Endpoint::builder(presets::N0).secret_key(secret);1237 if args.relay_only {1238 endpoint_builder = endpoint_builder.clear_ip_transports();1239 }1240 let endpoint = endpoint_builder1241 .bind()1242 .await1243 .context("binding probe endpoint")?;1244 let _ = tokio::time::timeout(Duration::from_secs(10), endpoint.online()).await;1245 let address = EndpointAddr::new(args.node_id);1246 let media_transport = if args.media_v3 {1247 MediaTransport::GroupedV31248 } else {1249 MediaTransport::UpstreamMoq1250 };12511252 let media_connection = endpoint1253 .connect(address.clone(), media_transport.alpn())1254 .await1255 .context("connecting media protocol")?;1256 let (mut media_send, mut media_recv) = media_connection1257 .open_bi()1258 .await1259 .context("opening media stream")?;1260 let handshake_name = if media_transport == MediaTransport::UpstreamMoq {1261 "control"1262 } else {1263 "media"1264 };1265 let media_negotiation = negotiate(1266 &mut media_send,1267 &mut media_recv,1268 nonce,1269 vec![Capability::VideoH264],1270 Capability::VideoH264,1271 handshake_name,1272 invitation.as_deref(),1273 )1274 .await?;1275 let session_id = media_negotiation.session_id;1276 // Sigil finishes its response half after HostHello. Keep our send half1277 // alive as the bounded keyframe-control stream.1278 drop(media_recv);12791280 let mut moq_receiver = if media_transport == MediaTransport::UpstreamMoq {1281 Some(1282 MoqProbeReceiver::connect(1283 &endpoint,1284 address.clone(),1285 session_id,1286 Duration::from_secs(args.timeout_seconds),1287 )1288 .await?,1289 )1290 } else {1291 None1292 };12931294 let input_connection = endpoint1295 .connect(address.clone(), INPUT_ALPN_V1)1296 .await1297 .context("connecting input protocol")?;1298 let (mut input_send, mut input_recv) = input_connection1299 .open_bi()1300 .await1301 .context("opening input stream")?;1302 let mut input_offers = vec![Capability::InputAck];1303 if args.pointer_smoke {1304 input_offers.push(Capability::RelativePointer);1305 }1306 if args.pointer_feedback_smoke {1307 input_offers.push(Capability::PointerPositionFeedback);1308 input_offers.push(Capability::PointerVisibilityFeedback);1309 }1310 if args.gamepad_smoke {1311 input_offers.push(Capability::Gamepad);1312 }1313 let input_negotiation = negotiate(1314 &mut input_send,1315 &mut input_recv,1316 nonce,1317 input_offers,1318 Capability::InputAck,1319 "input",1320 None,1321 )1322 .await?;1323 ensure!(1324 session_id == input_negotiation.session_id,1325 "media and input session IDs differ"1326 );1327 if args.pointer_feedback_smoke {1328 ensure!(1329 input_negotiation1330 .capabilities1331 .contains(&Capability::PointerPositionFeedback)1332 && input_negotiation1333 .capabilities1334 .contains(&Capability::PointerVisibilityFeedback),1335 "host did not accept the required pointer feedback capabilities"1336 );1337 let initial_feedback =1338 read_expected_input_ack(&mut input_recv, args.timeout_seconds, 0).await?;1339 ensure!(1340 initial_feedback.pointer_position.is_some(),1341 "host pointer feedback is unavailable"1342 );1343 ensure!(1344 initial_feedback.pointer_visible.is_some(),1345 "host omitted negotiated pointer visibility feedback"1346 );1347 }1348 let input_started = Instant::now();1349 write_input_event(&mut input_send, &InputEvent::Probe)1350 .await1351 .context("writing input probe")?;1352 read_expected_input_ack(&mut input_recv, args.timeout_seconds, 1).await?;1353 let input_ack_micros = input_started.elapsed().as_micros();1354 let mut expected_ack = 1_u64;1355 let mut pointer_sync_position = None;1356 let mut pointer_motion_position = None;13571358 if args.pointer_smoke {1359 ensure!(1360 input_negotiation1361 .capabilities1362 .contains(&Capability::RelativePointer),1363 "host did not accept the required relative pointer capability"1364 );1365 let pointer_dimensions = media_negotiation.pointer_surface_dimensions.context(1366 "host did not advertise pointer surface dimensions required by --pointer-smoke",1367 )?;1368 let [position_sync, relative_motion, click] =1369 pointer_smoke_events(Some(pointer_dimensions))?;1370 let [expected_sync_position, expected_motion_position] =1371 pointer_smoke_expected_positions(pointer_dimensions)?;1372 write_input_event(&mut input_send, &position_sync)1373 .await1374 .context("writing pointer position synchronization")?;1375 expected_ack += 1;1376 let sync_ack =1377 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;1378 if args.pointer_feedback_smoke {1379 pointer_sync_position = Some(1380 confirm_pointer_position(1381 &mut input_send,1382 &mut input_recv,1383 &mut expected_ack,1384 sync_ack,1385 expected_sync_position,1386 Duration::from_secs(args.timeout_seconds.min(5)),1387 )1388 .await1389 .context("verifying compositor position after pointer synchronization")?,1390 );1391 }1392 write_input_event(&mut input_send, &relative_motion)1393 .await1394 .context("writing relative pointer smoke motion")?;1395 expected_ack += 1;1396 let motion_ack =1397 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;1398 if args.pointer_feedback_smoke {1399 pointer_motion_position = Some(1400 confirm_pointer_position(1401 &mut input_send,1402 &mut input_recv,1403 &mut expected_ack,1404 motion_ack,1405 expected_motion_position,1406 Duration::from_secs(args.timeout_seconds.min(5)),1407 )1408 .await1409 .context("verifying compositor position after relative pointer motion")?,1410 );1411 }1412 write_input_event(&mut input_send, &click)1413 .await1414 .context("writing pointer smoke click")?;1415 expected_ack += 1;1416 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;1417 }14181419 if args.gamepad_smoke {1420 ensure!(1421 input_negotiation1422 .capabilities1423 .contains(&Capability::Gamepad),1424 "host did not accept the required gamepad capability"1425 );1426 write_input_event(1427 &mut input_send,1428 &InputEvent::Gamepad {1429 state: GamepadState {1430 a: true,1431 right_shoulder: true,1432 dpad_right: true,1433 left_x: GAMEPAD_AXIS_MAX,1434 right_y: GAMEPAD_AXIS_MIN,1435 left_trigger: GAMEPAD_TRIGGER_MAX,1436 right_trigger: GAMEPAD_TRIGGER_MAX,1437 ..GamepadState::default()1438 },1439 },1440 )1441 .await1442 .context("writing non-neutral gamepad smoke snapshot")?;1443 expected_ack += 1;1444 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;1445 tokio::time::sleep(Duration::from_millis(100)).await;1446 write_input_event(1447 &mut input_send,1448 &InputEvent::Gamepad {1449 state: GamepadState::default(),1450 },1451 )1452 .await1453 .context("writing neutral gamepad smoke snapshot")?;1454 expected_ack += 1;1455 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack).await?;1456 }14571458 if let Some(gate_directory) = args.resolution_stall_smoke.as_deref() {1459 let feedback_connection = endpoint1460 .connect(address.clone(), MEDIA_FEEDBACK_ALPN_V1)1461 .await1462 .context("connecting media feedback protocol")?;1463 let (mut feedback_send, mut feedback_recv) = feedback_connection1464 .open_bi()1465 .await1466 .context("opening media feedback stream")?;1467 let feedback_negotiation = negotiate(1468 &mut feedback_send,1469 &mut feedback_recv,1470 nonce,1471 vec![Capability::VideoH264],1472 Capability::VideoH264,1473 "media feedback",1474 None,1475 )1476 .await?;1477 ensure!(1478 session_id == feedback_negotiation.session_id,1479 "media and feedback session IDs differ"1480 );14811482 let evidence = run_resolution_stall_smoke(1483 gate_directory,1484 moq_receiver1485 .as_mut()1486 .expect("resolution-stall smoke requires upstream MoQ"),1487 &mut feedback_send,1488 &mut feedback_recv,1489 &mut input_send,1490 &mut input_recv,1491 &mut expected_ack,1492 args.expect_size,1493 Duration::from_secs(args.timeout_seconds),1494 )1495 .await?;14961497 feedback_send1498 .finish()1499 .context("finishing media feedback stream")?;1500 input_send.finish().context("finishing input stream")?;1501 media_send1502 .finish()1503 .context("finishing media request stream")?;1504 feedback_connection.close(0_u32.into(), b"probe complete");1505 input_connection.close(0_u32.into(), b"probe complete");1506 moq_receiver1507 .as_ref()1508 .expect("resolution-stall smoke requires upstream MoQ")1509 .connection()1510 .close(0_u32.into(), b"probe complete");1511 media_connection.close(0_u32.into(), b"probe complete");1512 tokio::time::sleep(Duration::from_millis(50)).await;1513 endpoint.close().await;15141515 println!("probe=ok");1516 println!("session_id={session_id}");1517 println!("resolution_stall_smoke=ok");1518 println!("resolution_stall_session_id={session_id}");1519 println!(1520 "resolution_stall_native_dimensions={}x{}",1521 evidence.native_dimensions.0, evidence.native_dimensions.11522 );1523 println!(1524 "resolution_stall_reduced_dimensions={}x{}",1525 evidence.reduced_dimensions.0, evidence.reduced_dimensions.11526 );1527 println!("resolution_stall_sequence={}", evidence.stall_sequence);1528 println!("resolution_stall_fresh_reports={RESOLUTION_STALL_FRESH_REPORTS}");1529 println!(1530 "resolution_stall_fresh_duration_ms={}",1531 evidence.fresh_duration.as_millis()1532 );1533 println!(1534 "resolution_stall_input_ack_micros={}",1535 evidence.stall_input_ack_micros1536 );1537 println!(1538 "resolution_resume_input_ack_micros={}",1539 evidence.resume_input_ack_micros1540 );1541 println!(1542 "resolution_resume_media_micros={}",1543 evidence.resume_media_micros1544 );1545 println!(1546 "resolution_resume_sequence_advance={}",1547 evidence.resume_sequence_advance1548 );1549 return Ok(());1550 }15511552 let started = Instant::now();1553 let mut received = 0_u32;1554 let mut bytes = 0_u64;1555 let mut keyframes = 0_u32;1556 let mut gaps = 0_u64;1557 let mut media_objects_dropped = 0_u64;1558 let mut media_objects_late = 0_u64;1559 let mut last_sequence: Option<u64> = None;1560 let mut dimensions = None;1561 let mut first_accepted_at = None;1562 let mut last_accepted_at = None;1563 let mut object_receiver_v3 = (media_transport == MediaTransport::GroupedV3).then(|| {1564 MediaObjectReceiverV3::new(1565 media_connection.clone(),1566 Duration::from_secs(args.timeout_seconds),1567 )1568 });1569 let mut object_sequence_v3 = MediaObjectSequenceV3::new();1570 let mut keyframe_request_sent = false;1571 let mut keyframe_recovery_verified = false;1572 let mut keyframe_request_sent_at: Option<Instant> = None;1573 let mut keyframe_recovery_micros = None;1574 let mut keyframe_request_group_id = None;1575 let mut keyframe_request_last_sequence = None;1576 let mut slow_consumer_stalled = false;1577 let mut slow_consumer_pending_recovery = false;1578 let mut slow_consumer_verified = false;1579 let mut slow_consumer_group_id = None;1580 let mut slow_consumer_last_sequence = None;1581 let mut slow_consumer_capture_timestamp_us = None;1582 let mut slow_consumer_cancelled_groups_before = None;1583 let mut slow_consumer_resumed_at: Option<Instant> = None;1584 let mut slow_consumer_recovery_micros = None;1585 let mut slow_consumer_cancellation_delta = None;1586 let mut slow_consumer_group_advance = None;1587 let mut slow_consumer_sequence_advance = None;1588 let mut slow_consumer_capture_advance_micros = None;1589 let mut slow_consumer_input_ack_micros = None;15901591 while received < args.frames {1592 let (frame, recovery_frame): (AcceptedMedia, bool) = match media_transport {1593 MediaTransport::UpstreamMoq => loop {1594 let media_wait = if slow_consumer_pending_recovery {1595 SLOW_CONSUMER_RECOVERY_TIMEOUT1596 } else {1597 Duration::from_secs(args.timeout_seconds).saturating_add(Duration::from_secs(1))1598 };1599 let outcome = tokio::time::timeout(1600 media_wait,1601 moq_receiver1602 .as_mut()1603 .expect("upstream MoQ receiver is present")1604 .next(),1605 )1606 .await1607 .context("timed out waiting for an upstream MoQ media object")??1608 .context("host closed the upstream MoQ video track")?;1609 match outcome {1610 MoqProbeOutcome::Dropped => {1611 media_objects_dropped = media_objects_dropped.saturating_add(1);1612 }1613 MoqProbeOutcome::Frame {1614 frame,1615 group_sequence,1616 recovery,1617 } => {1618 let requested_keyframe_recovery =1619 keyframe_request_sent && !keyframe_recovery_verified;1620 if requested_keyframe_recovery {1621 ensure!(1622 recovery1623 && frame.header.flags.contains(FrameFlags::KEYFRAME)1624 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),1625 "MoQ keyframe request recovery did not begin at a configured IDR barrier"1626 );1627 ensure!(1628 keyframe_request_group_id1629 .is_some_and(|prior| group_sequence > prior),1630 "MoQ keyframe request recovery did not advance the native group"1631 );1632 ensure!(1633 keyframe_request_last_sequence1634 .is_some_and(|prior| frame.header.sequence > prior),1635 "MoQ keyframe request recovery did not advance media sequence"1636 );1637 keyframe_recovery_micros = keyframe_request_sent_at.map(|sent_at| {1638 u64::try_from(sent_at.elapsed().as_micros()).unwrap_or(u64::MAX)1639 });1640 keyframe_recovery_verified = true;1641 }1642 let slow_consumer_recovery = slow_consumer_pending_recovery;1643 if slow_consumer_recovery {1644 ensure!(1645 recovery1646 && frame.header.flags.contains(FrameFlags::KEYFRAME)1647 && frame.header.flags.contains(FrameFlags::CODEC_CONFIG),1648 "MoQ slow-consumer recovery did not begin at a configured IDR barrier"1649 );1650 let prior_group = slow_consumer_group_id1651 .context("slow-consumer group was not recorded")?;1652 let group_advance = group_sequence.checked_sub(prior_group).context(1653 "MoQ slow-consumer recovery did not advance the native group",1654 )?;1655 ensure!(1656 group_advance > 0,1657 "MoQ slow-consumer recovery did not advance the native group"1658 );1659 let prior_sequence = slow_consumer_last_sequence1660 .context("slow-consumer sequence was not recorded")?;1661 let sequence_advance =1662 frame.header.sequence.checked_sub(prior_sequence).context(1663 "MoQ slow-consumer recovery did not advance media sequence",1664 )?;1665 ensure!(1666 sequence_advance > 0,1667 "MoQ slow-consumer recovery did not advance media sequence"1668 );1669 let prior_capture_timestamp = slow_consumer_capture_timestamp_us1670 .context("slow-consumer capture timestamp was not recorded")?;1671 let capture_advance = frame1672 .header1673 .capture_timestamp_us1674 .checked_sub(prior_capture_timestamp)1675 .context(1676 "MoQ slow-consumer recovery capture timestamp moved backward",1677 )?;1678 let minimum_capture_advance = args1679 .slow_consumer_ms1680 .expect("slow-consumer duration is configured")1681 .saturating_mul(1_000)1682 / 2;1683 ensure!(1684 capture_advance >= minimum_capture_advance,1685 "MoQ slow-consumer recovery advanced capture time by only {capture_advance}us; required at least {minimum_capture_advance}us"1686 );1687 let cancelled_before = slow_consumer_cancelled_groups_before1688 .context("slow-consumer cancellation count was not recorded")?;1689 let cancelled_after = moq_receiver1690 .as_ref()1691 .expect("upstream MoQ receiver is present")1692 .cancelled_groups;1693 let cancellation_delta = cancelled_after1694 .checked_sub(cancelled_before)1695 .context("MoQ slow-consumer cancellation count moved backward")?;1696 ensure!(1697 cancellation_delta > 0,1698 "MoQ slow-consumer recovery did not observe a cancelled stale group"1699 );1700 ensure!(1701 moq_receiver1702 .as_ref()1703 .expect("upstream MoQ receiver is present")1704 .historical_suffix_frames1705 == 0,1706 "MoQ slow-consumer recovery accepted historical delta suffix frames"1707 );1708 let recovery_micros = slow_consumer_resumed_at1709 .context("slow-consumer resume instant was not recorded")?1710 .elapsed()1711 .as_micros();1712 ensure!(1713 recovery_micros <= SLOW_CONSUMER_RECOVERY_TIMEOUT.as_micros(),1714 "MoQ slow-consumer recovery exceeded {}us",1715 SLOW_CONSUMER_RECOVERY_TIMEOUT.as_micros()1716 );1717 slow_consumer_recovery_micros =1718 Some(u64::try_from(recovery_micros).unwrap_or(u64::MAX));1719 slow_consumer_cancellation_delta = Some(cancellation_delta);1720 slow_consumer_group_advance = Some(group_advance);1721 slow_consumer_sequence_advance = Some(sequence_advance);1722 slow_consumer_capture_advance_micros = Some(capture_advance);1723 slow_consumer_pending_recovery = false;1724 slow_consumer_verified = true;1725 }1726 break (1727 AcceptedMedia {1728 flags: frame.header.flags,1729 width: frame.header.width,1730 height: frame.header.height,1731 sequence: frame.header.sequence,1732 capture_timestamp_us: frame.header.capture_timestamp_us,1733 payload_len: frame.payload.len(),1734 v3_group_id: Some(group_sequence),1735 },1736 recovery,1737 );1738 }1739 }1740 },1741 MediaTransport::GroupedV3 => loop {1742 let outcome = tokio::time::timeout(1743 Duration::from_secs(args.timeout_seconds)1744 .saturating_add(Duration::from_secs(1)),1745 object_receiver_v31746 .as_mut()1747 .expect("v3 media object receiver is present")1748 .next(),1749 )1750 .await1751 .context("timed out waiting for media v3 object")??1752 .context("host closed the media v3 object connection")?;1753 match outcome {1754 MediaObjectOutcomeV3::Dropped { .. } => {1755 if object_sequence_v3.note_drop() {1756 media_objects_dropped = media_objects_dropped.saturating_add(1);1757 } else {1758 media_objects_late = media_objects_late.saturating_add(1);1759 }1760 }1761 MediaObjectOutcomeV3::Malformed {1762 accept_index,1763 error,1764 } => {1765 bail!("media v3 object {accept_index} is malformed: {error}");1766 }1767 MediaObjectOutcomeV3::Object { object, .. } => {1768 match object_sequence_v3.classify(&object) {1769 MediaObjectDecisionV3::Deliver { discontinuity } => {1770 let recovery = keyframe_request_sent && !keyframe_recovery_verified;1771 if recovery {1772 ensure!(1773 discontinuity1774 && object.header.object_id == 01775 && object.header.flags.contains(FrameFlags::KEYFRAME)1776 && object1777 .header1778 .flags1779 .contains(FrameFlags::CODEC_CONFIG)1780 && object1781 .header1782 .flags1783 .contains(FrameFlags::DISCONTINUITY),1784 "keyframe request recovery did not begin at a configured discontinuity object zero"1785 );1786 ensure!(1787 keyframe_request_group_id1788 .is_some_and(1789 |group_id| object.header.group_id > group_id1790 ),1791 "keyframe request recovery did not advance the media group"1792 );1793 ensure!(1794 keyframe_request_last_sequence.is_some_and(|sequence| {1795 object.header.sequence > sequence1796 }),1797 "keyframe request recovery did not advance media sequence"1798 );1799 keyframe_recovery_micros =1800 keyframe_request_sent_at.map(|sent_at| {1801 u64::try_from(sent_at.elapsed().as_micros())1802 .unwrap_or(u64::MAX)1803 });1804 keyframe_recovery_verified = true;1805 }1806 break (object.into(), recovery);1807 }1808 MediaObjectDecisionV3::DropLate => {1809 media_objects_late = media_objects_late.saturating_add(1);1810 }1811 MediaObjectDecisionV3::DropUntilKeyframe => {1812 media_objects_dropped = media_objects_dropped.saturating_add(1);1813 }1814 }1815 }1816 }1817 },1818 };18191820 let accepted_at = Instant::now();1821 first_accepted_at.get_or_insert(accepted_at);1822 last_accepted_at = Some(accepted_at);18231824 if received == 0 {1825 ensure!(1826 frame.flags.contains(FrameFlags::KEYFRAME)1827 && frame.flags.contains(FrameFlags::CODEC_CONFIG),1828 "first media frame is not a decodable keyframe with codec configuration"1829 );1830 }18311832 let frame_dimensions = (frame.width, frame.height);1833 if let Some(expected) = dimensions {1834 ensure!(1835 frame_dimensions == expected,1836 "media dimensions changed during probe"1837 );1838 } else {1839 dimensions = Some(frame_dimensions);1840 }1841 if let Some(previous) = last_sequence1842 && !recovery_frame1843 {1844 gaps += sequence_gap(previous, frame.sequence)?;1845 }1846 last_sequence = Some(frame.sequence);1847 keyframes += u32::from(frame.flags.contains(FrameFlags::KEYFRAME));1848 bytes = bytes.saturating_add(frame.payload_len as u64);1849 received += 1;18501851 if let Some(stall_ms) = args.slow_consumer_ms1852 && !slow_consumer_stalled1853 {1854 slow_consumer_group_id = frame.v3_group_id;1855 slow_consumer_last_sequence = Some(frame.sequence);1856 slow_consumer_capture_timestamp_us = Some(frame.capture_timestamp_us);1857 slow_consumer_cancelled_groups_before = Some(1858 moq_receiver1859 .as_ref()1860 .expect("upstream MoQ receiver is present")1861 .cancelled_groups,1862 );1863 let first_half = Duration::from_millis(stall_ms / 2);1864 let second_half = Duration::from_millis(stall_ms).saturating_sub(first_half);1865 tokio::time::sleep(first_half).await;1866 let input_probe_started = Instant::now();1867 write_input_event(&mut input_send, &InputEvent::Probe)1868 .await1869 .context("writing input probe during slow-consumer media stall")?;1870 expected_ack = expected_ack.saturating_add(1);1871 tokio::time::timeout(1872 SLOW_CONSUMER_RECOVERY_TIMEOUT,1873 read_expected_input_ack(&mut input_recv, args.timeout_seconds, expected_ack),1874 )1875 .await1876 .context("input acknowledgement was blocked by the slow media consumer")??;1877 slow_consumer_input_ack_micros =1878 Some(u64::try_from(input_probe_started.elapsed().as_micros()).unwrap_or(u64::MAX));1879 tokio::time::sleep(second_half).await;1880 slow_consumer_resumed_at = Some(Instant::now());1881 slow_consumer_pending_recovery = true;1882 slow_consumer_stalled = true;1883 }18841885 if args.keyframe_smoke1886 && !keyframe_request_sent1887 && received == KEYFRAME_SMOKE_REQUEST_AFTER_FRAMES1888 {1889 let request = MediaControlRequestV3::request_keyframe(1890 args.keyframe_request_id,1891 Some(frame.sequence),1892 KeyframeRequestReasonV3::DecoderReset,1893 );1894 tokio::time::timeout(1895 Duration::from_secs(1),1896 write_media_control_request_v3(&mut media_send, &request),1897 )1898 .await1899 .context("timed out writing v3 keyframe request")??;1900 keyframe_request_group_id = frame.v3_group_id;1901 keyframe_request_last_sequence = Some(frame.sequence);1902 keyframe_request_sent_at = Some(Instant::now());1903 keyframe_request_sent = true;1904 match media_transport {1905 MediaTransport::UpstreamMoq => {1906 let cancelled_group = moq_receiver1907 .as_mut()1908 .expect("upstream MoQ receiver is present")1909 .request_recovery();1910 ensure!(1911 cancelled_group == keyframe_request_group_id,1912 "MoQ recovery did not cancel the currently accepted native group"1913 );1914 }1915 MediaTransport::GroupedV3 => object_sequence_v3.request_recovery(),1916 }1917 }1918 }19191920 let Some((width, height)) = dimensions else {1921 bail!("probe received no frames");1922 };1923 if let Some((expected_width, expected_height)) = args.expect_size {1924 ensure!(1925 (width, height) == (expected_width, expected_height),1926 "expected {expected_width}x{expected_height} but received {width}x{height}"1927 );1928 }1929 ensure!(keyframes > 0, "probe received no H.264 keyframe");1930 ensure!(gaps == 0, "probe observed {gaps} media sequence gaps");1931 ensure!(1932 !args.keyframe_smoke || keyframe_recovery_verified,1933 "probe did not verify requested keyframe recovery"1934 );1935 ensure!(1936 args.slow_consumer_ms.is_none() || slow_consumer_verified,1937 "probe did not verify bounded slow-consumer recovery"1938 );1939 let accepted_span = first_accepted_at1940 .zip(last_accepted_at)1941 .map_or(Duration::ZERO, |(first, last)| {1942 last.saturating_duration_since(first)1943 });1944 let accepted_fps = accepted_frame_rate(received, accepted_span);1945 if let Some(minimum_fps) = args.minimum_fps {1946 let accepted_fps = accepted_fps.context(1947 "--minimum-fps requires at least two accepted frames with measurable elapsed time",1948 )?;1949 ensure!(1950 accepted_fps >= minimum_fps,1951 "probe sustained only {accepted_fps:.3} accepted fps; required at least {minimum_fps:.3} fps"1952 );1953 }1954 let feedback_evidence = if args.feedback_smoke {1955 Some(1956 run_feedback_smoke(1957 &endpoint,1958 address.clone(),1959 nonce,1960 session_id,1961 last_sequence.context("feedback smoke requires an accepted media sequence")?,1962 )1963 .await?,1964 )1965 } else {1966 None1967 };1968 let media_diagnostics_connection = moq_receiver1969 .as_ref()1970 .map_or(&media_connection, MoqProbeReceiver::connection);1971 let (media_path_mode, media_path_rtt_ms) =1972 selected_path_diagnostics(media_diagnostics_connection);1973 let (control_path_mode, control_path_rtt_ms) = if media_transport == MediaTransport::UpstreamMoq1974 {1975 selected_path_diagnostics(&media_connection)1976 } else {1977 ("not-used", None)1978 };1979 let (input_path_mode, input_path_rtt_ms) = selected_path_diagnostics(&input_connection);1980 let feedback_path_mode = feedback_evidence1981 .as_ref()1982 .map_or("not-used", |(_, path_mode, _)| *path_mode);1983 if args.relay_only {1984 ensure!(1985 control_path_mode == "relay"1986 && media_path_mode == "relay"1987 && input_path_mode == "relay"1988 && (feedback_evidence.is_none() || feedback_path_mode == "relay"),1989 "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}"1990 );1991 }1992 let (1993 moq_cancelled_groups,1994 moq_group_gaps,1995 moq_maximum_group_objects,1996 moq_maximum_group_bytes,1997 moq_historical_suffix_frames,1998 ) = moq_receiver.as_ref().map_or((0, 0, 0, 0, 0), |receiver| {1999 (2000 receiver.cancelled_groups,2001 receiver.group_gaps,2002 receiver.maximum_group_objects,2003 receiver.maximum_group_bytes,2004 receiver.historical_suffix_frames,2005 )2006 });2007 input_send.finish().context("finishing input stream")?;2008 media_send2009 .finish()2010 .context("finishing media request stream")?;2011 // Close both protocol connections explicitly. Relying only on endpoint2012 // teardown can leave a peer waiting for QUIC idle timeout during repeated2013 // short-lived probes, which obscures whether the host released its2014 // one-client lease deterministically.2015 input_connection.close(0_u32.into(), b"probe complete");2016 if let Some(receiver) = moq_receiver.as_ref() {2017 receiver.connection().close(0_u32.into(), b"probe complete");2018 }2019 media_connection.close(0_u32.into(), b"probe complete");2020 tokio::time::sleep(Duration::from_millis(50)).await;2021 endpoint.close().await;20222023 println!("probe=ok");2024 println!("session_id={session_id}");2025 println!("frames={received}");2026 println!("dimensions={width}x{height}");2027 println!("keyframes={keyframes}");2028 println!("sequence_gaps={gaps}");2029 println!("media_objects_dropped={media_objects_dropped}");2030 println!("media_objects_late={media_objects_late}");2031 println!("transport={}", media_transport.label());2032 println!(2033 "control_alpn={}",2034 if media_transport == MediaTransport::UpstreamMoq {2035 "sigil/control/1"2036 } else {2037 "not-used"2038 }2039 );2040 println!("transport_alpn={}", media_transport.media_alpn_label());2041 println!("first_configured_idr=ok");2042 println!("frame_sequence=monotonic");2043 println!(2044 "group_sequence={}",2045 if media_transport == MediaTransport::UpstreamMoq {2046 "monotonic"2047 } else {2048 "not-applicable"2049 }2050 );2051 if media_transport == MediaTransport::UpstreamMoq {2052 println!(2053 "moq_catalog={}",2054 moq_receiver2055 .as_ref()2056 .expect("upstream MoQ receiver is present")2057 .catalog_mode2058 .label()2059 );2060 println!("moq_group_capacity=1");2061 println!("moq_cancelled_groups={moq_cancelled_groups}");2062 println!("moq_group_gaps={moq_group_gaps}");2063 println!("moq_unrecovered_group_gaps=0");2064 println!("moq_maximum_group_objects={moq_maximum_group_objects}");2065 println!("moq_maximum_group_bytes={moq_maximum_group_bytes}");2066 println!("moq_historical_suffix_frames={moq_historical_suffix_frames}");2067 } else {2068 println!("moq_catalog=not-applicable");2069 for field in [2070 "moq_group_capacity",2071 "moq_cancelled_groups",2072 "moq_group_gaps",2073 "moq_unrecovered_group_gaps",2074 "moq_maximum_group_objects",2075 "moq_maximum_group_bytes",2076 "moq_historical_suffix_frames",2077 ] {2078 println!("{field}=not-applicable");2079 }2080 }2081 println!(2082 "recovery_barrier={}",2083 if args.keyframe_smoke {2084 "configured-idr"2085 } else {2086 "not-requested"2087 }2088 );2089 println!(2090 "keyframe_recovery={}",2091 if args.keyframe_smoke {2092 "ok"2093 } else {2094 "not-requested"2095 }2096 );2097 if args.keyframe_smoke {2098 println!("keyframe_request_id={}", args.keyframe_request_id);2099 println!(2100 "keyframe_recovery_micros={}",2101 keyframe_recovery_micros.context("keyframe recovery latency was not measured")?2102 );2103 } else {2104 println!("keyframe_request_id=not-requested");2105 println!("keyframe_recovery_micros=not-requested");2106 }2107 if let Some(stall_ms) = args.slow_consumer_ms {2108 println!("slow_consumer=ok");2109 println!("slow_consumer_stall_ms={stall_ms}");2110 println!("slow_consumer_first_post_stall=configured-idr");2111 println!("slow_consumer_historical_suffix_frames=0");2112 println!(2113 "slow_consumer_recovery_micros={}",2114 slow_consumer_recovery_micros2115 .context("slow-consumer recovery latency was not measured")?2116 );2117 println!(2118 "slow_consumer_cancellation_delta={}",2119 slow_consumer_cancellation_delta2120 .context("slow-consumer cancellation delta was not measured")?2121 );2122 println!(2123 "slow_consumer_group_advance={}",2124 slow_consumer_group_advance.context("slow-consumer group advance was not measured")?2125 );2126 println!(2127 "slow_consumer_sequence_advance={}",2128 slow_consumer_sequence_advance2129 .context("slow-consumer sequence advance was not measured")?2130 );2131 println!(2132 "slow_consumer_capture_advance_micros={}",2133 slow_consumer_capture_advance_micros2134 .context("slow-consumer capture advance was not measured")?2135 );2136 println!(2137 "slow_consumer_input_ack_micros={}",2138 slow_consumer_input_ack_micros2139 .context("slow-consumer input acknowledgement was not measured")?2140 );2141 } else {2142 for field in [2143 "slow_consumer",2144 "slow_consumer_stall_ms",2145 "slow_consumer_first_post_stall",2146 "slow_consumer_historical_suffix_frames",2147 "slow_consumer_recovery_micros",2148 "slow_consumer_cancellation_delta",2149 "slow_consumer_group_advance",2150 "slow_consumer_sequence_advance",2151 "slow_consumer_capture_advance_micros",2152 "slow_consumer_input_ack_micros",2153 ] {2154 println!("{field}=not-requested");2155 }2156 }2157 match accepted_fps {2158 Some(fps) => println!("accepted_fps={fps:.3}"),2159 None => println!("accepted_fps=unknown"),2160 }2161 println!("input_ack_micros={input_ack_micros}");2162 println!(2163 "pointer_smoke={}",2164 if args.pointer_smoke {2165 "ok"2166 } else {2167 "not-requested"2168 }2169 );2170 println!(2171 "pointer_feedback_smoke={}",2172 if args.pointer_feedback_smoke {2173 "ok"2174 } else {2175 "not-requested"2176 }2177 );2178 match pointer_sync_position {2179 Some(position) => println!("pointer_sync_position={},{}", position.x, position.y),2180 None => println!("pointer_sync_position=not-requested"),2181 }2182 match pointer_motion_position {2183 Some(position) => println!("pointer_motion_position={},{}", position.x, position.y),2184 None => println!("pointer_motion_position=not-requested"),2185 }2186 println!(2187 "gamepad_smoke={}",2188 if args.gamepad_smoke {2189 "ok"2190 } else {2191 "not-requested"2192 }2193 );2194 if let Some((decision, path_mode, path_rtt_ms)) = &feedback_evidence {2195 println!("feedback_smoke=ok");2196 println!("feedback_report_id={}", decision.report_id);2197 println!("feedback_state={:?}", decision.state);2198 println!("feedback_target_kbps={}", decision.target_kbps);2199 println!("feedback_floor_kbps={}", decision.floor_kbps);2200 println!("feedback_ceiling_kbps={}", decision.ceiling_kbps);2201 println!("feedback_applied={}", decision.applied);2202 println!("feedback_path_mode={path_mode}");2203 match path_rtt_ms {2204 Some(rtt) => println!("feedback_path_rtt_ms={rtt:.3}"),2205 None => println!("feedback_path_rtt_ms=unknown"),2206 }2207 } else {2208 for field in [2209 "feedback_smoke",2210 "feedback_report_id",2211 "feedback_state",2212 "feedback_target_kbps",2213 "feedback_floor_kbps",2214 "feedback_ceiling_kbps",2215 "feedback_applied",2216 "feedback_path_mode",2217 "feedback_path_rtt_ms",2218 ] {2219 println!("{field}=not-requested");2220 }2221 }2222 println!("path_mode={media_path_mode}");2223 match media_path_rtt_ms {2224 Some(rtt) => println!("path_rtt_ms={rtt:.3}"),2225 None => println!("path_rtt_ms=unknown"),2226 }2227 println!("control_path_mode={control_path_mode}");2228 match control_path_rtt_ms {2229 Some(rtt) => println!("control_path_rtt_ms={rtt:.3}"),2230 None => println!("control_path_rtt_ms=unknown"),2231 }2232 println!("media_path_mode={media_path_mode}");2233 match media_path_rtt_ms {2234 Some(rtt) => println!("media_path_rtt_ms={rtt:.3}"),2235 None => println!("media_path_rtt_ms=unknown"),2236 }2237 println!("input_path_mode={input_path_mode}");2238 match input_path_rtt_ms {2239 Some(rtt) => println!("input_path_rtt_ms={rtt:.3}"),2240 None => println!("input_path_rtt_ms=unknown"),2241 }2242 println!(2243 "forced_relay={}",2244 if args.relay_only {2245 "ok"2246 } else {2247 "not-requested"2248 }2249 );2250 println!("encoded_bytes={bytes}");2251 println!("elapsed_ms={}", started.elapsed().as_millis());2252 Ok(())2253}22542255fn pointer_smoke_events(2256 pointer_surface_dimensions: Option<PointerSurfaceDimensions>,2257) -> Result<[InputEvent; 3]> {2258 let dimensions = pointer_surface_dimensions2259 .context("host did not advertise pointer surface dimensions required by --pointer-smoke")?;2260 let [sync_position, _motion_position] = pointer_smoke_expected_positions(dimensions)?;2261 Ok([2262 InputEvent::MousePositionSync {2263 x: sync_position.x,2264 y: sync_position.y,2265 },2266 InputEvent::MouseMoveRelative { dx: 32, dy: 16 },2267 InputEvent::MouseClick { b: 1 },2268 ])2269}22702271fn pointer_smoke_expected_positions(2272 dimensions: PointerSurfaceDimensions,2273) -> Result<[PointerPosition; 2]> {2274 let sync_position = PointerPosition::new(2275 i32::from(dimensions.width / 2),2276 i32::from(dimensions.height / 2),2277 )?;2278 let motion_position = PointerPosition::new(2279 sync_position2280 .x2281 .checked_add(32)2282 .context("pointer smoke X position overflowed")?,2283 sync_position2284 .y2285 .checked_add(16)2286 .context("pointer smoke Y position overflowed")?,2287 )?;2288 Ok([sync_position, motion_position])2289}22902291fn sequence_gap(previous: u64, current: u64) -> Result<u64> {2292 let Some(expected) = previous.checked_add(1) else {2293 bail!("media sequence overflowed after {previous}");2294 };2295 ensure!(2296 current >= expected,2297 "non-monotonic media sequence: previous={previous}, current={current}"2298 );2299 Ok(current - expected)2300}23012302fn accepted_frame_rate(frame_count: u32, elapsed: Duration) -> Option<f64> {2303 if frame_count < 2 || elapsed.is_zero() {2304 return None;2305 }2306 Some(f64::from(frame_count - 1) / elapsed.as_secs_f64())2307}23082309fn selected_path_diagnostics(2310 connection: &iroh::endpoint::Connection,2311) -> (&'static str, Option<f64>) {2312 let paths = connection.paths();2313 let Some(path) = paths.iter().find(|path| path.is_selected()) else {2314 return ("unknown", None);2315 };2316 let mode = if path.is_ip() {2317 "direct"2318 } else if path.is_relay() {2319 "relay"2320 } else {2321 "custom"2322 };2323 (mode, Some(path.rtt().as_secs_f64() * 1000.0))2324}23252326fn parse_size(value: &str) -> std::result::Result<(u16, u16), String> {2327 let (width, height) = value2328 .split_once('x')2329 .ok_or_else(|| "size must be WIDTHxHEIGHT".to_owned())?;2330 let width = width.parse().map_err(|_| "invalid width".to_owned())?;2331 let height = height.parse().map_err(|_| "invalid height".to_owned())?;2332 if width == 0 || height == 0 {2333 return Err("dimensions must be non-zero".to_owned());2334 }2335 Ok((width, height))2336}23372338fn parse_minimum_fps(value: &str) -> std::result::Result<f64, String> {2339 let minimum_fps = value2340 .parse::<f64>()2341 .map_err(|_| "minimum fps must be a number".to_owned())?;2342 if !minimum_fps.is_finite() || minimum_fps <= 0.0 || minimum_fps > 240.0 {2343 return Err("minimum fps must be finite and within (0, 240]".to_owned());2344 }2345 Ok(minimum_fps)2346}23472348fn load_invitation_file(2349 path: &Path,2350 expected_host: EndpointId,2351 expected_peer: EndpointId,2352 required_grants: InvitationGrants,2353) -> Result<String> {2354 let now = SystemTime::now()2355 .duration_since(UNIX_EPOCH)2356 .context("system clock is before the Unix epoch")?2357 .as_secs();2358 load_invitation_file_at(path, expected_host, expected_peer, required_grants, now)2359}23602361fn load_invitation_file_at(2362 path: &Path,2363 expected_host: EndpointId,2364 expected_peer: EndpointId,2365 required_grants: InvitationGrants,2366 now: u64,2367) -> Result<String> {2368 let mut options = OpenOptions::new();2369 options.read(true);2370 2371 options.custom_flags(libc::O_NOFOLLOW | libc::O_CLOEXEC);2372 let file = options.open(path).with_context(|| {2373 format!(2374 "opening invitation {} without following symlinks",2375 path.display()2376 )2377 })?;2378 let metadata = file2379 .metadata()2380 .with_context(|| format!("inspecting invitation {}", path.display()))?;2381 ensure!(metadata.is_file(), "invitation must be a regular file");2382 ensure!(2383 metadata.len() <= MAX_INVITATION_FILE_BYTES,2384 "invitation {} exceeds the bounded file size",2385 path.display()2386 );2387 2388 {2389 ensure!(2390 metadata.mode() & 0o077 == 0,2391 "invitation {} must not be readable or writable by group or other users",2392 path.display()2393 );2394 ensure!(2395 metadata.uid() == unsafe { libc::geteuid() },2396 "invitation {} is not owned by the current user",2397 path.display()2398 );2399 }2400 let mut bytes = Vec::with_capacity(metadata.len() as usize);2401 file.take(MAX_INVITATION_FILE_BYTES + 1)2402 .read_to_end(&mut bytes)2403 .with_context(|| format!("reading invitation {}", path.display()))?;2404 ensure!(2405 bytes.len() as u64 <= MAX_INVITATION_FILE_BYTES,2406 "invitation {} grew beyond the bounded file size while being read",2407 path.display()2408 );2409 let token_bytes = bytes2410 .strip_suffix(b"\r\n")2411 .or_else(|| bytes.strip_suffix(b"\n"))2412 .unwrap_or(&bytes);2413 ensure!(2414 !token_bytes.is_empty() && token_bytes.len() <= MAX_INVITATION_TOKEN_LEN,2415 "invitation token has an invalid length"2416 );2417 ensure!(2418 !token_bytes.contains(&b'\n') && !token_bytes.contains(&b'\r'),2419 "invitation file contains unexpected line breaks"2420 );2421 let token = std::str::from_utf8(token_bytes).context("invitation file is not UTF-8")?;2422 let invitation =2423 SignedInvitation::decode(token).context("decoding and verifying invitation file")?;2424 ensure!(2425 invitation.claims.host_node_id == *expected_host.as_bytes(),2426 "invitation belongs to a different Sigil host"2427 );2428 ensure!(2429 invitation.claims.intended_peer_id == *expected_peer.as_bytes(),2430 "invitation is bound to a different probe identity"2431 );2432 ensure!(2433 invitation.claims.grants.contains(required_grants),2434 "invitation does not grant every requested probe capability"2435 );2436 ensure!(2437 invitation2438 .claims2439 .grants2440 .contains(InvitationGrants::POINTER_KEYBOARD)2441 || invitation.claims.grants.contains(InvitationGrants::GAMEPAD),2442 "invitation must grant pointer/keyboard or gamepad for the probe input acknowledgment"2443 );2444 ensure!(2445 now >= invitation2446 .claims2447 .issued_at_unix2448 .saturating_sub(INVITATION_CLOCK_SKEW_SECS),2449 "invitation was issued too far in the future"2450 );2451 ensure!(2452 now <= invitation.claims.expires_at_unix,2453 "invitation has expired"2454 );2455 Ok(token.to_owned())2456}24572458fn probe_client_hello(2459 nonce: [u8; 16],2460 capabilities: Vec<Capability>,2461 invitation: Option<&str>,2462) -> ClientHello {2463 let hello = ClientHello::new("sigil-probe/0.1.0", nonce, capabilities);2464 match invitation {2465 Some(invitation) => hello.with_invitation(invitation),2466 None => hello,2467 }2468}24692470async fn negotiate(2471 send: &mut iroh::endpoint::SendStream,2472 recv: &mut iroh::endpoint::RecvStream,2473 nonce: [u8; 16],2474 capabilities: Vec<Capability>,2475 required: Capability,2476 name: &str,2477 invitation: Option<&str>,2478) -> Result<Negotiated> {2479 write_client_hello(send, &probe_client_hello(nonce, capabilities, invitation))2480 .await2481 .with_context(|| format!("writing {name} hello"))?;2482 let response = tokio::time::timeout(Duration::from_secs(10), read_host_hello(recv))2483 .await2484 .with_context(|| format!("timed out waiting for {name} hello"))??2485 .with_context(|| format!("host closed during {name} hello"))?;2486 if !response.accepted {2487 bail!(2488 "host rejected {name} stream: {}",2489 response.message.as_deref().unwrap_or("unspecified reason")2490 );2491 }2492 ensure!(2493 response.capabilities.contains(&required),2494 "host accepted {name} without required capability {required:?}"2495 );2496 let session_id = response2497 .session_id2498 .with_context(|| format!("host omitted {name} session ID"))?;2499 Ok(Negotiated {2500 session_id,2501 capabilities: response.capabilities,2502 pointer_surface_dimensions: response.pointer_surface_dimensions,2503 })2504}25052506struct Negotiated {2507 session_id: u64,2508 capabilities: Vec<Capability>,2509 pointer_surface_dimensions: Option<PointerSurfaceDimensions>,2510}25112512async fn read_expected_input_ack<R>(2513 recv: &mut R,2514 timeout_seconds: u64,2515 expected_sequence: u64,2516) -> Result<InputAck>2517where2518 R: tokio::io::AsyncRead + Unpin,2519{2520 let deadline = tokio::time::Instant::now() + Duration::from_secs(timeout_seconds.min(5));2521 read_expected_input_ack_before(recv, deadline, expected_sequence).await2522}25232524async fn read_expected_input_ack_before<R>(2525 recv: &mut R,2526 deadline: tokio::time::Instant,2527 expected_sequence: u64,2528) -> Result<InputAck>2529where2530 R: tokio::io::AsyncRead + Unpin,2531{2532 loop {2533 let input_ack = tokio::time::timeout_at(deadline, read_input_ack(recv))2534 .await2535 .context("timed out waiting for input acknowledgment")??2536 .context("host closed before input acknowledgment")?;2537 ensure!(2538 input_ack.sequence <= expected_sequence,2539 "unexpected input acknowledgment sequence {}; expected at most {expected_sequence}",2540 input_ack.sequence2541 );2542 if input_ack.sequence == expected_sequence {2543 return Ok(input_ack);2544 }2545 }2546}25472548async fn confirm_pointer_position<S, R>(2549 send: &mut S,2550 recv: &mut R,2551 expected_sequence: &mut u64,2552 mut input_ack: InputAck,2553 expected_position: PointerPosition,2554 timeout: Duration,2555) -> Result<PointerPosition>2556where2557 S: tokio::io::AsyncWrite + Unpin,2558 R: tokio::io::AsyncRead + Unpin,2559{2560 ensure!(2561 !timeout.is_zero(),2562 "pointer feedback confirmation timeout must be nonzero"2563 );2564 let deadline = tokio::time::Instant::now() + timeout.min(Duration::from_secs(5));2565 loop {2566 if input_ack.pointer_position == Some(expected_position) {2567 return Ok(expected_position);2568 }2569 let now = tokio::time::Instant::now();2570 if now >= deadline {2571 bail!(2572 "compositor pointer position did not converge to {},{}; last observed position was {:?}",2573 expected_position.x,2574 expected_position.y,2575 input_ack.pointer_position2576 );2577 }2578 tokio::time::sleep_until(std::cmp::min(now + Duration::from_millis(16), deadline)).await;2579 if tokio::time::Instant::now() >= deadline {2580 bail!(2581 "compositor pointer position did not converge to {},{}; last observed position was {:?}",2582 expected_position.x,2583 expected_position.y,2584 input_ack.pointer_position2585 );2586 }2587 write_input_event(send, &InputEvent::Probe)2588 .await2589 .context("writing pointer feedback convergence probe")?;2590 *expected_sequence = expected_sequence2591 .checked_add(1)2592 .context("input acknowledgment sequence overflowed")?;2593 input_ack = match read_expected_input_ack_before(recv, deadline, *expected_sequence).await {2594 Ok(input_ack) => input_ack,2595 Err(error) if tokio::time::Instant::now() >= deadline => {2596 bail!(2597 "compositor pointer position did not converge to {},{}; last observed position was {:?}: {error:#}",2598 expected_position.x,2599 expected_position.y,2600 input_ack.pointer_position2601 );2602 }2603 Err(error) => return Err(error),2604 };2605 }2606}260726082609mod tests {2610 use super::*;26112612 2613 async fn expected_input_ack_skips_older_feedback_under_one_deadline() {2614 let (mut sender, mut receiver) = tokio::io::duplex(512);2615 for sequence in [0, 0, 1] {2616 sigil_protocol::write_input_ack(2617 &mut sender,2618 &sigil_protocol::InputAck {2619 sequence,2620 pointer_position: None,2621 pointer_visible: Some(false),2622 },2623 )2624 .await2625 .unwrap();2626 }26272628 let input_ack = read_expected_input_ack(&mut receiver, 1, 1).await.unwrap();26292630 assert_eq!(input_ack.sequence, 1);2631 }26322633 2634 async fn expected_input_ack_rejects_future_feedback() {2635 let (mut sender, mut receiver) = tokio::io::duplex(256);2636 sigil_protocol::write_input_ack(2637 &mut sender,2638 &sigil_protocol::InputAck {2639 sequence: 2,2640 pointer_position: None,2641 pointer_visible: Some(false),2642 },2643 )2644 .await2645 .unwrap();26462647 let error = read_expected_input_ack(&mut receiver, 1, 1)2648 .await2649 .unwrap_err();26502651 assert!(error.to_string().contains("expected at most 1"));2652 }26532654 2655 async fn pointer_confirmation_polls_until_the_tracker_reports_the_exact_position() {2656 let target = PointerPosition::new(1_280, 800).unwrap();2657 let initial_ack = InputAck {2658 sequence: 2,2659 pointer_position: Some(PointerPosition::new(10, 20).unwrap()),2660 pointer_visible: Some(false),2661 };2662 let (client, server) = tokio::io::duplex(512);2663 let (mut client_recv, mut client_send) = tokio::io::split(client);2664 let (mut server_recv, mut server_send) = tokio::io::split(server);2665 let server_task = tokio::spawn(async move {2666 let event = sigil_protocol::read_input_event(&mut server_recv)2667 .await2668 .unwrap()2669 .unwrap();2670 assert_eq!(event, InputEvent::Probe);2671 sigil_protocol::write_input_ack(2672 &mut server_send,2673 &InputAck {2674 sequence: 3,2675 pointer_position: Some(target),2676 pointer_visible: Some(false),2677 },2678 )2679 .await2680 .unwrap();2681 });2682 let mut expected_sequence = 2;26832684 let observed = confirm_pointer_position(2685 &mut client_send,2686 &mut client_recv,2687 &mut expected_sequence,2688 initial_ack,2689 target,2690 Duration::from_secs(1),2691 )2692 .await2693 .unwrap();26942695 assert_eq!(observed, target);2696 assert_eq!(expected_sequence, 3);2697 server_task.await.unwrap();2698 }26992700 2701 async fn pointer_confirmation_fails_when_the_tracker_never_converges() {2702 let target = PointerPosition::new(1_280, 800).unwrap();2703 let initial_ack = InputAck {2704 sequence: 2,2705 pointer_position: Some(PointerPosition::new(10, 20).unwrap()),2706 pointer_visible: Some(false),2707 };2708 let (client, server_guard) = tokio::io::duplex(512);2709 let (mut client_recv, mut client_send) = tokio::io::split(client);2710 let mut expected_sequence = 2;27112712 let error = confirm_pointer_position(2713 &mut client_send,2714 &mut client_recv,2715 &mut expected_sequence,2716 initial_ack,2717 target,2718 Duration::from_millis(25),2719 )2720 .await2721 .unwrap_err();27222723 assert!(error.to_string().contains("did not converge to 1280,800"));2724 drop(server_guard);2725 }27262727 2728 fn feedback_report_is_complete_and_keeps_the_observed_sequence() {2729 let report = complete_feedback_report(7, MediaFeedbackFlags::NONE, 42);2730 report.validate().unwrap();2731 assert_eq!(report.report_id, 7);2732 assert_eq!(report.interval_ms, 1_000);2733 assert_eq!(report.last_sequence, Some(42));2734 assert_eq!(report.frontend_queue_depth, 0);2735 assert_eq!(report.decode_queue_depth, 0);2736 assert_eq!(report.presenter_queue_depth, 0);2737 assert_eq!(report.transport_delivery_p95_ms, Some(10));2738 assert_eq!(report.decode_p95_ms, Some(3));2739 assert_eq!(report.presentation_p95_ms, Some(5));27402741 let later = complete_feedback_report(8, MediaFeedbackFlags::NONE, 42);2742 assert_eq!(later.last_sequence, report.last_sequence);2743 }27442745 2746 fn resolution_stall_reduced_dimensions_are_exact_even_and_same_aspect() {2747 assert_eq!(2748 resolution_stall_reduced_dimensions((2_560, 1_600)).unwrap(),2749 (1_920, 1_200)2750 );2751 assert_eq!(2752 resolution_stall_reduced_dimensions((1_280, 800)).unwrap(),2753 (960, 600)2754 );2755 assert!(resolution_stall_reduced_dimensions((1_278, 800)).is_err());2756 assert!(resolution_stall_reduced_dimensions((80, 64)).is_err());2757 }27582759 2760 fn sequence_checks_reject_duplicates_regressions_and_overflow() {2761 assert_eq!(sequence_gap(41, 42).unwrap(), 0);2762 assert_eq!(sequence_gap(41, 45).unwrap(), 3);2763 assert!(sequence_gap(41, 41).is_err());2764 assert!(sequence_gap(41, 40).is_err());2765 assert!(sequence_gap(u64::MAX, 0).is_err());2766 }27672768 2769 fn pointer_smoke_uses_negotiated_native_surface_center_and_order() {2770 let dimensions = PointerSurfaceDimensions::new(2_560, 1_600).unwrap();27712772 assert_eq!(2773 pointer_smoke_events(Some(dimensions)).unwrap(),2774 [2775 InputEvent::MousePositionSync { x: 1_280, y: 800 },2776 InputEvent::MouseMoveRelative { dx: 32, dy: 16 },2777 InputEvent::MouseClick { b: 1 },2778 ]2779 );2780 assert_eq!(2781 pointer_smoke_expected_positions(dimensions).unwrap(),2782 [2783 PointerPosition::new(1_280, 800).unwrap(),2784 PointerPosition::new(1_312, 816).unwrap(),2785 ]2786 );2787 }27882789 2790 fn pointer_smoke_requires_negotiated_native_surface_dimensions() {2791 let error = pointer_smoke_events(None).unwrap_err();27922793 assert!(2794 error2795 .to_string()2796 .contains("host did not advertise pointer surface dimensions")2797 );2798 }27992800 fn media_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {2801 let payload = vec![0x65, 0x88, 0x84];2802 MediaFrame::new(2803 sigil_protocol::MediaFrameHeader::h264(2804 1_280,2805 800,2806 payload.len(),2807 sequence,2808 sequence * 1_000,2809 i64::try_from(sequence * 1_000).unwrap(),2810 flags,2811 )2812 .unwrap(),2813 payload,2814 )2815 .unwrap()2816 }28172818 fn media_object_v3(2819 group_id: u64,2820 object_id: u32,2821 sequence: u64,2822 flags: FrameFlags,2823 ) -> MediaObjectV3 {2824 let payload = vec![0x65, 0x88, 0x84];2825 let header = sigil_protocol::MediaObjectHeaderV3::h264(2826 1_280,2827 800,2828 payload.len(),2829 if object_id == 0 { 0 } else { 128 },2830 flags,2831 object_id,2832 group_id,2833 sequence,2834 sequence * 1_000,2835 i64::try_from(sequence * 1_000).unwrap(),2836 100,2837 )2838 .unwrap();2839 MediaObjectV3::new(header, payload).unwrap()2840 }28412842 fn media_outcome_v3(2843 accept_index: u64,2844 group_id: u64,2845 object_id: u32,2846 sequence: u64,2847 flags: FrameFlags,2848 ) -> MediaObjectOutcomeV3 {2849 MediaObjectOutcomeV3::Object {2850 accept_index,2851 object: media_object_v3(group_id, object_id, sequence, flags),2852 }2853 }28542855 2856 fn media_v3_discontinuity_object_zero_fast_forwards_accept_order() {2857 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2858 let barrier = keyframe.union(FrameFlags::DISCONTINUITY);2859 let mut reorder = MediaObjectReorderV3::new(0);2860 let mut sequence = MediaObjectSequenceV3::new();28612862 assert!(2863 reorder2864 .push(media_outcome_v3(1, 10, 1, 11, FrameFlags::NONE))2865 .unwrap()2866 .is_none()2867 );2868 let recovered = reorder2869 .push(media_outcome_v3(2, 20, 0, 20, barrier))2870 .unwrap()2871 .unwrap();2872 assert_eq!(recovered.accept_index(), 2);2873 let MediaObjectOutcomeV3::Object { object, .. } = recovered else {2874 panic!("recovery barrier must remain an object");2875 };2876 assert_eq!(2877 sequence.classify(&object),2878 MediaObjectDecisionV3::Deliver {2879 discontinuity: true2880 }2881 );2882 assert_eq!(reorder.pending_len(), 0);2883 assert!(2884 reorder2885 .push(MediaObjectOutcomeV3::Dropped { accept_index: 0 })2886 .unwrap()2887 .is_none()2888 );2889 let next = reorder2890 .push(media_outcome_v3(3, 20, 1, 21, FrameFlags::NONE))2891 .unwrap()2892 .unwrap();2893 let MediaObjectOutcomeV3::Object { object, .. } = next else {2894 panic!("recovered-group delta must remain an object");2895 };2896 assert_eq!(2897 sequence.classify(&object),2898 MediaObjectDecisionV3::Deliver {2899 discontinuity: false2900 }2901 );2902 }29032904 2905 fn minimum_fps_parser_and_first_to_last_evaluation_are_bounded() {2906 assert_eq!(parse_minimum_fps("55").unwrap(), 55.0);2907 assert_eq!(parse_minimum_fps("240").unwrap(), 240.0);2908 for invalid in ["0", "-1", "240.1", "NaN", "inf", "not-a-number"] {2909 assert!(parse_minimum_fps(invalid).is_err(), "{invalid}");2910 }29112912 assert_eq!(accepted_frame_rate(56, Duration::from_secs(1)), Some(55.0));2913 assert_eq!(accepted_frame_rate(1, Duration::from_secs(1)), None);2914 assert_eq!(accepted_frame_rate(2, Duration::ZERO), None);2915 assert!(accepted_frame_rate(56, Duration::from_secs(1)).unwrap() >= 55.0);2916 assert!(accepted_frame_rate(55, Duration::from_secs(1)).unwrap() < 55.0);2917 }29182919 2920 fn media_v3_enforces_wire_group_object_and_sequence_continuity() {2921 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2922 let barrier = keyframe.union(FrameFlags::DISCONTINUITY);2923 let mut sequence = MediaObjectSequenceV3::new();29242925 assert_eq!(2926 sequence.classify(&media_object_v3(10, 0, 10, keyframe)),2927 MediaObjectDecisionV3::Deliver {2928 discontinuity: true2929 }2930 );2931 assert_eq!(2932 sequence.classify(&media_object_v3(10, 1, 11, FrameFlags::NONE)),2933 MediaObjectDecisionV3::Deliver {2934 discontinuity: false2935 }2936 );2937 assert_eq!(2938 sequence.classify(&media_object_v3(10, 3, 13, FrameFlags::NONE)),2939 MediaObjectDecisionV3::DropUntilKeyframe2940 );2941 assert_eq!(2942 sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),2943 MediaObjectDecisionV3::DropUntilKeyframe2944 );2945 assert_eq!(2946 sequence.classify(&media_object_v3(20, 0, 20, barrier)),2947 MediaObjectDecisionV3::Deliver {2948 discontinuity: true2949 }2950 );2951 assert_eq!(2952 sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),2953 MediaObjectDecisionV3::DropLate2954 );2955 }29562957 2958 fn explicit_v3_recovery_never_delivers_same_group_delta_history() {2959 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2960 let barrier = keyframe.union(FrameFlags::DISCONTINUITY);2961 let mut sequence = MediaObjectSequenceV3::new();29622963 assert!(matches!(2964 sequence.classify(&media_object_v3(10, 0, 10, keyframe)),2965 MediaObjectDecisionV3::Deliver { .. }2966 ));2967 assert!(matches!(2968 sequence.classify(&media_object_v3(10, 1, 11, FrameFlags::NONE)),2969 MediaObjectDecisionV3::Deliver { .. }2970 ));2971 sequence.request_recovery();2972 assert_eq!(2973 sequence.classify(&media_object_v3(10, 2, 12, FrameFlags::NONE)),2974 MediaObjectDecisionV3::DropUntilKeyframe2975 );2976 assert_eq!(2977 sequence.classify(&media_object_v3(20, 0, 20, keyframe)),2978 MediaObjectDecisionV3::DropUntilKeyframe,2979 "recovery after skipped history must be explicitly discontinuous"2980 );2981 assert_eq!(2982 sequence.classify(&media_object_v3(30, 0, 30, barrier)),2983 MediaObjectDecisionV3::Deliver {2984 discontinuity: true2985 }2986 );2987 }29882989 2990 fn media_v3_group_payload_is_bounded_by_the_shared_protocol_limit() {2991 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2992 let mut sequence = MediaObjectSequenceV3::new();2993 assert!(matches!(2994 sequence.classify(&media_object_v3(1, 0, 1, keyframe)),2995 MediaObjectDecisionV3::Deliver { .. }2996 ));2997 sequence.group_payload_bytes = MAX_MEDIA_GROUP_BYTES_V3;2998 assert_eq!(2999 sequence.classify(&media_object_v3(1, 1, 2, FrameFlags::NONE)),3000 MediaObjectDecisionV3::DropUntilKeyframe3001 );3002 }30033004 3005 fn upstream_moq_group_and_object_limits_fail_closed() {3006 assert!(!classify_moq_probe_group(None, 8).unwrap());3007 assert!(!classify_moq_probe_group(Some(8), 9).unwrap());3008 assert!(classify_moq_probe_group(Some(8), 10).unwrap());3009 assert!(classify_moq_probe_group(Some(8), 8).is_err());3010 assert!(classify_moq_probe_group(Some(8), 7).is_err());30113012 assert_eq!(validate_moq_probe_object_bounds(4, 0, 0, 42).unwrap(), 42);3013 assert!(3014 validate_moq_probe_object_bounds(4, MAX_MEDIA_OBJECT_ID_V3 as usize + 1, 0, 1,)3015 .is_err()3016 );3017 assert!(validate_moq_probe_object_bounds(4, 0, MAX_MEDIA_GROUP_BYTES_V3, 1).is_err());3018 assert!(validate_moq_probe_object_bounds(4, 0, usize::MAX, 1).is_err());3019 }30203021 3022 fn upstream_moq_requires_configured_group_zero_and_idr_recovery() {3023 let keyframe = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);3024 let recovery = keyframe.union(FrameFlags::DISCONTINUITY);3025 let delta = media_frame(11, FrameFlags::NONE);3026 let configured = media_frame(10, keyframe);3027 let configured_next = media_frame(12, keyframe);3028 let recovered = media_frame(20, recovery);30293030 assert!(validate_moq_probe_frame(3, true, None, &delta).is_err());3031 assert!(validate_moq_probe_frame(3, true, None, &configured).is_ok());3032 assert!(validate_moq_probe_frame(4, true, Some(11), &configured).is_err());3033 assert!(validate_moq_probe_frame(4, true, Some(11), &configured_next).is_ok());3034 assert!(validate_moq_probe_frame(4, true, Some(11), &recovered).is_ok());3035 assert!(validate_moq_probe_frame(4, false, Some(10), &delta).is_ok());3036 assert!(validate_moq_probe_frame(4, false, Some(9), &delta).is_err());3037 }30383039 3040 fn upstream_moq_transport_reports_the_pinned_native_alpn() {3041 assert_eq!(MediaTransport::UpstreamMoq.label(), "iroh-moq");3042 assert_eq!(MediaTransport::UpstreamMoq.alpn(), CONTROL_ALPN_V1);3043 assert_eq!(3044 MediaTransport::UpstreamMoq.media_alpn_label(),3045 "moq-lite-04"3046 );3047 assert_eq!(3048 MediaTransport::UpstreamMoq.media_alpn_label().as_bytes(),3049 iroh_moq::ALPN3050 );3051 }30523053 fn test_invitation(3054 host: &SecretKey,3055 peer: &SecretKey,3056 grants: InvitationGrants,3057 issued_at_unix: u64,3058 expires_at_unix: u64,3059 ) -> String {3060 let claims = sigil_protocol::InvitationClaims::new(3061 *host.public().as_bytes(),3062 *peer.public().as_bytes(),3063 issued_at_unix,3064 expires_at_unix,3065 1,3066 [0x55; 32],3067 grants,3068 )3069 .unwrap();3070 SignedInvitation::issue(claims, &host.to_bytes())3071 .unwrap()3072 .encode()3073 }30743075 fn write_private_invitation(path: &Path, token: &str, ending: &[u8]) {3076 let mut bytes = token.as_bytes().to_vec();3077 bytes.extend_from_slice(ending);3078 std::fs::write(path, bytes).unwrap();3079 3080 {3081 use std::os::unix::fs::PermissionsExt as _;3082 std::fs::set_permissions(path, std::fs::Permissions::from_mode(0o600)).unwrap();3083 }3084 }30853086 3087 fn persistent_probe_credentials_are_explicit_and_paired() {3088 let node_id = SecretKey::from_bytes(&[0x2a; 32]).public().to_string();3089 assert!(3090 Args::try_parse_from([3091 "sigil-probe",3092 "--node-id",3093 &node_id,3094 "--invitation",3095 "probe.goq-invite",3096 ])3097 .is_err()3098 );3099 let persistent = Args::try_parse_from([3100 "sigil-probe",3101 "--node-id",3102 &node_id,3103 "--identity",3104 "probe.key",3105 "--invitation",3106 "probe.goq-invite",3107 ])3108 .unwrap();3109 assert_eq!(persistent.identity, Some(PathBuf::from("probe.key")));3110 assert_eq!(3111 persistent.invitation,3112 Some(PathBuf::from("probe.goq-invite"))3113 );3114 }31153116 3117 fn private_invitation_preflight_accepts_only_the_bound_current_grants() {3118 let temp = tempfile::tempdir().unwrap();3119 let path = temp.path().join("probe.goq-invite");3120 let host = SecretKey::from_bytes(&[0x31; 32]);3121 let peer = SecretKey::from_bytes(&[0x32; 32]);3122 let other = SecretKey::from_bytes(&[0x33; 32]);3123 let now = SystemTime::now()3124 .duration_since(UNIX_EPOCH)3125 .unwrap()3126 .as_secs();3127 let token = test_invitation(3128 &host,3129 &peer,3130 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),3131 now.saturating_sub(1),3132 now + 600,3133 );31343135 write_private_invitation(&path, &token, b"\n");3136 assert_eq!(3137 load_invitation_file_at(3138 &path,3139 host.public(),3140 peer.public(),3141 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),3142 now,3143 )3144 .unwrap(),3145 token3146 );3147 write_private_invitation(&path, &token, b"\r\n");3148 assert!(3149 load_invitation_file_at(3150 &path,3151 host.public(),3152 peer.public(),3153 InvitationGrants::VIEW,3154 now,3155 )3156 .is_ok()3157 );3158 assert!(3159 load_invitation_file_at(3160 &path,3161 other.public(),3162 peer.public(),3163 InvitationGrants::VIEW,3164 now,3165 )3166 .is_err()3167 );3168 assert!(3169 load_invitation_file_at(3170 &path,3171 host.public(),3172 other.public(),3173 InvitationGrants::VIEW,3174 now,3175 )3176 .is_err()3177 );3178 assert!(3179 load_invitation_file_at(3180 &path,3181 host.public(),3182 peer.public(),3183 InvitationGrants::POINTER_KEYBOARD,3184 now,3185 )3186 .is_err()3187 );31883189 let view_only = test_invitation(3190 &host,3191 &peer,3192 InvitationGrants::VIEW,3193 now.saturating_sub(1),3194 now + 600,3195 );3196 write_private_invitation(&path, &view_only, b"\n");3197 assert!(3198 load_invitation_file_at(3199 &path,3200 host.public(),3201 peer.public(),3202 InvitationGrants::VIEW,3203 now,3204 )3205 .unwrap_err()3206 .to_string()3207 .contains("input acknowledgment")3208 );32093210 let expired = test_invitation(3211 &host,3212 &peer,3213 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),3214 now.saturating_sub(120),3215 now.saturating_sub(60),3216 );3217 write_private_invitation(&path, &expired, b"\n");3218 assert!(3219 load_invitation_file_at(3220 &path,3221 host.public(),3222 peer.public(),3223 InvitationGrants::VIEW,3224 now,3225 )3226 .unwrap_err()3227 .to_string()3228 .contains("expired")3229 );3230 write_private_invitation(&path, &token, b"\n\n");3231 assert!(3232 load_invitation_file_at(3233 &path,3234 host.public(),3235 peer.public(),3236 InvitationGrants::VIEW,3237 now,3238 )3239 .is_err()3240 );3241 let future = test_invitation(3242 &host,3243 &peer,3244 InvitationGrants::VIEW.union(InvitationGrants::GAMEPAD),3245 now + INVITATION_CLOCK_SKEW_SECS + 1,3246 now + INVITATION_CLOCK_SKEW_SECS + 61,3247 );3248 write_private_invitation(&path, &future, b"\n");3249 assert!(3250 load_invitation_file_at(3251 &path,3252 host.public(),3253 peer.public(),3254 InvitationGrants::VIEW,3255 now,3256 )3257 .unwrap_err()3258 .to_string()3259 .contains("future")3260 );3261 }32623263 3264 3265 fn invitation_reader_rejects_unsafe_files() {3266 use std::os::unix::fs::{PermissionsExt as _, symlink};32673268 let temp = tempfile::tempdir().unwrap();3269 let path = temp.path().join("probe.goq-invite");3270 let link = temp.path().join("linked.goq-invite");3271 let host = SecretKey::from_bytes(&[0x41; 32]);3272 let peer = SecretKey::from_bytes(&[0x42; 32]);3273 let now = SystemTime::now()3274 .duration_since(UNIX_EPOCH)3275 .unwrap()3276 .as_secs();3277 let token = test_invitation(3278 &host,3279 &peer,3280 InvitationGrants::VIEW,3281 now.saturating_sub(1),3282 now + 60,3283 );3284 write_private_invitation(&path, &token, b"\n");3285 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o644)).unwrap();3286 assert!(3287 load_invitation_file(&path, host.public(), peer.public(), InvitationGrants::VIEW,)3288 .is_err()3289 );3290 std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o600)).unwrap();3291 symlink(&path, &link).unwrap();3292 assert!(3293 load_invitation_file(&link, host.public(), peer.public(), InvitationGrants::VIEW,)3294 .is_err()3295 );3296 std::fs::write(&path, vec![b'x'; MAX_INVITATION_FILE_BYTES as usize + 1]).unwrap();3297 assert!(3298 load_invitation_file(&path, host.public(), peer.public(), InvitationGrants::VIEW,)3299 .is_err()3300 );3301 }33023303 3304 fn invitation_is_attached_only_when_the_caller_selects_the_media_hello() {3305 let host = SecretKey::from_bytes(&[0x51; 32]);3306 let peer = SecretKey::from_bytes(&[0x52; 32]);3307 let now = SystemTime::now()3308 .duration_since(UNIX_EPOCH)3309 .unwrap()3310 .as_secs();3311 let token = test_invitation(3312 &host,3313 &peer,3314 InvitationGrants::VIEW,3315 now.saturating_sub(1),3316 now + 60,3317 );3318 let media = probe_client_hello([7; 16], vec![Capability::VideoH264], Some(&token));3319 let input = probe_client_hello([7; 16], vec![Capability::InputAck], None);3320 assert_eq!(media.invitation.as_deref(), Some(token.as_str()));3321 assert_eq!(input.invitation, None);3322 }33233324 3325 fn media_compatibility_flag_defaults_and_conflicts_are_enforced() {3326 let node_id = SecretKey::from_bytes(&[0x2a; 32]).public().to_string();3327 let default = Args::try_parse_from(["sigil-probe", "--node-id", &node_id]).unwrap();3328 assert!(!default.media_v3);3329 assert_eq!(default.identity, None);3330 assert_eq!(default.invitation, None);3331 assert_eq!(default.expect_size, None);3332 assert!(!default.feedback_smoke);3333 let strict = Args::try_parse_from([3334 "sigil-probe",3335 "--node-id",3336 &node_id,3337 "--expect-size",3338 "2560x1600",3339 ])3340 .unwrap();3341 assert_eq!(strict.expect_size, Some((2_560, 1_600)));3342 let feedback =3343 Args::try_parse_from(["sigil-probe", "--node-id", &node_id, "--feedback-smoke"])3344 .unwrap();3345 assert!(feedback.feedback_smoke);3346 assert!(3347 Args::try_parse_from([3348 "sigil-probe",3349 "--node-id",3350 &node_id,3351 "--feedback-smoke",3352 "--media-v3",3353 ])3354 .is_err()3355 );3356 }3357}