1use std::sync::atomic::{AtomicU64, Ordering};2use std::sync::{Arc, Mutex as StdMutex};3use std::time::Duration;45use iroh::Endpoint;6use serde::{Deserialize, Serialize};7use sigil_protocol::{8 AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1, Capability,9 MEDIA_FEEDBACK_ALPN_V1, MediaFeedbackFlags, MediaFeedbackReportV1,10 read_adaptive_bitrate_decision_v1, write_media_feedback_report_v1,11};12use tauri::{AppHandle, Emitter, Manager};1314use crate::commands::state::{AccumulatedMediaFeedback, AppState};15use crate::media::frame_channel::take_generation_owned_triple;16use crate::media::network_diagnostics::{NetworkSessionDiagnostics, lock_network_diagnostics};17use crate::media::transport::{connect_error_is_unsupported_alpn, negotiate_v1};1819pub(crate) const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);20const CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT: Duration = Duration::from_secs(5);21const CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL: Duration = Duration::from_secs(1);2223async fn retire_media_feedback_generation(app: &AppHandle, generation: u64) -> bool {24 let state = app.state::<AppState>();25 let connection = {26 let _connection_serial = state.client_connection_serial.lock().await;27 let mut slot = state.media_feedback.lock().await;28 take_generation_owned_triple(&mut slot, generation).map(|(connection, _)| connection)29 };30 let Some(connection) = connection else {31 return false;32 };33 connection.close(0_u32.into(), b"adaptive feedback ended");34 true35}363738pub struct ClientMediaFeedbackReport {39 pub interval_ms: u16,40 pub last_sequence: Option<u64>,41 pub transport_dropped_delta: u32,42 pub frontend_dropped_delta: u32,43 pub decoder_dropped_delta: u32,44 pub presenter_dropped_delta: u32,45 pub frontend_queue_depth: u8,46 pub frontend_queue_capacity: u8,47 pub decode_queue_depth: u8,48 pub decode_queue_capacity: u8,49 pub presenter_queue_depth: u8,50 pub presenter_queue_capacity: u8,51 pub transport_delivery_p95_ms: Option<f64>,52 pub decode_p95_ms: Option<f64>,53 pub presentation_p95_ms: Option<f64>,54 pub resync_active: bool,55}565758struct AdaptiveBitrateDecisionPayload {59 generation: u64,60 decision: AdaptiveBitrateDecisionDiagnostic,61}626364struct AdaptiveBitrateDecisionDiagnostic {65 decision_id: u64,66 report_id: u64,67 target_kbps: u32,68 floor_kbps: u32,69 ceiling_kbps: u32,70 state: &'static str,71 reasons: Vec<&'static str>,72 applied: bool,73 contribution: &'static str,74 recovery_state: &'static str,75 scope: &'static str,76}7778pub(crate) async fn open_negotiated_feedback_stream(79 endpoint: &Endpoint,80 address: &iroh::EndpointAddr,81 nonce: [u8; 16],82 media_session_id: u64,83) -> Result<84 Option<(85 iroh::endpoint::Connection,86 iroh::endpoint::SendStream,87 iroh::endpoint::RecvStream,88 )>,89 String,90> {91 let connection = match endpoint92 .connect(address.clone(), MEDIA_FEEDBACK_ALPN_V1)93 .await94 {95 Ok(connection) => connection,96 Err(error) if connect_error_is_unsupported_alpn(&error) => return Ok(None),97 Err(error) => {98 return Err(format!(99 "Failed to connect adaptive feedback stream: {error}"100 ));101 }102 };103 let (mut send, mut recv) = connection104 .open_bi()105 .await106 .map_err(|error| format!("Failed to open adaptive feedback handshake: {error}"))?;107 let negotiation = negotiate_v1(108 &mut send,109 &mut recv,110 nonce,111 vec![Capability::VideoH264],112 Some(Capability::VideoH264),113 "adaptive feedback",114 None,115 )116 .await?;117 if negotiation.session_id != media_session_id {118 return Err("Host returned mismatched media and adaptive feedback sessions".to_string());119 }120 Ok(Some((connection, send, recv)))121}122123fn adaptive_bitrate_state_name(state: AdaptiveBitrateStateV1) -> &'static str {124 match state {125 AdaptiveBitrateStateV1::Hold => "hold",126 AdaptiveBitrateStateV1::Decrease => "decrease",127 AdaptiveBitrateStateV1::Increase => "increase",128 }129}130131fn adaptive_bitrate_reason_names(reasons: AdaptiveBitrateReasonFlagsV1) -> Vec<&'static str> {132 [133 (AdaptiveBitrateReasonFlagsV1::RTT_INFLATION, "rtt-inflation"),134 (135 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,136 "loss-or-cancellation",137 ),138 (139 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,140 "sender-backpressure",141 ),142 (143 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,144 "receiver-queue",145 ),146 (147 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,148 "decode-backlog",149 ),150 (151 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,152 "delivery-latency",153 ),154 (155 AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,156 "clean-recovery",157 ),158 (159 AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE,160 "feedback-stale",161 ),162 ]163 .into_iter()164 .filter_map(|(flag, name)| reasons.contains(flag).then_some(name))165 .collect()166}167168fn adaptive_bitrate_decision_diagnostic(169 decision: AdaptiveBitrateDecisionV1,170) -> AdaptiveBitrateDecisionDiagnostic {171 let pressured = decision172 .reasons173 .contains(AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE)174 || decision175 .reasons176 .contains(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG)177 || decision178 .reasons179 .contains(AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY)180 || decision181 .reasons182 .contains(AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION)183 || decision184 .reasons185 .contains(AdaptiveBitrateReasonFlagsV1::RTT_INFLATION);186 let contribution = if decision187 .reasons188 .contains(AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE)189 {190 "stale"191 } else if pressured {192 "limiting"193 } else if decision194 .reasons195 .contains(AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY)196 {197 "clean"198 } else {199 "steady"200 };201 let recovery_state = if pressured && decision.target_kbps == decision.floor_kbps {202 "floor-limited"203 } else {204 "active"205 };206 AdaptiveBitrateDecisionDiagnostic {207 decision_id: decision.decision_id,208 report_id: decision.report_id,209 target_kbps: decision.target_kbps,210 floor_kbps: decision.floor_kbps,211 ceiling_kbps: decision.ceiling_kbps,212 state: adaptive_bitrate_state_name(decision.state),213 reasons: adaptive_bitrate_reason_names(decision.reasons),214 applied: decision.applied,215 contribution,216 recovery_state,217 scope: "local-viewer",218 }219}220221222struct AdaptiveDecisionSequence {223 last_decision_id: Option<u64>,224 last_report_id: Option<u64>,225}226227impl AdaptiveDecisionSequence {228 fn accept(&mut self, decision: &AdaptiveBitrateDecisionV1) -> Result<(), String> {229 if self230 .last_decision_id231 .is_some_and(|previous| decision.decision_id <= previous)232 {233 return Err(format!(234 "adaptive decision ID did not increase: previous={:?}, current={}",235 self.last_decision_id, decision.decision_id236 ));237 }238 if self239 .last_report_id240 .is_some_and(|previous| decision.report_id <= previous)241 {242 return Err(format!(243 "adaptive decision report ID did not increase: previous={:?}, current={}",244 self.last_report_id, decision.report_id245 ));246 }247 self.last_decision_id = Some(decision.decision_id);248 self.last_report_id = Some(decision.report_id);249 Ok(())250 }251}252253async fn emit_paced_adaptive_decisions<F>(254 generation: u64,255 mut decisions: tokio::sync::watch::Receiver<Option<AdaptiveBitrateDecisionV1>>,256 delivery_interval: Duration,257 mut emit: F,258) -> Result<(), String>259where260 F: FnMut(AdaptiveBitrateDecisionPayload) -> Result<(), String>,261{262 let mut delivery = tokio::time::interval(delivery_interval);263 delivery.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);264 loop {265 decisions266 .changed()267 .await268 .map_err(|_| "adaptive decision receiver closed".to_string())?;269 // The first tick is immediate. Later ticks pace the webview boundary;270 // any host burst replaces the watch value while this task waits.271 delivery.tick().await;272 let Some(decision) = *decisions.borrow_and_update() else {273 continue;274 };275 emit(AdaptiveBitrateDecisionPayload {276 generation,277 decision: adaptive_bitrate_decision_diagnostic(decision),278 })?;279 }280}281282pub(crate) async fn run_media_feedback_session(283 app: AppHandle,284 generation: u64,285 mut send: iroh::endpoint::SendStream,286 mut recv: iroh::endpoint::RecvStream,287 mut reports: tokio::sync::watch::Receiver<Option<AccumulatedMediaFeedback>>,288 network_diagnostics: Arc<StdMutex<NetworkSessionDiagnostics>>,289) {290 let writer_diagnostics = Arc::clone(&network_diagnostics);291 let (decision_tx, decision_rx) = tokio::sync::watch::channel(None);292 let writer = async {293 let mut last_written = None;294 loop {295 reports296 .changed()297 .await298 .map_err(|_| "adaptive feedback sender closed".to_string())?;299 let accumulated = *reports.borrow_and_update();300 let Some(accumulated) = accumulated else {301 continue;302 };303 let report = accumulated.report_since(last_written);304 lock_network_diagnostics(&writer_diagnostics).observe_adaptive_report(&report);305 tokio::time::timeout(306 CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT,307 write_media_feedback_report_v1(&mut send, &report),308 )309 .await310 .map_err(|_| "adaptive feedback write timed out".to_string())?311 .map_err(|error| format!("adaptive feedback write failed: {error}"))?;312 last_written = Some(accumulated);313 }314 315 Ok::<(), String>(())316 };317 let reader_diagnostics = Arc::clone(&network_diagnostics);318 let reader = async {319 let mut sequence = AdaptiveDecisionSequence::default();320 loop {321 let decision = read_adaptive_bitrate_decision_v1(&mut recv)322 .await323 .map_err(|error| format!("adaptive decision read failed: {error}"))?324 .ok_or_else(|| "adaptive decision stream closed".to_string())?;325 sequence.accept(&decision)?;326 lock_network_diagnostics(&reader_diagnostics).observe_adaptive_decision(&decision);327 decision_tx.send_replace(Some(decision));328 }329 330 Ok::<(), String>(())331 };332 let decision_app = app.clone();333 let decision_emitter = emit_paced_adaptive_decisions(334 generation,335 decision_rx,336 CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL,337 move |payload| {338 decision_app339 .emit("adaptive-bitrate-decision", payload)340 .map_err(|error| format!("adaptive decision event delivery failed: {error}"))341 },342 );343 let terminal = tokio::select! {344 terminal = writer => terminal,345 terminal = reader => terminal,346 terminal = decision_emitter => terminal,347 };348 if let Err(error) = terminal {349 lock_network_diagnostics(&network_diagnostics).mark_adaptive_detached();350 eprintln!("[client] {error}");351 if let Err(emit_error) = app.emit(352 "adaptive-feedback-state",353 serde_json::json!({354 "generation": generation,355 "available": false,356 "recovery_state": "detached",357 "error": error,358 }),359 ) {360 eprintln!("[client] adaptive feedback terminal event delivery failed: {emit_error}");361 }362 }363 retire_media_feedback_generation(&app, generation).await;364}365366fn feedback_latency_ms(value: Option<f64>, name: &str) -> Result<Option<u16>, String> {367 value368 .map(|value| {369 if !value.is_finite() || !(0.0..=60_000.0).contains(&value) {370 return Err(format!("{name} must be finite and between 0 and 60000 ms"));371 }372 Ok(value.round() as u16)373 })374 .transpose()375}376377fn next_feedback_report_id(counter: &AtomicU64) -> Result<u64, String> {378 counter379 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {380 current.checked_add(1)381 })382 .map(|previous| previous + 1)383 .map_err(|_| "Adaptive feedback report ID overflowed".to_string())384}385386fn media_feedback_report(387 report_id: u64,388 input: ClientMediaFeedbackReport,389) -> Result<MediaFeedbackReportV1, String> {390 let flags = if input.resync_active {391 MediaFeedbackFlags::RESYNC_ACTIVE392 } else {393 MediaFeedbackFlags::NONE394 };395 let report = MediaFeedbackReportV1 {396 report_id,397 interval_ms: input.interval_ms,398 flags,399 last_sequence: input.last_sequence,400 transport_dropped_delta: input.transport_dropped_delta,401 frontend_dropped_delta: input.frontend_dropped_delta,402 decoder_dropped_delta: input.decoder_dropped_delta,403 presenter_dropped_delta: input.presenter_dropped_delta,404 frontend_queue_depth: input.frontend_queue_depth,405 frontend_queue_capacity: input.frontend_queue_capacity,406 decode_queue_depth: input.decode_queue_depth,407 decode_queue_capacity: input.decode_queue_capacity,408 presenter_queue_depth: input.presenter_queue_depth,409 presenter_queue_capacity: input.presenter_queue_capacity,410 transport_delivery_p95_ms: feedback_latency_ms(411 input.transport_delivery_p95_ms,412 "delivery p95",413 )?,414 decode_p95_ms: feedback_latency_ms(input.decode_p95_ms, "decode p95")?,415 presentation_p95_ms: feedback_latency_ms(input.presentation_p95_ms, "presentation p95")?,416 };417 report418 .validate()419 .map_err(|error| format!("Invalid adaptive feedback report: {error}"))?;420 Ok(report)421}422423pub(crate) async fn send_media_feedback(424 state: &AppState,425 generation: u64,426 report: ClientMediaFeedbackReport,427) -> Result<bool, String> {428 let feedback = state.media_feedback.lock().await;429 let Some((current_generation, _connection, sender)) = feedback.as_ref() else {430 return Ok(false);431 };432 if *current_generation != generation {433 return Ok(false);434 }435 if sender.receiver_count() == 0 {436 return Err("Adaptive feedback channel closed".to_string());437 }438 let report_id = next_feedback_report_id(&state.media_feedback_report_id)?;439 let report = media_feedback_report(report_id, report)?;440 sender.send_modify(|pending| match pending {441 Some(pending) => pending.merge(report),442 None => *pending = Some(AccumulatedMediaFeedback::new(report)),443 });444 Ok(true)445}446447448mod tests {449 use super::*;450451 fn client_feedback() -> ClientMediaFeedbackReport {452 ClientMediaFeedbackReport {453 interval_ms: 1_000,454 last_sequence: Some(99),455 transport_dropped_delta: 1,456 frontend_dropped_delta: 2,457 decoder_dropped_delta: 3,458 presenter_dropped_delta: 4,459 frontend_queue_depth: 1,460 frontend_queue_capacity: 4,461 decode_queue_depth: 1,462 decode_queue_capacity: 2,463 presenter_queue_depth: 0,464 presenter_queue_capacity: 2,465 transport_delivery_p95_ms: Some(17.4),466 decode_p95_ms: Some(4.6),467 presentation_p95_ms: None,468 resync_active: true,469 }470 }471472 473 fn adaptive_feedback_conversion_is_bounded_and_protocol_valid() {474 let report = media_feedback_report(7, client_feedback()).unwrap();475 assert_eq!(report.report_id, 7);476 assert_eq!(report.interval_ms, 1_000);477 assert_eq!(report.last_sequence, Some(99));478 assert_eq!(report.transport_delivery_p95_ms, Some(17));479 assert_eq!(report.decode_p95_ms, Some(5));480 assert!(report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE));481 report.validate().unwrap();482483 let mut invalid = client_feedback();484 invalid.decode_queue_depth = 3;485 assert!(media_feedback_report(8, invalid).is_err());486 let mut invalid_interval = client_feedback();487 invalid_interval.interval_ms = 249;488 assert!(media_feedback_report(9, invalid_interval).is_err());489 assert!(feedback_latency_ms(Some(f64::INFINITY), "decode").is_err());490 assert!(feedback_latency_ms(Some(60_001.0), "decode").is_err());491 }492493 494 fn feedback_report_ids_are_nonzero_monotonic_and_checked() {495 let counter = AtomicU64::new(0);496 assert_eq!(next_feedback_report_id(&counter).unwrap(), 1);497 assert_eq!(next_feedback_report_id(&counter).unwrap(), 2);498 assert!(next_feedback_report_id(&AtomicU64::new(u64::MAX)).is_err());499 }500501 502 async fn adaptive_feedback_watch_coalesces_latest_state_without_losing_pressure() {503 let (sender, mut receiver) = tokio::sync::watch::channel(None);504 let first = media_feedback_report(1, client_feedback()).unwrap();505 sender.send_replace(Some(AccumulatedMediaFeedback::new(first)));506 receiver.changed().await.unwrap();507 let stalled_write = receiver.borrow_and_update().unwrap();508509 let mut latest = client_feedback();510 latest.interval_ms = 1_250;511 latest.transport_dropped_delta = 9;512 let second = media_feedback_report(2, latest).unwrap();513 sender.send_modify(|pending| pending.as_mut().unwrap().merge(second));514 let mut newest = client_feedback();515 newest.interval_ms = 1_500;516 newest.transport_dropped_delta = 11;517 newest.frontend_queue_depth = 3;518 let third = media_feedback_report(3, newest).unwrap();519 sender.send_modify(|pending| pending.as_mut().unwrap().merge(third));520521 assert_eq!(stalled_write.report_since(None), first);522 receiver.changed().await.unwrap();523 let coalesced = receiver524 .borrow_and_update()525 .unwrap()526 .report_since(Some(stalled_write));527 assert_eq!(coalesced.report_id, 3);528 assert_eq!(coalesced.interval_ms, 2_750);529 assert_eq!(coalesced.transport_dropped_delta, 20);530 assert_eq!(coalesced.frontend_queue_depth, 3);531 }532533 534 fn adaptive_decision_diagnostics_are_explicitly_shadow_state() {535 let diagnostic = adaptive_bitrate_decision_diagnostic(adaptive_decision(2, 1));536 assert_eq!(diagnostic.state, "decrease");537 assert_eq!(diagnostic.reasons, vec!["receiver-queue", "decode-backlog"]);538 assert!(!diagnostic.applied);539 assert_eq!(diagnostic.contribution, "limiting");540 assert_eq!(diagnostic.recovery_state, "active");541 assert_eq!(diagnostic.scope, "local-viewer");542 }543544 fn adaptive_decision(decision_id: u64, report_id: u64) -> AdaptiveBitrateDecisionV1 {545 AdaptiveBitrateDecisionV1 {546 decision_id,547 report_id,548 target_kbps: 8_000,549 floor_kbps: 4_000,550 ceiling_kbps: 20_000,551 state: AdaptiveBitrateStateV1::Decrease,552 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE553 .union(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG),554 applied: false,555 }556 }557558 559 fn adaptive_decision_sequence_rejects_duplicate_or_regressing_ids() {560 let mut sequence = AdaptiveDecisionSequence::default();561 sequence.accept(&adaptive_decision(1, 10)).unwrap();562 sequence.accept(&adaptive_decision(3, 12)).unwrap();563 assert!(sequence.accept(&adaptive_decision(3, 13)).is_err());564565 let mut report_regression = AdaptiveDecisionSequence::default();566 report_regression.accept(&adaptive_decision(1, 10)).unwrap();567 assert!(report_regression.accept(&adaptive_decision(2, 10)).is_err());568 assert!(report_regression.accept(&adaptive_decision(3, 9)).is_err());569 }570571 572 async fn adaptive_decision_delivery_is_paced_latest_value_and_emit_failure_is_terminal() {573 let (sender, receiver) = tokio::sync::watch::channel(None);574 let (delivered_tx, mut delivered_rx) = tokio::sync::mpsc::channel(2);575 let delivery = tokio::spawn(emit_paced_adaptive_decisions(576 7,577 receiver,578 Duration::from_millis(20),579 move |payload| {580 delivered_tx581 .try_send(payload)582 .map_err(|error| format!("test delivery failed: {error}"))583 },584 ));585 sender.send_replace(Some(adaptive_decision(1, 1)));586 sender.send_replace(Some(adaptive_decision(2, 2)));587 let first = tokio::time::timeout(Duration::from_millis(100), delivered_rx.recv())588 .await589 .unwrap()590 .unwrap();591 assert_eq!(first.generation, 7);592 assert_eq!(first.decision.decision_id, 2);593594 sender.send_replace(Some(adaptive_decision(3, 3)));595 assert!(596 tokio::time::timeout(Duration::from_millis(5), delivered_rx.recv())597 .await598 .is_err()599 );600 let second = tokio::time::timeout(Duration::from_millis(100), delivered_rx.recv())601 .await602 .unwrap()603 .unwrap();604 assert_eq!(second.decision.decision_id, 3);605 delivery.abort();606607 let (sender, receiver) = tokio::sync::watch::channel(None);608 let terminal = tokio::spawn(emit_paced_adaptive_decisions(609 7,610 receiver,611 Duration::from_millis(1),612 |_payload| Err("webview channel closed".to_string()),613 ));614 sender.send_replace(Some(adaptive_decision(4, 4)));615 let error = tokio::time::timeout(Duration::from_millis(100), terminal)616 .await617 .unwrap()618 .unwrap()619 .unwrap_err();620 assert_eq!(error, "webview channel closed");621 }622}