Add bounded independent QUIC media objects with keyframe-prioritized cancellation and v1 compatibility. Teach the client and probe to recover from reset or late objects, expose transport diagnostics, and extend the loopback release proof.
- commit
- 0bc5b443ddd38f84f95865de54b5337ca8a6906e
- parent
- e3268e48cffd9ff7410fe96fdfbdf2e5623457d2
- author
- Dami <dami.etoile@gmail.com>
| M | crates/ | 361 | ++++++++++++++++- |
| M | crates/ | 11 | +- |
| M | crates/ | 560 | ++++++++++++++++++++++++++- |
| M | crates/ | 8 | +- |
| M | crates/ | 61 | +++ |
| M | scripts/ | 40 | ++- |
| M | src-tauri/ | 633 | +++++++++++++++++++++++++----- |
| M | src/ | 10 | + |
| M | src/ | 22 | + |
| M | src/ | 3 | + |
| M | src/ | 21 | + |
@@ -5,16 +5,15 @@ use clap::Parser;
55 use iroh::{Endpoint, EndpointAddr, EndpointId, SecretKey, endpoint::presets};66 use sigil_protocol::{77 Capability, ClientHello, FrameFlags, GAMEPAD_AXIS_MAX, GAMEPAD_AXIS_MIN, GAMEPAD_TRIGGER_MAX,8- GamepadState, INPUT_ALPN_V1, InputEvent, MEDIA_ALPN_V1, PointerSurfaceDimensions,9- read_host_hello, read_input_ack, read_media_frame, write_client_hello, write_input_event,8+ GamepadState, INPUT_ALPN_V1, InputEvent, MEDIA_ALPN_V1, MEDIA_ALPN_V2, MediaFrame,9+ PointerSurfaceDimensions, ProtocolError, read_host_hello, read_input_ack, read_media_frame,10+ read_media_object, write_client_hello, write_input_event,1011 };1112 13+const MEDIA_OBJECT_CAPACITY: usize = 4;14+1215 13-14-15-16-17-16+1817 struct Args {1918 2019 node_id: EndpointId,@@ -24,6 +23,10 @@ struct Args {
2423 timeout_seconds: u64,2524 2625 expect_size: (u16, u16),26+ /// Exercise the reliable ordered v1 media stream instead of independent27+ /// v2 media objects. Intended only for compatibility validation.28+ 29+ media_v1: bool,2730 /// Require gamepad negotiation and emit one bounded non-neutral snapshot2831 /// followed by neutral. Intended for evtest-backed uinput proof.2932 @@ -34,6 +37,166 @@ struct Args {
3437 pointer_smoke: bool,3538 }3639 40+41+enum MediaTransport {42+ IndependentV2,43+ ReliableV1,44+}45+46+impl MediaTransport {47+ fn alpn(self) -> &'static [u8] {48+ match self {49+ Self::IndependentV2 => MEDIA_ALPN_V2,50+ Self::ReliableV1 => MEDIA_ALPN_V1,51+ }52+ }53+54+ fn label(self) -> &'static str {55+ match self {56+ Self::IndependentV2 => "independent-v2",57+ Self::ReliableV1 => "reliable-v1",58+ }59+ }60+}61+62+63+enum MediaObjectOutcome {64+ Frame { index: u64, frame: MediaFrame },65+ Dropped { index: u64 },66+ Malformed { index: u64, error: ProtocolError },67+}68+69+struct MediaObjectReceiver {70+ connection: iroh::endpoint::Connection,71+ reads: tokio::task::JoinSet<MediaObjectOutcome>,72+ next_index: u64,73+ accepting: bool,74+ read_timeout: Duration,75+}76+77+impl MediaObjectReceiver {78+ fn new(connection: iroh::endpoint::Connection, read_timeout: Duration) -> Self {79+ Self {80+ connection,81+ reads: tokio::task::JoinSet::new(),82+ next_index: 0,83+ accepting: true,84+ read_timeout,85+ }86+ }87+88+ async fn next(&mut self) -> Result<Option<MediaObjectOutcome>> {89+ loop {90+ if !self.accepting && self.reads.is_empty() {91+ return Ok(None);92+ }93+94+ tokio::select! {95+ accepted = self.connection.accept_uni(),96+ if self.accepting && self.reads.len() < MEDIA_OBJECT_CAPACITY => {97+ match accepted {98+ Ok(mut stream) => {99+ let index = self.next_index;100+ self.next_index = self101+ .next_index102+ .checked_add(1)103+ .context("media object index overflowed")?;104+ let read_timeout = self.read_timeout;105+ self.reads.spawn(async move {106+ match tokio::time::timeout(read_timeout, read_media_object(&mut stream)).await {107+ Ok(Ok(frame)) => MediaObjectOutcome::Frame { index, frame },108+ Ok(Err(ProtocolError::Io(_))) => MediaObjectOutcome::Dropped { index },109+ Ok(Err(error)) => MediaObjectOutcome::Malformed { index, error },110+ Err(_) => MediaObjectOutcome::Dropped { index },111+ }112+ });113+ }114+ Err(_) => self.accepting = false,115+ }116+ }117+ completed = self.reads.join_next(), if !self.reads.is_empty() => {118+ return match completed.expect("guarded by non-empty media object task set") {119+ Ok(outcome) => Ok(Some(outcome)),120+ Err(error) if error.is_cancelled() => continue,121+ Err(error) => Err(error).context("media object read task failed"),122+ };123+ }124+ }125+ }126+ }127+}128+129+impl Drop for MediaObjectReceiver {130+ fn drop(&mut self) {131+ self.reads.abort_all();132+ }133+}134+135+136+enum MediaObjectDecision {137+ Deliver,138+ DropLate,139+ DropUntilKeyframe,140+}141+142+143+struct MediaObjectSequence {144+ completion_watermark: Option<u64>,145+ last_sequence: Option<u64>,146+ waiting_for_keyframe: bool,147+}148+149+impl MediaObjectSequence {150+ fn new() -> Self {151+ Self {152+ waiting_for_keyframe: true,153+ ..Self::default()154+ }155+ }156+157+ fn note_drop(&mut self, index: u64) -> bool {158+ if self159+ .completion_watermark160+ .is_some_and(|watermark| index <= watermark)161+ {162+ return false;163+ }164+ self.completion_watermark = Some(index);165+ self.waiting_for_keyframe = true;166+ true167+ }168+169+ fn classify(&mut self, index: u64, frame: &MediaFrame) -> MediaObjectDecision {170+ if self171+ .completion_watermark172+ .is_some_and(|watermark| index <= watermark)173+ {174+ return MediaObjectDecision::DropLate;175+ }176+ self.completion_watermark = Some(index);177+178+ let independently_decodable = frame.header.flags.contains(FrameFlags::KEYFRAME)179+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG);180+ let sequence_contiguous = self181+ .last_sequence182+ .is_none_or(|last| last.checked_add(1) == Some(frame.header.sequence));183+ let sequence_monotonic = self184+ .last_sequence185+ .is_none_or(|last| frame.header.sequence > last);186+187+ let resync_required =188+ !sequence_monotonic || self.waiting_for_keyframe || !sequence_contiguous;189+ if resync_required && (!independently_decodable || !sequence_monotonic) {190+ self.waiting_for_keyframe = true;191+ return MediaObjectDecision::DropUntilKeyframe;192+ }193+194+ self.last_sequence = Some(frame.header.sequence);195+ self.waiting_for_keyframe = false;196+ MediaObjectDecision::Deliver197+ }198+}199+37200 38201 async fn main() -> Result<()> {39202 let args = Args::parse();@@ -52,9 +215,14 @@ async fn main() -> Result<()> {
52215 .context("binding probe endpoint")?;53216 let _ = tokio::time::timeout(Duration::from_secs(10), endpoint.online()).await;54217 let address = EndpointAddr::new(args.node_id);218+ let media_transport = if args.media_v1 {219+ MediaTransport::ReliableV1220+ } else {221+ MediaTransport::IndependentV2222+ };55223 56224 let media_connection = endpoint57- .connect(address.clone(), MEDIA_ALPN_V1)225+ .connect(address.clone(), media_transport.alpn())58226 .await59227 .context("connecting media protocol")?;60228 let (mut media_send, mut media_recv) = media_connection@@ -71,6 +239,16 @@ async fn main() -> Result<()> {
71239 )72240 .await?;73241 let session_id = media_negotiation.session_id;242+ let mut media_send = Some(media_send);243+ let mut media_recv = Some(media_recv);244+ if media_transport == MediaTransport::IndependentV2 {245+ media_send246+ .take()247+ .expect("media handshake stream is present")248+ .finish()249+ .context("finishing media v2 handshake request")?;250+ drop(media_recv.take());251+ }74252 75253 let input_connection = endpoint76254 .connect(address, INPUT_ALPN_V1)@@ -178,17 +356,68 @@ async fn main() -> Result<()> {
178356 let mut bytes = 0_u64;179357 let mut keyframes = 0_u32;180358 let mut gaps = 0_u64;359+ let mut media_objects_dropped = 0_u64;360+ let mut media_objects_late = 0_u64;181361 let mut last_sequence: Option<u64> = None;182362 let mut dimensions = None;183-184- while received < args.frames {185- let frame = tokio::time::timeout(363+ let mut object_receiver = (media_transport == MediaTransport::IndependentV2).then(|| {364+ MediaObjectReceiver::new(365+ media_connection.clone(),186366 Duration::from_secs(args.timeout_seconds),187- read_media_frame(&mut media_recv),188367 )189- .await190- .context("timed out waiting for media frame")??191- .context("host closed the media stream")?;368+ });369+ let mut object_sequence = MediaObjectSequence::new();370+371+ while received < args.frames {372+ let frame = match media_transport {373+ MediaTransport::ReliableV1 => tokio::time::timeout(374+ Duration::from_secs(args.timeout_seconds),375+ read_media_frame(376+ media_recv377+ .as_mut()378+ .expect("v1 media receive stream is present"),379+ ),380+ )381+ .await382+ .context("timed out waiting for media frame")??383+ .context("host closed the media stream")?,384+ MediaTransport::IndependentV2 => loop {385+ let outcome = tokio::time::timeout(386+ Duration::from_secs(args.timeout_seconds)387+ .saturating_add(Duration::from_secs(1)),388+ object_receiver389+ .as_mut()390+ .expect("v2 media object receiver is present")391+ .next(),392+ )393+ .await394+ .context("timed out waiting for media object")??395+ .context("host closed the media object connection")?;396+ match outcome {397+ MediaObjectOutcome::Dropped { index } => {398+ if object_sequence.note_drop(index) {399+ media_objects_dropped = media_objects_dropped.saturating_add(1);400+ } else {401+ media_objects_late = media_objects_late.saturating_add(1);402+ }403+ }404+ MediaObjectOutcome::Malformed { index, error } => {405+ bail!("media object {index} is malformed: {error}");406+ }407+ MediaObjectOutcome::Frame { index, frame } => {408+ match object_sequence.classify(index, &frame) {409+ MediaObjectDecision::Deliver => break frame,410+ MediaObjectDecision::DropLate => {411+ media_objects_late = media_objects_late.saturating_add(1);412+ }413+ MediaObjectDecision::DropUntilKeyframe => {414+ media_objects_dropped = media_objects_dropped.saturating_add(1);415+ }416+ }417+ }418+ }419+ },420+ };192421 193422 if received == 0 {194423 ensure!(@@ -229,9 +458,11 @@ async fn main() -> Result<()> {
229458 ensure!(gaps == 0, "probe observed {gaps} media sequence gaps");230459 let (path_mode, path_rtt_ms) = selected_path_diagnostics(&media_connection);231460 input_send.finish().context("finishing input stream")?;232- media_send233- .finish()234- .context("finishing media request stream")?;461+ if let Some(mut media_send) = media_send {462+ media_send463+ .finish()464+ .context("finishing media request stream")?;465+ }235466 // Close both protocol connections explicitly. Relying only on endpoint236467 // teardown can leave a peer waiting for QUIC idle timeout during repeated237468 // short-lived probes, which obscures whether the host released its@@ -247,6 +478,9 @@ async fn main() -> Result<()> {
247478 println!("dimensions={width}x{height}");248479 println!("keyframes={keyframes}");249480 println!("sequence_gaps={gaps}");481+ println!("media_objects_dropped={media_objects_dropped}");482+ println!("media_objects_late={media_objects_late}");483+ println!("transport={}", media_transport.label());250484 println!("input_ack_micros={input_ack_micros}");251485 println!(252486 "pointer_smoke={}",@@ -430,4 +664,95 @@ mod tests {
430664 .contains("host did not advertise pointer surface dimensions")431665 );432666 }667+668+ fn media_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {669+ let payload = vec![0x65, 0x88, 0x84];670+ MediaFrame::new(671+ sigil_protocol::MediaFrameHeader::h264(672+ 1_280,673+ 800,674+ payload.len(),675+ sequence,676+ sequence * 1_000,677+ i64::try_from(sequence * 1_000).unwrap(),678+ flags,679+ )680+ .unwrap(),681+ payload,682+ )683+ .unwrap()684+ }685+686+ 687+ fn media_objects_wait_for_a_decodable_keyframe_then_deliver_contiguous_frames() {688+ let mut sequence = MediaObjectSequence::new();689+ let delta = media_frame(1, FrameFlags::NONE);690+ let keyframe = media_frame(2, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));691+ let next_delta = media_frame(3, FrameFlags::NONE);692+693+ assert_eq!(694+ sequence.classify(0, &delta),695+ MediaObjectDecision::DropUntilKeyframe696+ );697+ assert_eq!(698+ sequence.classify(1, &keyframe),699+ MediaObjectDecision::Deliver700+ );701+ assert_eq!(702+ sequence.classify(2, &next_delta),703+ MediaObjectDecision::Deliver704+ );705+ }706+707+ 708+ fn dropped_objects_force_resync_and_late_completions_cannot_rewind() {709+ let mut sequence = MediaObjectSequence::new();710+ let first_keyframe = media_frame(10, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));711+ let stale_delta = media_frame(11, FrameFlags::NONE);712+ let replacement_keyframe =713+ media_frame(20, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));714+715+ assert_eq!(716+ sequence.classify(0, &first_keyframe),717+ MediaObjectDecision::Deliver718+ );719+ assert!(sequence.note_drop(1));720+ assert_eq!(721+ sequence.classify(2, &replacement_keyframe),722+ MediaObjectDecision::Deliver723+ );724+ assert_eq!(725+ sequence.classify(1, &stale_delta),726+ MediaObjectDecision::DropLate727+ );728+ assert!(!sequence.note_drop(1));729+730+ let next_delta = media_frame(21, FrameFlags::NONE);731+ assert_eq!(732+ sequence.classify(3, &next_delta),733+ MediaObjectDecision::Deliver734+ );735+ }736+737+ 738+ fn media_sequence_gap_drops_deltas_until_replacement_keyframe() {739+ let mut sequence = MediaObjectSequence::new();740+ let keyframe = media_frame(40, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));741+ let gap_delta = media_frame(42, FrameFlags::NONE);742+ let replacement_keyframe =743+ media_frame(50, FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG));744+745+ assert_eq!(746+ sequence.classify(0, &keyframe),747+ MediaObjectDecision::Deliver748+ );749+ assert_eq!(750+ sequence.classify(1, &gap_delta),751+ MediaObjectDecision::DropUntilKeyframe752+ );753+ assert_eq!(754+ sequence.classify(2, &replacement_keyframe),755+ MediaObjectDecision::Deliver756+ );757+ }433758 }@@ -16,14 +16,14 @@ use clap::{Args, Parser, Subcommand, ValueEnum};
1616 use iroh::Endpoint;1717 use iroh::endpoint::{QuicTransportConfig, presets};1818 use iroh::protocol::Router;19-use sigil_protocol::{AUDIO_ALPN_V1, INPUT_ALPN_V1, MEDIA_ALPN_V1};19+use sigil_protocol::{AUDIO_ALPN_V1, INPUT_ALPN_V1, MEDIA_ALPN_V1, MEDIA_ALPN_V2};2020 use tracing::{info, warn};2121 use tracing_subscriber::EnvFilter;2222 2323 use crate::config::{HostConfig, InputMode, VideoSource};2424 use crate::cursor::PointerPositionTracker;2525 use crate::input::InputBackend;26-use crate::server::{AudioHandler, InputHandler, MediaHandler, SessionRegistry};26+use crate::server::{AudioHandler, InputHandler, MediaHandler, MediaV2Handler, SessionRegistry};2727 2828 const CONNECTION_IDLE_TIMEOUT: Duration = Duration::from_secs(5);2929 const CONNECTION_KEEP_ALIVE_INTERVAL: Duration = Duration::from_secs(1);@@ -476,6 +476,13 @@ async fn serve_command(args: ServeArgs) -> Result<()> {
476476 477477 let sessions = Arc::new(SessionRegistry::default());478478 let router = Router::builder(endpoint)479+ .accept(480+ MEDIA_ALPN_V2,481+ MediaV2Handler {482+ config: config.clone(),483+ sessions: Arc::clone(&sessions),484+ },485+ )479486 .accept(480487 MEDIA_ALPN_V1,481488 MediaHandler {@@ -6,7 +6,7 @@ use std::time::{Duration, Instant};
66 77 use anyhow::{Context, Result, bail, ensure};88 use iroh::EndpointId;9-use iroh::endpoint::Connection;9+use iroh::endpoint::{Connection, SendStream};1010 use iroh::protocol::ProtocolHandler;1111 use sigil_protocol::{1212 AUDIO_HEADER_LEN, AudioFlags, AudioPacket, AudioPacketHeader, Capability, ClientHello,@@ -29,6 +29,11 @@ const MEDIA_CAPABILITIES: &[Capability] = &[Capability::VideoH264];
2929 const AUDIO_CAPABILITIES: &[Capability] = &[Capability::AudioOpus];3030 const HANDSHAKE_TIMEOUT: Duration = Duration::from_secs(10);3131 const MEDIA_WRITE_TIMEOUT: Duration = Duration::from_secs(5);32+const MEDIA_V2_PEER_ACK_TIMEOUT: Duration = Duration::from_secs(2);33+const MEDIA_V2_IN_FLIGHT_CAPACITY: usize = 4;34+const MEDIA_V2_KEYFRAME_PRIORITY: i32 = 10;35+const MEDIA_V2_DELTA_PRIORITY: i32 = 0;36+const MEDIA_V2_RESET_CODE: u32 = 0x5356;3237 const INPUT_ACK_TIMEOUT: Duration = Duration::from_secs(1);3338 const REJECTION_DRAIN_TIMEOUT: Duration = Duration::from_secs(1);3439 const MAX_PENDING_HANDSHAKES: usize = 4;@@ -417,6 +422,140 @@ enum MediaReplayDecision {
417422 DiscardStaleSuffix { through_sequence: u64 },418423 }419424 425+426+enum MediaV2ScheduleDecision {427+ Send {428+ discontinuity: bool,429+ cancel_sequences: Vec<u64>,430+ },431+ SkipUntilKeyframe,432+ EnterResync {433+ cancel_sequences: Vec<u64>,434+ },435+}436+437+438+struct MediaV2Scheduler {439+ in_flight: Vec<u64>,440+ last_scheduled_sequence: Option<u64>,441+ waiting_for_keyframe: bool,442+ discontinuity_pending: bool,443+}444+445+impl Default for MediaV2Scheduler {446+ fn default() -> Self {447+ Self {448+ in_flight: Vec::with_capacity(MEDIA_V2_IN_FLIGHT_CAPACITY),449+ last_scheduled_sequence: None,450+ waiting_for_keyframe: true,451+ discontinuity_pending: false,452+ }453+ }454+}455+456+impl MediaV2Scheduler {457+ fn schedule(458+ &mut self,459+ sequence: u64,460+ independently_decodable: bool,461+ ) -> MediaV2ScheduleDecision {462+ if self463+ .last_scheduled_sequence464+ .is_some_and(|last| sequence <= last)465+ {466+ return MediaV2ScheduleDecision::SkipUntilKeyframe;467+ }468+469+ let sequence_discontinuity = self470+ .last_scheduled_sequence471+ .is_some_and(|last| last.checked_add(1) != Some(sequence));472+ if independently_decodable {473+ let cancel_sequences = std::mem::take(&mut self.in_flight);474+ let discontinuity = self.discontinuity_pending475+ || sequence_discontinuity476+ || !cancel_sequences.is_empty();477+ self.in_flight.push(sequence);478+ self.last_scheduled_sequence = Some(sequence);479+ self.waiting_for_keyframe = false;480+ self.discontinuity_pending = false;481+ return MediaV2ScheduleDecision::Send {482+ discontinuity,483+ cancel_sequences,484+ };485+ }486+487+ if self.waiting_for_keyframe {488+ return MediaV2ScheduleDecision::SkipUntilKeyframe;489+ }490+ if sequence_discontinuity || self.in_flight.len() == MEDIA_V2_IN_FLIGHT_CAPACITY {491+ return MediaV2ScheduleDecision::EnterResync {492+ cancel_sequences: self.enter_resync(),493+ };494+ }495+496+ self.in_flight.push(sequence);497+ self.last_scheduled_sequence = Some(sequence);498+ MediaV2ScheduleDecision::Send {499+ discontinuity: false,500+ cancel_sequences: Vec::new(),501+ }502+ }503+504+ fn complete(&mut self, sequence: u64) {505+ if let Some(index) = self.in_flight.iter().position(|value| *value == sequence) {506+ self.in_flight.swap_remove(index);507+ }508+ }509+510+ fn fail(&mut self, sequence: u64) -> Vec<u64> {511+ let Some(index) = self.in_flight.iter().position(|value| *value == sequence) else {512+ // A completion from an already-cancelled GOP must not poison the513+ // newer GOP which replaced it.514+ return Vec::new();515+ };516+ self.in_flight.swap_remove(index);517+ self.enter_resync()518+ }519+520+ fn fail_all(&mut self) -> Vec<u64> {521+ self.enter_resync()522+ }523+524+ fn enter_resync(&mut self) -> Vec<u64> {525+ self.waiting_for_keyframe = true;526+ self.discontinuity_pending = true;527+ std::mem::take(&mut self.in_flight)528+ }529+}530+531+struct ResetOnDropSendStream(Option<SendStream>);532+533+impl ResetOnDropSendStream {534+ fn new(stream: SendStream) -> Self {535+ Self(Some(stream))536+ }537+538+ fn stream(&self) -> &SendStream {539+ self.0.as_ref().expect("send stream guard is armed")540+ }541+542+ fn stream_mut(&mut self) -> &mut SendStream {543+ self.0.as_mut().expect("send stream guard is armed")544+ }545+546+ fn disarm(&mut self) {547+ self.0.take();548+ }549+}550+551+impl Drop for ResetOnDropSendStream {552+ fn drop(&mut self) {553+ if let Some(stream) = self.0.as_mut() {554+ let _ = stream.reset(MEDIA_V2_RESET_CODE.into());555+ }556+ }557+}558+420559 421560 struct MediaReplayCursor {422561 last_sequence: Option<u64>,@@ -519,6 +658,12 @@ pub struct MediaHandler {
519658 pub sessions: Arc<SessionRegistry>,520659 }521660 661+662+pub struct MediaV2Handler {663+ pub config: HostConfig,664+ pub sessions: Arc<SessionRegistry>,665+}666+522667 impl ProtocolHandler for MediaHandler {523668 async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {524669 let remote = connection.remote_id();@@ -529,6 +674,16 @@ impl ProtocolHandler for MediaHandler {
529674 }530675 }531676 677+impl ProtocolHandler for MediaV2Handler {678+ async fn accept(&self, connection: Connection) -> Result<(), iroh::protocol::AcceptError> {679+ let remote = connection.remote_id();680+ if let Err(error) = serve_media_v2(connection, self.config.clone(), &self.sessions).await {681+ warn!(%remote, %error, "media v2 connection ended");682+ }683+ Ok(())684+ }685+}686+532687 533688 pub struct InputHandler {534689 pub backend: InputBackend,@@ -726,6 +881,288 @@ async fn serve_media(
726881 session_result727882 }728883 884+async fn serve_media_v2(885+ connection: Connection,886+ config: HostConfig,887+ sessions: &Arc<SessionRegistry>,888+) -> Result<()> {889+ let remote = connection.remote_id();890+ let handshake_permit = sessions891+ .pending_handshakes892+ .try_acquire()893+ .context("too many pending handshakes")?;894+ let (mut send, mut recv) = tokio::time::timeout(HANDSHAKE_TIMEOUT, connection.accept_bi())895+ .await896+ .context("timed out accepting media v2 handshake stream")?897+ .context("accepting media v2 handshake stream")?;898+ let hello = receive_hello(&mut recv, Capability::VideoH264).await?;899+ drop(handshake_permit);900+ debug!(%remote, agent = %hello.agent, "media v2 hello received");901+902+ let lease = match sessions.claim(remote, hello.nonce) {903+ Ok(lease) => lease,904+ Err(error) => {905+ send_rejection(&mut send, "host already has an active client").await?;906+ return Err(error);907+ }908+ };909+910+ let source = match config.source {911+ VideoSource::TestPattern => Ok(spawn_test_pattern(config.clone(), lease.session_clock)),912+ VideoSource::GamescopePipewire => {913+ let primary = spawn_gamescope_pipewire_after_static_preflight(914+ config.clone(),915+ lease.session_clock,916+ )917+ .await?;918+ Ok(select_gamescope_startup_source(config.clone(), lease.session_clock, primary).await)919+ }920+ };921+ let EncodedSource {922+ frames: _frame_receiver,923+ current_gop: mut current_gop_receiver,924+ task: source_task,925+ pointer_surface_dimensions,926+ } = match source {927+ Ok(source) => source,928+ Err(error) => {929+ send_rejection(&mut send, "video source is unavailable").await?;930+ return Err(error);931+ }932+ };933+ let source_task = SourceTaskGuard::new(source_task);934+935+ let mut media_hello = HostHello::accepted(936+ lease.session_id,937+ negotiated_capabilities(&hello, MEDIA_CAPABILITIES),938+ );939+ if let Some(dimensions) = pointer_surface_dimensions {940+ media_hello = media_hello.with_pointer_surface_dimensions(dimensions);941+ }942+ write_host_hello(&mut send, &media_hello).await?;943+ send.finish()944+ .context("finishing media v2 handshake response")?;945+ drop(send);946+ drop(recv);947+ info!(%remote, session_id = lease.session_id, "media v2 client accepted");948+949+ let session_result =950+ run_media_v2_session(&connection, &config, &mut current_gop_receiver, remote).await;951+952+ source_task.abort_and_wait().await;953+ drop(lease);954+ info!(%remote, "media v2 client released");955+ session_result956+}957+958+async fn run_media_v2_session(959+ connection: &Connection,960+ config: &HostConfig,961+ current_gop_receiver: &mut tokio::sync::watch::Receiver<Option<EncodedGop>>,962+ remote: EndpointId,963+) -> Result<()> {964+ let maximum_replay_age = maximum_media_replay_age(config.framerate);965+ let mut replay_cursor = MediaReplayCursor::default();966+ let mut scheduler = MediaV2Scheduler::default();967+ let mut send_tasks = tokio::task::JoinSet::new();968+969+ let result = loop {970+ tokio::select! {971+ closed = connection.closed() => {972+ debug!(%remote, ?closed, "media v2 connection closed");973+ break Ok(());974+ }975+ task = send_tasks.join_next(), if !send_tasks.is_empty() => {976+ match task.expect("guarded by non-empty send task set") {977+ Ok((sequence, Ok(()))) => scheduler.complete(sequence),978+ Ok((sequence, Err(error))) => {979+ warn!(sequence, %error, "media v2 object send failed; waiting for keyframe");980+ if !scheduler.fail(sequence).is_empty() {981+ send_tasks.abort_all();982+ }983+ }984+ Err(error) if error.is_cancelled() => {}985+ Err(error) => {986+ warn!(%error, "media v2 object task failed; waiting for keyframe");987+ if !scheduler.fail_all().is_empty() {988+ send_tasks.abort_all();989+ }990+ }991+ }992+ }993+ changed = current_gop_receiver.changed() => {994+ if let Err(error) = changed {995+ break Err(error).context("encoded source stopped");996+ }997+ let Some(current_gop) = current_gop_receiver.borrow_and_update().clone() else {998+ continue;999+ };1000+ let initial_replay_started_at =1001+ replay_cursor.last_sequence.is_none().then(Instant::now);1002+ let replay_through_sequence = current_gop1003+ .frames1004+ .last()1005+ .map(|frame| frame.sequence)1006+ .context("current GOP snapshot is empty")?;1007+1008+ for frame in new_current_gop_frames(current_gop, replay_cursor.last_sequence) {1009+ let replay_discontinuity = match replay_cursor.classify(1010+ &frame,1011+ replay_through_sequence,1012+ initial_replay_started_at,1013+ Instant::now(),1014+ maximum_replay_age,1015+ ) {1016+ MediaReplayDecision::Send { discontinuity } => discontinuity,1017+ MediaReplayDecision::SkipUntilKeyframe => continue,1018+ MediaReplayDecision::DiscardStaleSuffix { .. } => break,1019+ };1020+ let independently_decodable = frame.keyframe && frame.codec_config;1021+ let (scheduler_discontinuity, cancel_sequences) =1022+ match scheduler.schedule(frame.sequence, independently_decodable) {1023+ MediaV2ScheduleDecision::Send {1024+ discontinuity,1025+ cancel_sequences,1026+ } => (discontinuity, cancel_sequences),1027+ MediaV2ScheduleDecision::SkipUntilKeyframe => continue,1028+ MediaV2ScheduleDecision::EnterResync { cancel_sequences } => {1029+ if !cancel_sequences.is_empty() {1030+ send_tasks.abort_all();1031+ }1032+ continue;1033+ }1034+ };1035+ if !cancel_sequences.is_empty() {1036+ debug!(1037+ sequence = frame.sequence,1038+ ?cancel_sequences,1039+ "keyframe superseding media v2 objects"1040+ );1041+ send_tasks.abort_all();1042+ }1043+1044+ let media_frame = media_frame_for_encoded(1045+ config,1046+ &frame,1047+ replay_discontinuity || scheduler_discontinuity,1048+ )?;1049+ let sequence = frame.sequence;1050+ let keyframe = independently_decodable;1051+ // Reserve stream IDs in encoded-frame order. Opening them1052+ // inside the concurrent writer tasks would let task1053+ // scheduling reorder the QUIC object sequence even though1054+ // each object carries a monotonic media sequence number.1055+ let stream = match tokio::time::timeout(1056+ MEDIA_WRITE_TIMEOUT,1057+ connection.open_uni(),1058+ )1059+ .await1060+ {1061+ Ok(Ok(stream)) => stream,1062+ Ok(Err(error)) => {1063+ warn!(sequence, %error, "opening media v2 object stream failed");1064+ if !scheduler.fail(sequence).is_empty() {1065+ send_tasks.abort_all();1066+ }1067+ continue;1068+ }1069+ Err(_) => {1070+ warn!(sequence, "opening media v2 object stream timed out");1071+ if !scheduler.fail(sequence).is_empty() {1072+ send_tasks.abort_all();1073+ }1074+ continue;1075+ }1076+ };1077+ let stream = ResetOnDropSendStream::new(stream);1078+ if let Err(error) = stream1079+ .stream()1080+ .set_priority(media_v2_priority(keyframe))1081+ {1082+ warn!(sequence, %error, "setting media v2 object priority failed");1083+ if !scheduler.fail(sequence).is_empty() {1084+ send_tasks.abort_all();1085+ }1086+ continue;1087+ }1088+ send_tasks.spawn(async move {1089+ (1090+ sequence,1091+ send_media_v2_object(stream, media_frame).await,1092+ )1093+ });1094+ replay_cursor.commit_sent(&frame);1095+ }1096+ }1097+ }1098+ };1099+1100+ send_tasks.abort_all();1101+ while send_tasks.join_next().await.is_some() {}1102+ result1103+}1104+1105+fn media_frame_for_encoded(1106+ config: &HostConfig,1107+ frame: &EncodedFrame,1108+ discontinuity: bool,1109+) -> Result<MediaFrame> {1110+ let mut flags = FrameFlags::NONE;1111+ if frame.keyframe {1112+ flags = flags.union(FrameFlags::KEYFRAME);1113+ }1114+ if frame.codec_config {1115+ flags = flags.union(FrameFlags::CODEC_CONFIG);1116+ }1117+ if discontinuity {1118+ flags = flags.union(FrameFlags::DISCONTINUITY);1119+ }1120+ let width = u16::try_from(config.width).context("width exceeds protocol")?;1121+ let height = u16::try_from(config.height).context("height exceeds protocol")?;1122+ let header = MediaFrameHeader::h264(1123+ width,1124+ height,1125+ frame.data.len(),1126+ frame.sequence,1127+ frame.capture_timestamp_micros,1128+ frame.presentation_timestamp_micros,1129+ flags,1130+ )?;1131+ MediaFrame::new(header, frame.data.as_ref().to_vec()).map_err(Into::into)1132+}1133+1134+async fn send_media_v2_object(mut stream: ResetOnDropSendStream, frame: MediaFrame) -> Result<()> {1135+ tokio::time::timeout(1136+ MEDIA_WRITE_TIMEOUT,1137+ write_media_frame(stream.stream_mut(), &frame),1138+ )1139+ .await1140+ .context("timed out writing media v2 object")??;1141+ stream1142+ .stream_mut()1143+ .finish()1144+ .context("finishing media v2 object stream")?;1145+ match tokio::time::timeout(MEDIA_V2_PEER_ACK_TIMEOUT, stream.stream().stopped())1146+ .await1147+ .context("timed out waiting for media v2 object acknowledgement")?1148+ .context("waiting for media v2 object acknowledgement")?1149+ {1150+ None => {1151+ stream.disarm();1152+ Ok(())1153+ }1154+ Some(code) => bail!("peer stopped media v2 object stream with code {code}"),1155+ }1156+}1157+1158+fn media_v2_priority(keyframe: bool) -> i32 {1159+ if keyframe {1160+ MEDIA_V2_KEYFRAME_PRIORITY1161+ } else {1162+ MEDIA_V2_DELTA_PRIORITY1163+ }1164+}1165+7291166 fn new_current_gop_frames(7301167 current_gop: EncodedGop,7311168 last_sequence: Option<u64>,@@ -1121,6 +1558,127 @@ mod tests {
11211558 }11221559 }11231560 1561+ 1562+ fn media_v2_scheduler_is_bounded_and_drops_a_saturated_delta_suffix() {1563+ let mut scheduler = MediaV2Scheduler::default();1564+ assert_eq!(1565+ scheduler.schedule(10, true),1566+ MediaV2ScheduleDecision::Send {1567+ discontinuity: false,1568+ cancel_sequences: vec![],1569+ }1570+ );1571+ for sequence in 11..14 {1572+ assert!(matches!(1573+ scheduler.schedule(sequence, false),1574+ MediaV2ScheduleDecision::Send { .. }1575+ ));1576+ }1577+ assert_eq!(scheduler.in_flight.len(), MEDIA_V2_IN_FLIGHT_CAPACITY);1578+1579+ assert_eq!(1580+ scheduler.schedule(14, false),1581+ MediaV2ScheduleDecision::EnterResync {1582+ cancel_sequences: vec![10, 11, 12, 13],1583+ }1584+ );1585+ assert!(scheduler.in_flight.is_empty());1586+ assert_eq!(1587+ scheduler.schedule(15, false),1588+ MediaV2ScheduleDecision::SkipUntilKeyframe1589+ );1590+ }1591+1592+ 1593+ fn blocked_old_streams_cannot_prevent_a_keyframe_decision() {1594+ let mut scheduler = MediaV2Scheduler::default();1595+ for sequence in 20..24 {1596+ assert!(matches!(1597+ scheduler.schedule(sequence, sequence == 20),1598+ MediaV2ScheduleDecision::Send { .. }1599+ ));1600+ }1601+ assert_eq!(scheduler.in_flight.len(), MEDIA_V2_IN_FLIGHT_CAPACITY);1602+1603+ assert_eq!(1604+ scheduler.schedule(30, true),1605+ MediaV2ScheduleDecision::Send {1606+ discontinuity: true,1607+ cancel_sequences: vec![20, 21, 22, 23],1608+ }1609+ );1610+ assert_eq!(scheduler.in_flight, vec![30]);1611+ }1612+1613+ 1614+ fn media_v2_send_failure_cancels_dependents_and_resyncs_on_keyframe() {1615+ let mut scheduler = MediaV2Scheduler::default();1616+ assert!(matches!(1617+ scheduler.schedule(40, true),1618+ MediaV2ScheduleDecision::Send { .. }1619+ ));1620+ assert!(matches!(1621+ scheduler.schedule(41, false),1622+ MediaV2ScheduleDecision::Send { .. }1623+ ));1624+1625+ assert_eq!(scheduler.fail(40), vec![41]);1626+ assert_eq!(1627+ scheduler.schedule(42, false),1628+ MediaV2ScheduleDecision::SkipUntilKeyframe1629+ );1630+ assert_eq!(1631+ scheduler.schedule(50, true),1632+ MediaV2ScheduleDecision::Send {1633+ discontinuity: true,1634+ cancel_sequences: vec![],1635+ }1636+ );1637+ scheduler.complete(50);1638+ assert!(scheduler.in_flight.is_empty());1639+ assert!(matches!(1640+ scheduler.schedule(51, false),1641+ MediaV2ScheduleDecision::Send {1642+ discontinuity: false,1643+ ..1644+ }1645+ ));1646+ }1647+1648+ 1649+ fn cancelled_old_completion_cannot_poison_the_replacement_gop() {1650+ let mut scheduler = MediaV2Scheduler::default();1651+ assert!(matches!(1652+ scheduler.schedule(60, true),1653+ MediaV2ScheduleDecision::Send { .. }1654+ ));1655+ assert!(matches!(1656+ scheduler.schedule(61, false),1657+ MediaV2ScheduleDecision::Send { .. }1658+ ));1659+ assert!(matches!(1660+ scheduler.schedule(70, true),1661+ MediaV2ScheduleDecision::Send {1662+ discontinuity: true,1663+ ..1664+ }1665+ ));1666+1667+ assert!(scheduler.fail(60).is_empty());1668+ assert!(matches!(1669+ scheduler.schedule(71, false),1670+ MediaV2ScheduleDecision::Send {1671+ discontinuity: false,1672+ ..1673+ }1674+ ));1675+ }1676+1677+ 1678+ fn media_v2_keyframes_have_strictly_higher_transport_priority() {1679+ assert!(media_v2_priority(true) > media_v2_priority(false));1680+ }1681+11241682 fn endpoint(byte: u8) -> EndpointId {11251683 iroh::SecretKey::from_bytes(&[byte; 32]).public()11261684 }@@ -27,7 +27,7 @@ pub use input::{
2727 };2828 pub use media::{2929 FrameFlags, MEDIA_HEADER_LEN, MediaCodec, MediaFrame, MediaFrameHeader, read_media_frame,30- write_media_frame,30+ read_media_object, write_media_frame,3131 };3232 3333 /// Protocol version encoded in v1 messages.@@ -35,6 +35,8 @@ pub const PROTOCOL_VERSION: u16 = 1;
3535 3636 /// ALPN for the v1 encoded video stream.3737 pub const MEDIA_ALPN_V1: &[u8] = b"sigil/media/1";38+/// ALPN for one encoded video object per host-opened unidirectional stream.39+pub const MEDIA_ALPN_V2: &[u8] = b"sigil/media/2";3840 /// ALPN for the v1 latency-independent input stream.3941 pub const INPUT_ALPN_V1: &[u8] = b"sigil/input/1";4042 /// ALPN for the v1 session-control stream.@@ -43,7 +45,8 @@ pub const CONTROL_ALPN_V1: &[u8] = b"sigil/control/1";
4345 pub const AUDIO_ALPN_V1: &[u8] = b"sigil/audio/1";4446 4547 /// The inherited frame-stream ALPN, provided only for an explicit migration46-/// adapter. New peers must advertise [`MEDIA_ALPN_V1`].48+/// adapter. New peers should prefer [`MEDIA_ALPN_V2`] and may retain49+/// [`MEDIA_ALPN_V1`] only as a compatibility fallback.4750 pub const LEGACY_FRAME_ALPN_V0: &[u8] = b"sigil/frame-stream/0";4851 /// The inherited input-stream ALPN, provided only for an explicit migration4952 /// adapter. New peers must advertise [`INPUT_ALPN_V1`].@@ -72,6 +75,7 @@ mod tests {
7275 7376 fn alpn_values_are_deployment_golden_vectors() {7477 assert_eq!(MEDIA_ALPN_V1, b"sigil/media/1");78+ assert_eq!(MEDIA_ALPN_V2, b"sigil/media/2");7579 assert_eq!(INPUT_ALPN_V1, b"sigil/input/1");7680 assert_eq!(CONTROL_ALPN_V1, b"sigil/control/1");7781 assert_eq!(AUDIO_ALPN_V1, b"sigil/audio/1");@@ -240,6 +240,31 @@ where
240240 Ok(Some(MediaFrame { header, payload }))241241 }242242 243+/// Read exactly one complete frame followed by a clean stream EOF.244+///245+/// This is the object boundary used by the v2 media ALPN. An empty stream or246+/// any bytes after the declared frame are rejected so one QUIC stream can247+/// never be interpreted as zero or multiple media objects.248+pub async fn read_media_object<R>(reader: &mut R) -> Result<MediaFrame>249+where250+ R: AsyncRead + Unpin,251+{252+ let frame = read_media_frame(reader)253+ .await?254+ .ok_or(ProtocolError::InvalidMessage {255+ message_type: "media object",256+ reason: "stream ended before the media frame",257+ })?;258+ let mut trailing = [0_u8; 1];259+ if reader.read(&mut trailing).await? != 0 {260+ return Err(ProtocolError::InvalidMessage {261+ message_type: "media object",262+ reason: "trailing bytes after the media frame",263+ });264+ }265+ Ok(frame)266+}267+243268 /// Validate and write one complete frame.244269 pub async fn write_media_frame<W>(writer: &mut W, frame: &MediaFrame) -> Result<()>245270 where@@ -351,6 +376,42 @@ mod tests {
351376 assert_eq!(read_media_frame(&mut receiver).await.unwrap(), None);352377 }353378 379+ 380+ async fn media_object_requires_exactly_one_frame_and_clean_eof() {381+ let expected = MediaFrame::new(header(), vec![0, 0, 0, 1]).unwrap();382+ let (mut sender, mut receiver) = duplex(256);383+ write_media_frame(&mut sender, &expected).await.unwrap();384+ sender.shutdown().await.unwrap();385+386+ assert_eq!(read_media_object(&mut receiver).await.unwrap(), expected);387+ }388+389+ 390+ async fn media_object_rejects_empty_and_trailing_streams() {391+ let (mut empty_sender, mut empty_receiver) = duplex(16);392+ empty_sender.shutdown().await.unwrap();393+ assert!(matches!(394+ read_media_object(&mut empty_receiver).await,395+ Err(ProtocolError::InvalidMessage {396+ message_type: "media object",397+ ..398+ })399+ ));400+401+ let frame = MediaFrame::new(header(), vec![0, 0, 0, 1]).unwrap();402+ let (mut sender, mut receiver) = duplex(256);403+ write_media_frame(&mut sender, &frame).await.unwrap();404+ sender.write_all(&[0xff]).await.unwrap();405+ sender.shutdown().await.unwrap();406+ assert!(matches!(407+ read_media_object(&mut receiver).await,408+ Err(ProtocolError::InvalidMessage {409+ message_type: "media object",410+ ..411+ })412+ ));413+ }414+354415 355416 async fn truncated_payload_is_an_io_error() {356417 let (mut sender, mut receiver) = duplex(256);@@ -13,6 +13,10 @@ probe_timeout_seconds=15
1313 command_timeout_seconds=451414 readiness_timeout_seconds=301515 host_runtime_seconds=12016+media_transport="independent-v2"17+media_accept_log="media v2 client accepted"18+media_release_log="media v2 client released"19+media_v1=false1620 1721 usage() {1822 cat <<'EOF'@@ -28,6 +32,8 @@ Options:
2832 --reconnect-cycles COUNT Fresh sessions after the primary (default: 3)2933 --reconnect-frames COUNT Frames per reconnect session (default: 30)3034 --probe-timeout-seconds SEC Probe per-operation timeout (default: 15)35+ --media-v1 Exercise ordered media v1 compatibility36+ instead of independent media v23137 --help Show this help3238 EOF3339 }@@ -75,6 +81,13 @@ while [[ $# -gt 0 ]]; do
7581 probe_timeout_seconds="$2"7682 shift 27783 ;;84+ --media-v1)85+ media_transport="reliable-v1"86+ media_accept_log="media client accepted"87+ media_release_log="media client released"88+ media_v1=true89+ shift90+ ;;7891 --help|-h)7992 usage8093 exit 0@@ -298,12 +311,25 @@ validate_probe_log() {
298311 grep -Fxq "frames=${expected_frames}" "$log_path" || return 1299312 grep -Fxq 'dimensions=1280x800' "$log_path" || return 1300313 grep -Fxq 'sequence_gaps=0' "$log_path" || return 1314+ grep -Fxq "transport=${media_transport}" "$log_path" || return 1315+ awk -F= '$1 == "media_objects_dropped" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \316+ "$log_path" || return 1317+ awk -F= '$1 == "media_objects_late" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \318+ "$log_path" || return 1301319 awk -F= '$1 == "keyframes" && $2 ~ /^[0-9]+$/ && $2 > 0 { found=1 } END { exit !found }' \302320 "$log_path" || return 1303321 awk -F= '$1 == "input_ack_micros" && $2 ~ /^[0-9]+$/ { found=1 } END { exit !found }' \304322 "$log_path"305323 }306324 325+run_probe() {326+ if [[ "$media_v1" == true ]]; then327+ "$probe_bin" --media-v1 "$@"328+ else329+ "$probe_bin" "$@"330+ fi331+}332+307333 cd "$workspace_dir"308334 cargo build --locked --profile "$cargo_profile" -p sigil-host --bins309335 @@ -344,7 +370,7 @@ wait_for_log_line "$host_log" 'status=ready' "$host_pid" 'sigil-host' || die "ho
344370 host_node_id="$(sed -n 's/^node_id=//p' "$host_log" | tail -n 1)"345371 [[ "$host_node_id" == "$node_id" ]] || die "ready host node ID does not match its identity"346372 347-"$probe_bin" \373+run_probe \348374 --node-id "$node_id" \349375 --frames "$primary_frames" \350376 --timeout-seconds "$probe_timeout_seconds" \@@ -354,14 +380,14 @@ primary_pid=$!
354380 start_watchdog "$primary_pid" "$command_timeout_seconds" "$tmp_root/primary.timeout"355381 primary_watchdog_pid="$watchdog_pid"356382 357-wait_for_log_count "$host_log" 'media client accepted' 1 "$primary_pid" 'primary probe' \383+wait_for_log_count "$host_log" "$media_accept_log" 1 "$primary_pid" 'primary probe' \358384 || die "primary session did not become active"359385 if ! kill -0 "$primary_pid" 2>/dev/null; then360386 die "primary probe ended before the concurrent rejection check"361387 fi362388 363389 if run_bounded "$probe_timeout_seconds" "$secondary_log" \364- "$probe_bin" --node-id "$node_id" --frames 1 \390+ run_probe --node-id "$node_id" --frames 1 \365391 --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800; then366392 die "secondary client was accepted while the primary was active"367393 fi@@ -392,7 +418,7 @@ validate_probe_log "$primary_log" "$primary_frames" || {
392418 die "primary probe evidence is incomplete"393419 }394420 395-wait_for_log_count "$host_log" 'media client released' 1 "$host_pid" 'sigil-host' \421+wait_for_log_count "$host_log" "$media_release_log" 1 "$host_pid" 'sigil-host' \396422 || die "primary session was not released"397423 wait_for_log_count "$host_log" 'input client released' 1 "$host_pid" 'sigil-host' \398424 || die "primary input session was not drained"@@ -401,7 +427,7 @@ cycle=1
401427 while [[ "$cycle" -le "$reconnect_cycles" ]]; do402428 reconnect_log="$tmp_root/reconnect-${cycle}.log"403429 if ! run_bounded "$command_timeout_seconds" "$reconnect_log" \404- "$probe_bin" --node-id "$node_id" --frames "$reconnect_frames" \430+ run_probe --node-id "$node_id" --frames "$reconnect_frames" \405431 --timeout-seconds "$probe_timeout_seconds" --expect-size 1280x800; then406432 sed -n '1,240p' "$reconnect_log" >&2 || true407433 die "reconnect cycle $cycle failed"@@ -410,7 +436,7 @@ while [[ "$cycle" -le "$reconnect_cycles" ]]; do
410436 sed -n '1,240p' "$reconnect_log" >&2 || true411437 die "reconnect cycle $cycle evidence is incomplete"412438 }413- wait_for_log_count "$host_log" 'media client released' "$((cycle + 1))" "$host_pid" 'sigil-host' \439+ wait_for_log_count "$host_log" "$media_release_log" "$((cycle + 1))" "$host_pid" 'sigil-host' \414440 || die "reconnect cycle $cycle was not released"415441 wait_for_log_count "$host_log" 'input client released' "$((cycle + 1))" "$host_pid" 'sigil-host' \416442 || die "reconnect input cycle $cycle was not drained"@@ -441,7 +467,7 @@ printf 'probe_binary=%s\n' "$probe_bin"
441467 printf 'probe_sha256=%s\n' "$(sha256_file "$probe_bin")"442468 printf 'node_id=%s\n' "$node_id"443469 printf 'primary_frames=%s\n' "$primary_frames"444-grep -E '^(keyframes|sequence_gaps|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"470+grep -E '^(transport|keyframes|sequence_gaps|media_objects_dropped|media_objects_late|input_ack_micros|path_mode|path_rtt_ms)=' "$primary_log"445471 printf 'active_client_rejection=ok\n'446472 printf 'reconnect_cycles=%s\n' "$reconnect_cycles"447473 printf 'cleanup=ok\n'@@ -10,9 +10,10 @@ use serde::Serialize;
1010 use sigil_protocol::{1111 AUDIO_ALPN_V1, AudioFlags, AudioPacket, AudioPacketHeader, Capability, ClientHello, FrameFlags,1212 INPUT_ALPN_V1, InputEvent, MAX_MEDIA_PAYLOAD_LEN, MAX_VIDEO_DIMENSION, MAX_VIDEO_PIXELS,13- MEDIA_ALPN_V1, MediaCodec, PointerPosition, PointerSurfaceDimensions,14- RELATIVE_POINTER_DELTA_MAX, RELATIVE_POINTER_DELTA_MIN, read_host_hello, read_input_ack,15- read_media_frame, write_client_hello, write_input_event,13+ MEDIA_ALPN_V1, MEDIA_ALPN_V2, MediaCodec, MediaFrame, PointerPosition,14+ PointerSurfaceDimensions, ProtocolError, RELATIVE_POINTER_DELTA_MAX,15+ RELATIVE_POINTER_DELTA_MIN, read_host_hello, read_input_ack, read_media_frame,16+ read_media_object, write_client_hello, write_input_event,1617 };1718 use std::collections::{BTreeMap, VecDeque};1819 use std::io::Cursor;@@ -65,6 +66,8 @@ fn emit_frame_error(app: &AppHandle, generation: u64, error: impl Into<String>)
6566 }6667 6768 const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);69+const CLIENT_MEDIA_OBJECT_CAPACITY: usize = 4;70+const CLIENT_MEDIA_OBJECT_READ_TIMEOUT: Duration = Duration::from_secs(1);6871 // Absorb brief webview→Rust acknowledgment jitter without allowing Tauri IPC6972 // to grow without bound. Four 60 fps frames cap this handoff at about 67 ms;7073 // WebCodecs has a separate, stricter decode-queue bound in the frontend.@@ -206,6 +209,8 @@ struct ClientMediaMetrics {
206209 transport_received_total: u64,207210 frontend_sent_total: u64,208211 sequence_dropped_total: u64,212+ transport_object_dropped_total: u64,213+ transport_late_object_dropped_total: u64,209214 frontend_queue_dropped_total: u64,210215 frontend_resync_dropped_total: u64,211216 frontend_queue_peak: usize,@@ -250,6 +255,14 @@ impl ClientMediaMetrics {
250255 self.sequence_dropped_total = self.sequence_dropped_total.saturating_add(count);251256 }252257 258+ fn observe_transport_object_drop(&mut self, late: bool) {259+ self.transport_object_dropped_total = self.transport_object_dropped_total.saturating_add(1);260+ if late {261+ self.transport_late_object_dropped_total =262+ self.transport_late_object_dropped_total.saturating_add(1);263+ }264+ }265+253266 fn observe_frontend_queue_drop(&mut self) {254267 self.frontend_queue_dropped_total = self.frontend_queue_dropped_total.saturating_add(1);255268 }@@ -311,12 +324,14 @@ impl ClientMediaMetrics {
311324 .max(frontend_resync_current_duration.unwrap_or_default());312325 FrameStatsPayload {313326 generation,314- stats_version: 2,327+ stats_version: 3,315328 transport_receive_fps,316329 frontend_send_fps,317330 transport_received_total: self.transport_received_total,318331 frontend_sent_total: self.frontend_sent_total,319332 sequence_dropped_total: self.sequence_dropped_total,333+ transport_object_dropped_total: self.transport_object_dropped_total,334+ transport_late_object_dropped_total: self.transport_late_object_dropped_total,320335 frontend_queue_dropped_total: self.frontend_queue_dropped_total,321336 frontend_resync_dropped_total: self.frontend_resync_dropped_total,322337 frontend_dropped_total,@@ -381,6 +396,8 @@ struct FrameStatsPayload {
381396 transport_received_total: u64,382397 frontend_sent_total: u64,383398 sequence_dropped_total: u64,399+ transport_object_dropped_total: u64,400+ transport_late_object_dropped_total: u64,384401 frontend_queue_dropped_total: u64,385402 frontend_resync_dropped_total: u64,386403 frontend_dropped_total: u64,@@ -611,6 +628,7 @@ pub struct ConnectResult {
611628 pub connected: bool,612629 pub host_node_id: Option<String>,613630 pub development_mode: bool,631+ pub media_transport: &'static str,614632 pub pointer_surface_dimensions: Option<PointerSurfaceDimensions>,615633 pub relative_pointer_available: bool,616634 pub pointer_position_feedback_available: bool,@@ -632,6 +650,171 @@ struct NegotiatedV1Stream {
632650 pointer_surface_dimensions: Option<PointerSurfaceDimensions>,633651 }634652 653+654+enum MediaTransport {655+ LegacyV0,656+ ReliableStreamV1,657+ IndependentObjectsV2,658+}659+660+impl MediaTransport {661+ const fn diagnostic_name(self) -> &'static str {662+ match self {663+ Self::LegacyV0 => "reliable-v0",664+ Self::ReliableStreamV1 => "reliable-v1",665+ Self::IndependentObjectsV2 => "independent-v2",666+ }667+ }668+}669+670+671+enum MediaObjectReadOutcome {672+ Frame {673+ object_index: u64,674+ frame: MediaFrame,675+ },676+ Dropped {677+ object_index: u64,678+ },679+ Malformed(String),680+}681+682+struct MediaObjectReceiver {683+ connection: iroh::endpoint::Connection,684+ reads: tokio::task::JoinSet<MediaObjectReadOutcome>,685+ next_object_index: u64,686+ connection_closed: bool,687+}688+689+impl MediaObjectReceiver {690+ fn new(connection: iroh::endpoint::Connection) -> Self {691+ Self {692+ connection,693+ reads: tokio::task::JoinSet::new(),694+ next_object_index: 0,695+ connection_closed: false,696+ }697+ }698+699+ async fn next(&mut self) -> Result<Option<MediaObjectReadOutcome>, String> {700+ loop {701+ if self.connection_closed && self.reads.is_empty() {702+ return Ok(None);703+ }704+705+ tokio::select! {706+ biased;707+ completed = self.reads.join_next(), if !self.reads.is_empty() => {708+ let completed = completed709+ .ok_or_else(|| "Media object reader ended unexpectedly".to_string())?710+ .map_err(|error| format!("Media object reader task failed: {error}"))?;711+ return Ok(Some(completed));712+ }713+ accepted = self.connection.accept_uni(), if !self.connection_closed714+ && self.reads.len() < CLIENT_MEDIA_OBJECT_CAPACITY => {715+ let mut stream = match accepted {716+ Ok(stream) => stream,717+ Err(_) => {718+ self.connection_closed = true;719+ continue;720+ }721+ };722+ self.next_object_index = self.next_object_index.checked_add(1)723+ .ok_or_else(|| "Media object index overflowed".to_string())?;724+ let object_index = self.next_object_index;725+ self.reads.spawn(async move {726+ match tokio::time::timeout(727+ CLIENT_MEDIA_OBJECT_READ_TIMEOUT,728+ read_media_object(&mut stream),729+ )730+ .await731+ {732+ Err(_) => MediaObjectReadOutcome::Dropped { object_index },733+ Ok(Err(ProtocolError::Io(_))) => {734+ MediaObjectReadOutcome::Dropped { object_index }735+ }736+ Ok(Err(error)) => {737+ MediaObjectReadOutcome::Malformed(format!(738+ "Invalid media object: {error}"739+ ))740+ }741+ Ok(Ok(frame)) => MediaObjectReadOutcome::Frame {742+ object_index,743+ frame,744+ },745+ }746+ });747+ }748+ }749+ }750+ }751+}752+753+impl Drop for MediaObjectReceiver {754+ fn drop(&mut self) {755+ self.reads.abort_all();756+ }757+}758+759+760+enum MediaObjectSequenceDecision {761+ Deliver { discontinuity: bool },762+ DropLate,763+ DropUntilKeyframe,764+}765+766+767+struct MediaObjectSequence {768+ last_sequence: Option<u64>,769+ last_object_index: u64,770+ waiting_for_keyframe: bool,771+}772+773+impl MediaObjectSequence {774+ fn new() -> Self {775+ Self {776+ waiting_for_keyframe: true,777+ ..Self::default()778+ }779+ }780+781+ fn note_dropped_object(&mut self, object_index: u64) -> bool {782+ if object_index <= self.last_object_index {783+ return false;784+ }785+ self.waiting_for_keyframe = true;786+ true787+ }788+789+ fn classify(&mut self, object_index: u64, frame: &MediaFrame) -> MediaObjectSequenceDecision {790+ if object_index <= self.last_object_index791+ || self792+ .last_sequence793+ .is_some_and(|sequence| frame.header.sequence <= sequence)794+ {795+ return MediaObjectSequenceDecision::DropLate;796+ }797+798+ let keyframe = frame.header.flags.contains(FrameFlags::KEYFRAME)799+ && frame.header.flags.contains(FrameFlags::CODEC_CONFIG);800+ let sequence_contiguous = self801+ .last_sequence802+ .is_none_or(|sequence| sequence.checked_add(1) == Some(frame.header.sequence));803+ if !keyframe && (self.waiting_for_keyframe || !sequence_contiguous) {804+ self.waiting_for_keyframe = true;805+ return MediaObjectSequenceDecision::DropUntilKeyframe;806+ }807+808+ let discontinuity = frame.header.flags.contains(FrameFlags::DISCONTINUITY)809+ || self.waiting_for_keyframe810+ || !sequence_contiguous;811+ self.last_sequence = Some(frame.header.sequence);812+ self.last_object_index = object_index;813+ self.waiting_for_keyframe = false;814+ MediaObjectSequenceDecision::Deliver { discontinuity }815+ }816+}817+635818 636819 struct InputAvailability {637820 relative_pointer: bool,@@ -807,6 +990,77 @@ async fn open_negotiated_input_stream(
807990 Ok((send, recv, negotiation))808991 }809992 993+async fn open_negotiated_media_stream(994+ endpoint: &Endpoint,995+ address: &iroh::EndpointAddr,996+ nonce: [u8; 16],997+) -> Result<998+ (999+ iroh::endpoint::Connection,1000+ iroh::endpoint::RecvStream,1001+ NegotiatedV1Stream,1002+ MediaTransport,1003+ ),1004+ String,1005+> {1006+ match endpoint.connect(address.clone(), MEDIA_ALPN_V2).await {1007+ Ok(connection) => {1008+ let (mut send, mut recv) = connection1009+ .open_bi()1010+ .await1011+ .map_err(|error| format!("Failed to open media v2 handshake: {error}"))?;1012+ let negotiation = negotiate_v1(1013+ &mut send,1014+ &mut recv,1015+ nonce,1016+ vec![Capability::VideoH264],1017+ Some(Capability::VideoH264),1018+ "media v2",1019+ )1020+ .await?;1021+ send.finish()1022+ .map_err(|error| format!("Failed to finish media v2 handshake: {error}"))?;1023+ Ok((1024+ connection,1025+ recv,1026+ negotiation,1027+ MediaTransport::IndependentObjectsV2,1028+ ))1029+ }1030+ Err(v2_error) => {1031+ let connection = endpoint1032+ .connect(address.clone(), MEDIA_ALPN_V1)1033+ .await1034+ .map_err(|v1_error| {1035+ format!(1036+ "Failed to connect media v2 ({v2_error}); v1 compatibility connection also failed ({v1_error})"1037+ )1038+ })?;1039+ let (mut send, mut recv) = connection1040+ .open_bi()1041+ .await1042+ .map_err(|error| format!("Failed to open media v1 stream: {error}"))?;1043+ let negotiation = negotiate_v1(1044+ &mut send,1045+ &mut recv,1046+ nonce,1047+ vec![Capability::VideoH264],1048+ Some(Capability::VideoH264),1049+ "media v1",1050+ )1051+ .await?;1052+ send.finish()1053+ .map_err(|error| format!("Failed to finish media v1 handshake: {error}"))?;1054+ Ok((1055+ connection,1056+ recv,1057+ negotiation,1058+ MediaTransport::ReliableStreamV1,1059+ ))1060+ }1061+ }1062+}1063+8101064 fn input_capability_offers() -> [Vec<Capability>; 4] {8111065 [8121066 vec![@@ -1255,39 +1509,29 @@ pub async fn iroh_client_connect(
12551509 let addr = iroh::EndpointAddr::new(host_node_id);12561510 12571511 let use_v1 = development_mode;1258- let media_alpn = if use_v1 { MEDIA_ALPN_V1 } else { FRAME_ALPN };12591512 let input_alpn = if use_v1 { INPUT_ALPN_V1 } else { INPUT_ALPN };12601513 1261- let frame_conn = endpoint1262- .connect(addr.clone(), media_alpn)1263- .await1264- .map_err(|e| format!("Failed to connect frame stream: {}", e))?;1265-1266- let (mut frame_send, mut frame_recv) = frame_conn1267- .open_bi()1268- .await1269- .map_err(|e| format!("Failed to open frame stream: {}", e))?;1270- let frame_connection_for_stats = frame_conn.clone();1271-1272- let media_negotiation = if use_v1 {1273- Some(1274- negotiate_v1(1275- &mut frame_send,1276- &mut frame_recv,1277- handshake_nonce,1278- vec![Capability::VideoH264],1279- Some(Capability::VideoH264),1280- "media",1281- )1282- .await?,1283- )1514+ let (frame_conn, mut frame_recv, media_negotiation, media_transport) = if use_v1 {1515+ let (connection, recv, negotiation, transport) =1516+ open_negotiated_media_stream(&endpoint, &addr, handshake_nonce).await?;1517+ (connection, recv, Some(negotiation), transport)12841518 } else {1285- frame_send1286- .write_all(&[1u8])1519+ let connection = endpoint1520+ .connect(addr.clone(), FRAME_ALPN)1521+ .await1522+ .map_err(|e| format!("Failed to connect frame stream: {e}"))?;1523+ let (mut send, recv) = connection1524+ .open_bi()12871525 .await1288- .map_err(|e| format!("Failed to send start: {}", e))?;1289- None1526+ .map_err(|e| format!("Failed to open frame stream: {e}"))?;1527+ send.write_all(&[1u8])1528+ .await1529+ .map_err(|e| format!("Failed to send start: {e}"))?;1530+ send.finish()1531+ .map_err(|e| format!("Failed to finish frame start stream: {e}"))?;1532+ (connection, recv, None, MediaTransport::LegacyV0)12901533 };1534+ let frame_connection_for_stats = frame_conn.clone();12911535 let media_session_id = media_negotiation12921536 .as_ref()12931537 .map(|negotiation| negotiation.session_id);@@ -1573,6 +1817,9 @@ pub async fn iroh_client_connect(
15731817 let metrics = Arc::new(StdMutex::new(initial_metrics));15741818 let mut previous_sequence: Option<u64> = None;15751819 let mut frontend_waiting_for_keyframe = true;1820+ let mut media_objects = (media_transport == MediaTransport::IndependentObjectsV2)1821+ .then(|| MediaObjectReceiver::new(frame_connection_for_stats.clone()));1822+ let mut media_object_sequence = MediaObjectSequence::new();15761823 15771824 let mut decoder = if use_webcodecs {15781825 None@@ -1638,88 +1885,163 @@ pub async fn iroh_client_connect(
16381885 capture_timestamp_micros,16391886 pts_micros,16401887 discontinuity,1641- ) = if use_v1 {1642- // Gamescope's PipeWire stream is damage-driven: a static1643- // screen can legitimately produce no encoded frame for an1644- // arbitrary period. Connection closure and parser/source1645- // errors remain terminal; frame silence does not.1646- let frame = match read_media_frame(&mut frame_recv).await {1647- Ok(Some(frame)) => frame,1648- Ok(None) => {1649- emit_frame_error(&app, media_generation, "Connection closed");1650- break;1651- }1652- Err(error) => {1653- emit_frame_error(1654- &app,1655- media_generation,1656- format!("Invalid media stream: {error}"),1657- );1658- break;1888+ ) = match media_transport {1889+ MediaTransport::IndependentObjectsV2 => {1890+ let receiver = media_objects1891+ .as_mut()1892+ .expect("media v2 receiver must exist for media v2 transport");1893+ loop {1894+ let outcome = match receiver.next().await {1895+ Ok(Some(outcome)) => outcome,1896+ Ok(None) => {1897+ emit_frame_error(&app, media_generation, "Connection closed");1898+ break 'frames;1899+ }1900+ Err(error) => {1901+ emit_frame_error(&app, media_generation, error);1902+ break 'frames;1903+ }1904+ };1905+ match outcome {1906+ MediaObjectReadOutcome::Dropped { object_index } => {1907+ let begins_resync =1908+ media_object_sequence.note_dropped_object(object_index);1909+ let mut metrics = lock_client_media_metrics(&metrics);1910+ metrics.observe_transport_object_drop(!begins_resync);1911+ if begins_resync {1912+ frontend_waiting_for_keyframe = true;1913+ metrics.begin_frontend_resync(metrics_started.elapsed());1914+ }1915+ }1916+ MediaObjectReadOutcome::Malformed(error) => {1917+ emit_frame_error(&app, media_generation, error);1918+ break 'frames;1919+ }1920+ MediaObjectReadOutcome::Frame {1921+ object_index,1922+ frame,1923+ } => {1924+ let discontinuity = match media_object_sequence1925+ .classify(object_index, &frame)1926+ {1927+ MediaObjectSequenceDecision::Deliver { discontinuity } => {1928+ discontinuity1929+ }1930+ MediaObjectSequenceDecision::DropLate => {1931+ lock_client_media_metrics(&metrics)1932+ .observe_transport_object_drop(true);1933+ continue;1934+ }1935+ MediaObjectSequenceDecision::DropUntilKeyframe => {1936+ frontend_waiting_for_keyframe = true;1937+ let mut metrics = lock_client_media_metrics(&metrics);1938+ metrics.observe_transport_object_drop(false);1939+ metrics.begin_frontend_resync(metrics_started.elapsed());1940+ continue;1941+ }1942+ };1943+ let codec = match frame.header.codec {1944+ MediaCodec::H264 => "h264".to_string(),1945+ };1946+ break (1947+ u32::from(frame.header.width),1948+ u32::from(frame.header.height),1949+ frame.payload,1950+ frame.header.flags.contains(FrameFlags::KEYFRAME),1951+ codec,1952+ Some(frame.header.sequence),1953+ Some(frame.header.capture_timestamp_us),1954+ Some(frame.header.pts_us),1955+ discontinuity,1956+ );1957+ }1958+ }16591959 }1660- };1661- let codec = match frame.header.codec {1662- MediaCodec::H264 => "h264".to_string(),1663- };1664- (1665- u32::from(frame.header.width),1666- u32::from(frame.header.height),1667- frame.payload,1668- frame.header.flags.contains(FrameFlags::KEYFRAME),1669- codec,1670- Some(frame.header.sequence),1671- Some(frame.header.capture_timestamp_us),1672- Some(frame.header.pts_us),1673- frame.header.flags.contains(FrameFlags::DISCONTINUITY),1674- )1675- } else {1676- let mut header = [0u8; 14];1677- match tokio::time::timeout(1678- LEGACY_MEDIA_IDLE_TIMEOUT,1679- frame_recv.read_exact(&mut header),1680- )1681- .await1682- {1683- Err(_) => {1684- emit_frame_error(&app, media_generation, "Media stream idle timeout");1685- break;1960+ }1961+ MediaTransport::ReliableStreamV1 => {1962+ // Gamescope's PipeWire stream is damage-driven: a static1963+ // screen can legitimately produce no encoded frame for an1964+ // arbitrary period. Connection closure and parser/source1965+ // errors remain terminal; frame silence does not.1966+ let frame = match read_media_frame(&mut frame_recv).await {1967+ Ok(Some(frame)) => frame,1968+ Ok(None) => {1969+ emit_frame_error(&app, media_generation, "Connection closed");1970+ break;1971+ }1972+ Err(error) => {1973+ emit_frame_error(1974+ &app,1975+ media_generation,1976+ format!("Invalid media stream: {error}"),1977+ );1978+ break;1979+ }1980+ };1981+ let codec = match frame.header.codec {1982+ MediaCodec::H264 => "h264".to_string(),1983+ };1984+ (1985+ u32::from(frame.header.width),1986+ u32::from(frame.header.height),1987+ frame.payload,1988+ frame.header.flags.contains(FrameFlags::KEYFRAME),1989+ codec,1990+ Some(frame.header.sequence),1991+ Some(frame.header.capture_timestamp_us),1992+ Some(frame.header.pts_us),1993+ frame.header.flags.contains(FrameFlags::DISCONTINUITY),1994+ )1995+ }1996+ MediaTransport::LegacyV0 => {1997+ let mut header = [0u8; 14];1998+ match tokio::time::timeout(1999+ LEGACY_MEDIA_IDLE_TIMEOUT,2000+ frame_recv.read_exact(&mut header),2001+ )2002+ .await2003+ {2004+ Err(_) => {2005+ emit_frame_error(&app, media_generation, "Media stream idle timeout");2006+ break;2007+ }2008+ Ok(Err(_)) => {2009+ emit_frame_error(&app, media_generation, "Connection lost");2010+ break;2011+ }2012+ Ok(Ok(_)) => {}16862013 }1687- Ok(Err(_)) => {1688- emit_frame_error(&app, media_generation, "Connection lost");2014+ let w = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);2015+ let h = u32::from_be_bytes([header[4], header[5], header[6], header[7]]);2016+ let frame_len =2017+ u32::from_be_bytes([header[8], header[9], header[10], header[11]]) as usize;2018+ if let Err(error) = validate_legacy_media_header(w, h, frame_len) {2019+ emit_frame_error(&app, media_generation, error);16892020 break;16902021 }1691- Ok(Ok(_)) => {}1692- }1693- let w = u32::from_be_bytes([header[0], header[1], header[2], header[3]]);1694- let h = u32::from_be_bytes([header[4], header[5], header[6], header[7]]);1695- let frame_len =1696- u32::from_be_bytes([header[8], header[9], header[10], header[11]]) as usize;1697- if let Err(error) = validate_legacy_media_header(w, h, frame_len) {1698- emit_frame_error(&app, media_generation, error);1699- break;1700- }17012022 1702- let is_keyframe = header[12] == 1;1703- let codec = byte_to_codec(header[13]).to_string();2023+ let is_keyframe = header[12] == 1;2024+ let codec = byte_to_codec(header[13]).to_string();17042025 1705- let mut frame_buf = vec![0u8; frame_len];1706- match tokio::time::timeout(1707- LEGACY_MEDIA_IDLE_TIMEOUT,1708- frame_recv.read_exact(&mut frame_buf),1709- )1710- .await1711- {1712- Err(_) => {1713- emit_frame_error(&app, media_generation, "Media payload idle timeout");1714- break;1715- }1716- Ok(Err(_)) => {1717- emit_frame_error(&app, media_generation, "Connection lost");1718- break;2026+ let mut frame_buf = vec![0u8; frame_len];2027+ match tokio::time::timeout(2028+ LEGACY_MEDIA_IDLE_TIMEOUT,2029+ frame_recv.read_exact(&mut frame_buf),2030+ )2031+ .await2032+ {2033+ Err(_) => {2034+ emit_frame_error(&app, media_generation, "Media payload idle timeout");2035+ break;2036+ }2037+ Ok(Err(_)) => {2038+ emit_frame_error(&app, media_generation, "Connection lost");2039+ break;2040+ }2041+ Ok(Ok(_)) => {}17192042 }1720- Ok(Ok(_)) => {}2043+ (w, h, frame_buf, is_keyframe, codec, None, None, None, false)17212044 }1722- (w, h, frame_buf, is_keyframe, codec, None, None, None, false)17232045 };17242046 lock_client_media_metrics(&metrics).observe_transport_receive(17252047 metrics_started.elapsed(),@@ -1867,6 +2189,7 @@ pub async fn iroh_client_connect(
18672189 connected: true,18682190 host_node_id: Some(host_node_id.to_string()),18692191 development_mode,2192+ media_transport: media_transport.diagnostic_name(),18702193 pointer_surface_dimensions,18712194 relative_pointer_available: input_availability.relative_pointer,18722195 pointer_position_feedback_available: input_availability.pointer_position_feedback,@@ -1935,6 +2258,98 @@ fn selected_path_diagnostics(
19352258 mod tests {19362259 use super::*;19372260 2261+ fn media_object_frame(sequence: u64, flags: FrameFlags) -> MediaFrame {2262+ let payload = vec![sequence as u8];2263+ let header = sigil_protocol::MediaFrameHeader::h264(2264+ 1280,2265+ 800,2266+ payload.len(),2267+ sequence,2268+ sequence * 1_000,2269+ sequence as i64 * 1_000,2270+ flags,2271+ )2272+ .unwrap();2273+ MediaFrame::new(header, payload).unwrap()2274+ }2275+2276+ 2277+ fn media_objects_begin_on_a_configured_keyframe_then_deliver_contiguously() {2278+ let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2279+ let mut sequence = MediaObjectSequence::new();2280+2281+ assert_eq!(2282+ sequence.classify(1, &media_object_frame(1, FrameFlags::NONE)),2283+ MediaObjectSequenceDecision::DropUntilKeyframe2284+ );2285+ assert_eq!(2286+ sequence.classify(2, &media_object_frame(2, keyframe_flags)),2287+ MediaObjectSequenceDecision::Deliver {2288+ discontinuity: true2289+ }2290+ );2291+ assert_eq!(2292+ sequence.classify(3, &media_object_frame(3, FrameFlags::NONE)),2293+ MediaObjectSequenceDecision::Deliver {2294+ discontinuity: false2295+ }2296+ );2297+ }2298+2299+ 2300+ fn media_object_sequence_gaps_drop_deltas_until_a_new_keyframe() {2301+ let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2302+ let mut sequence = MediaObjectSequence::new();2303+2304+ assert!(matches!(2305+ sequence.classify(1, &media_object_frame(10, keyframe_flags)),2306+ MediaObjectSequenceDecision::Deliver { .. }2307+ ));2308+ assert_eq!(2309+ sequence.classify(3, &media_object_frame(12, FrameFlags::NONE)),2310+ MediaObjectSequenceDecision::DropUntilKeyframe2311+ );2312+ assert_eq!(2313+ sequence.classify(4, &media_object_frame(13, FrameFlags::NONE)),2314+ MediaObjectSequenceDecision::DropUntilKeyframe2315+ );2316+ assert_eq!(2317+ sequence.classify(5, &media_object_frame(14, keyframe_flags)),2318+ MediaObjectSequenceDecision::Deliver {2319+ discontinuity: true2320+ }2321+ );2322+ }2323+2324+ 2325+ fn late_media_object_completion_cannot_rewind_a_recovered_stream() {2326+ let keyframe_flags = FrameFlags::KEYFRAME.union(FrameFlags::CODEC_CONFIG);2327+ let mut sequence = MediaObjectSequence::new();2328+2329+ assert!(matches!(2330+ sequence.classify(10, &media_object_frame(10, keyframe_flags)),2331+ MediaObjectSequenceDecision::Deliver { .. }2332+ ));2333+ assert!(sequence.note_dropped_object(12));2334+ assert_eq!(2335+ sequence.classify(13, &media_object_frame(13, keyframe_flags)),2336+ MediaObjectSequenceDecision::Deliver {2337+ discontinuity: true2338+ }2339+ );2340+ assert!(!sequence.note_dropped_object(11));2341+ assert_eq!(2342+ sequence.classify(12, &media_object_frame(12, FrameFlags::NONE)),2343+ MediaObjectSequenceDecision::DropLate2344+ );2345+ assert_eq!(2346+ sequence.classify(14, &media_object_frame(14, FrameFlags::NONE)),2347+ MediaObjectSequenceDecision::Deliver {2348+ discontinuity: false2349+ }2350+ );2351+ }2352+19382353 19392354 fn binary_frame_envelope_has_exact_stable_layout() {19402355 let payload = [0, 0, 0, 1, 0x65];@@ -2350,6 +2765,8 @@ mod tests {
23502765 metrics.observe_transport_receive(Duration::from_millis(100), Some(40), false);23512766 metrics.observe_transport_receive(Duration::from_millis(200), Some(43), true);23522767 metrics.observe_sequence_drop(2);2768+ metrics.observe_transport_object_drop(false);2769+ metrics.observe_transport_object_drop(true);23532770 metrics.observe_frontend_queue_drop();23542771 metrics.begin_frontend_resync(Duration::from_millis(200));23552772 metrics.begin_frontend_resync(Duration::from_millis(210));@@ -2367,10 +2784,12 @@ mod tests {
23672784 let json = serde_json::to_value(payload).unwrap();23682785 23692786 assert_eq!(json["generation"], 9);2370- assert_eq!(json["stats_version"], 2);2787+ assert_eq!(json["stats_version"], 3);23712788 assert_eq!(json["transport_received_total"], 2);23722789 assert_eq!(json["frontend_sent_total"], 1);23732790 assert_eq!(json["sequence_dropped_total"], 2);2791+ assert_eq!(json["transport_object_dropped_total"], 2);2792+ assert_eq!(json["transport_late_object_dropped_total"], 1);23742793 assert_eq!(json["frontend_queue_dropped_total"], 1);23752794 assert_eq!(json["frontend_resync_dropped_total"], 2);23762795 assert_eq!(json["frontend_dropped_total"], 3);@@ -81,12 +81,20 @@ export function normalizeFrameStatsPayload(payload) {
8181 8282 const statsVersion = exactUnsigned(payload.stats_version) ?? 1;8383 const v2 = statsVersion >= 2;84+ const v3 = statsVersion >= 3;8485 const queueDroppedFrames = v2 ? exactUnsigned(payload.frontend_queue_dropped_total) : null;8586 const resyncDroppedFrames = v2 ? exactUnsigned(payload.frontend_resync_dropped_total) : null;8687 const splitFrontendTotal = queueDroppedFrames !== null && resyncDroppedFrames !== null8788 && Number.isSafeInteger(queueDroppedFrames + resyncDroppedFrames)8889 ? queueDroppedFrames + resyncDroppedFrames8990 : null;91+ const rawObjectDroppedFrames = v392+ ? exactUnsigned(payload.transport_object_dropped_total) : null;93+ const rawLateObjectDroppedFrames = v394+ ? exactUnsigned(payload.transport_late_object_dropped_total) : null;95+ const objectCountersValid = rawObjectDroppedFrames !== null96+ && rawLateObjectDroppedFrames !== null97+ && rawLateObjectDroppedFrames <= rawObjectDroppedFrames;9098 9199 let queue = null;92100 let resync = null;@@ -128,6 +136,8 @@ export function normalizeFrameStatsPayload(payload) {
128136 payload.sequence_dropped_total,129137 payload.host_dropped_frames,130138 ),139+ objectDroppedFrames: objectCountersValid ? rawObjectDroppedFrames : null,140+ lateObjectDroppedFrames: objectCountersValid ? rawLateObjectDroppedFrames : null,131141 frontendDroppedFrames: firstExactUnsigned(132142 payload.frontend_dropped_total,133143 payload.frontend_dropped_frames,@@ -80,6 +80,8 @@ test('normalizes exact v2 drop, queue, timing, and resync units', () => {
8080 }), {8181 statsVersion: 2,8282 transportDroppedFrames: 3,83+ objectDroppedFrames: null,84+ lateObjectDroppedFrames: null,8385 frontendDroppedFrames: 9,8486 queueDroppedFrames: 4,8587 resyncDroppedFrames: 5,@@ -98,6 +100,8 @@ test('uses legacy aggregate drop aliases without inventing v2 splits', () => {
98100 });99101 assert.equal(stats.statsVersion, 1);100102 assert.equal(stats.transportDroppedFrames, 7);103+ assert.equal(stats.objectDroppedFrames, null);104+ assert.equal(stats.lateObjectDroppedFrames, null);101105 assert.equal(stats.frontendDroppedFrames, 11);102106 assert.equal(stats.queueDroppedFrames, null);103107 assert.equal(stats.resyncDroppedFrames, null);@@ -108,6 +112,24 @@ test('uses legacy aggregate drop aliases without inventing v2 splits', () => {
108112 assert.equal(stats.timingWindow, null);109113 });110114 115+test('normalizes bounded v3 independent-object discard counters', () => {116+ const stats = normalizeFrameStatsPayload({117+ stats_version: 3,118+ transport_object_dropped_total: 8,119+ transport_late_object_dropped_total: 3,120+ });121+ assert.equal(stats.objectDroppedFrames, 8);122+ assert.equal(stats.lateObjectDroppedFrames, 3);123+124+ const malformed = normalizeFrameStatsPayload({125+ stats_version: 3,126+ transport_object_dropped_total: 2,127+ transport_late_object_dropped_total: 3,128+ });129+ assert.equal(malformed.objectDroppedFrames, null);130+ assert.equal(malformed.lateObjectDroppedFrames, null);131+});132+111133 test('does not display malformed or unit-ambiguous v2 diagnostics', () => {112134 const stats = normalizeFrameStatsPayload({113135 stats_version: 2,@@ -121,6 +121,7 @@
121121 <div class="panel-section" id="stream-section">122122 <h3># stream</h3>123123 <div class="info-row"><span class="info-label">codec</span><span class="info-value" id="stream-codec">—</span></div>124+ <div class="info-row"><span class="info-label">transport</span><span class="info-value" id="stream-transport">—</span></div>124125 <div class="info-row"><span class="info-label">path</span><span class="info-value" id="stream-path">—</span></div>125126 <div class="info-row"><span class="info-label">RTT</span><span class="info-value" id="stream-rtt">— ms</span></div>126127 <div class="info-row"><span class="info-label">control</span><span class="info-value" id="stream-control">—</span></div>@@ -158,6 +159,8 @@
158159 <div class="info-row metric-stack"><span class="info-label">present interval p50/p95/p99/max</span><span class="info-value" id="stream-present-cadence">—</span></div>159160 <div class="info-row"><span class="info-label">total discarded</span><span class="info-value" id="stream-dropped">0 frames</span></div>160161 <div class="info-row"><span class="info-label">transport sequence gap</span><span class="info-value" id="stream-transport-dropped">0 frames</span></div>162+ <div class="info-row"><span class="info-label">transport object discarded</span><span class="info-value" id="stream-transport-object-dropped">—</span></div>163+ <div class="info-row"><span class="info-label">late object discarded</span><span class="info-value" id="stream-transport-late-object-dropped">—</span></div>161164 <div class="info-row"><span class="info-label">IPC discarded</span><span class="info-value" id="stream-frontend-dropped">0 frames</span></div>162165 <div class="info-row"><span class="info-label">IPC queue discarded</span><span class="info-value" id="stream-frontend-queue-dropped">—</span></div>163166 <div class="info-row"><span class="info-label">IPC resync discarded</span><span class="info-value" id="stream-frontend-resync-dropped">—</span></div>@@ -666,6 +666,13 @@ async function connectHost() {
666666 gamepad: result.gamepad_available === true,667667 control: result.control_available === true,668668 };669+ streamTransportMode = [670+ 'independent-v2',671+ 'reliable-v1',672+ 'reliable-v0',673+ ].includes(result.media_transport)674+ ? result.media_transport675+ : 'unknown';669676 pointerSurfaceDimensions = connectedPointerSurfaceDimensions;670677 if (!inputCapabilities.pointerPositionFeedback) {671678 remotePointerPosition = null;@@ -878,6 +885,8 @@ let decoderConfigured = false;
878885 let activeCodec = 'h264';879886 let droppedFrames = 0;880887 let transportDroppedFrames = 0;888+let transportObjectDroppedFrames = null;889+let transportLateObjectDroppedFrames = null;881890 let frontendDroppedFrames = 0;882891 let frontendQueueDroppedFrames = null;883892 let frontendResyncDroppedFrames = null;@@ -908,6 +917,7 @@ const avSkew = new BoundedValueWindow();
908917 const decodeTimings = new Map();909918 const MAX_DECODE_TIMINGS = 8;910919 let streamPathMode = 'unknown';920+let streamTransportMode = 'unknown';911921 let streamRttMs = null;912922 const MAX_DECODE_QUEUE_SIZE = 2;913923 let waitingForDecoderKeyframe = true;@@ -999,6 +1009,8 @@ function resetStreamTelemetry() {
9991009 teardownDecoderPipeline();10001010 droppedFrames = 0;10011011 transportDroppedFrames = 0;1012+ transportObjectDroppedFrames = null;1013+ transportLateObjectDroppedFrames = null;10021014 frontendDroppedFrames = 0;10031015 frontendQueueDroppedFrames = null;10041016 frontendResyncDroppedFrames = null;@@ -1026,6 +1038,7 @@ function resetStreamTelemetry() {
10261038 clientPresentationLatency.reset();10271039 drawLatency.reset();10281040 streamPathMode = 'unknown';1041+ streamTransportMode = 'unknown';10291042 streamRttMs = null;10301043 waitingForDecoderKeyframe = true;10311044 }@@ -1120,6 +1133,11 @@ function updateStreamStats() {
11201133 });11211134 document.getElementById('stream-dropped').textContent = discardTelemetry.total;11221135 document.getElementById('stream-transport-dropped').textContent = discardTelemetry.transport;1136+ document.getElementById('stream-transport-object-dropped').textContent =1137+ transportObjectDroppedFrames === null ? '—' : `${transportObjectDroppedFrames} frames`;1138+ document.getElementById('stream-transport-late-object-dropped').textContent =1139+ transportLateObjectDroppedFrames === null1140+ ? '—' : `${transportLateObjectDroppedFrames} frames`;11231141 document.getElementById('stream-frontend-dropped').textContent = discardTelemetry.frontend;11241142 document.getElementById('stream-frontend-queue-dropped').textContent = frontendQueueDroppedFrames === null11251143 ? '—' : `${frontendQueueDroppedFrames} frames`;@@ -1161,6 +1179,7 @@ function updateStreamStats() {
11611179 document.getElementById('stream-decoder-output-cadence').textContent = formatCadence(decoderOutputIntervals);11621180 document.getElementById('stream-present-cadence').textContent = formatCadence(presentationIntervals);11631181 document.getElementById('stream-codec').textContent = activeCodec;1182+ document.getElementById('stream-transport').textContent = streamTransportMode;11641183 document.getElementById('stream-path').textContent = streamPathMode;11651184 document.getElementById('stream-rtt').textContent = Number.isFinite(streamRttMs)11661185 ? `${streamRttMs.toFixed(1)} ms`@@ -1569,6 +1588,8 @@ listen('frame-stats', (event) => {
15691588 if (!isCurrentFrameGeneration(event.payload?.generation, activeFrameGeneration)) return;15701589 const diagnostics = normalizeFrameStatsPayload(event.payload);15711590 transportDroppedFrames = diagnostics.transportDroppedFrames ?? transportDroppedFrames;1591+ transportObjectDroppedFrames = diagnostics.objectDroppedFrames;1592+ transportLateObjectDroppedFrames = diagnostics.lateObjectDroppedFrames;15721593 frontendDroppedFrames = diagnostics.frontendDroppedFrames ?? frontendDroppedFrames;15731594 frontendQueueDroppedFrames = diagnostics.queueDroppedFrames;15741595 frontendResyncDroppedFrames = diagnostics.resyncDroppedFrames;