1use std::sync::atomic::{AtomicU64, Ordering};2use std::time::Duration;34use iroh::Endpoint;5use serde::{Deserialize, Serialize};6use sigil_protocol::{7 AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1, Capability,8 MEDIA_FEEDBACK_ALPN_V1, MediaFeedbackFlags, MediaFeedbackReportV1,9 read_adaptive_bitrate_decision_v1, write_media_feedback_report_v1,10};11use tauri::{AppHandle, Emitter, Manager};1213use crate::commands::state::{AccumulatedMediaFeedback, AppState};14use crate::media::frame_channel::take_generation_owned_triple;15use crate::media::transport::{connect_error_is_unsupported_alpn, negotiate_v1};1617pub(crate) const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);18const CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT: Duration = Duration::from_secs(5);19const CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL: Duration = Duration::from_secs(1);2021async fn retire_media_feedback_generation(app: &AppHandle, generation: u64) -> bool {22 let state = app.state::<AppState>();23 let connection = {24 let _connection_serial = state.client_connection_serial.lock().await;25 let mut slot = state.media_feedback.lock().await;26 take_generation_owned_triple(&mut slot, generation).map(|(connection, _)| connection)27 };28 let Some(connection) = connection else {29 return false;30 };31 connection.close(0_u32.into(), b"adaptive feedback ended");32 true33}343536pub struct ClientMediaFeedbackReport {37 pub interval_ms: u16,38 pub last_sequence: Option<u64>,39 pub transport_dropped_delta: u32,40 pub frontend_dropped_delta: u32,41 pub decoder_dropped_delta: u32,42 pub presenter_dropped_delta: u32,43 pub frontend_queue_depth: u8,44 pub frontend_queue_capacity: u8,45 pub decode_queue_depth: u8,46 pub decode_queue_capacity: u8,47 pub presenter_queue_depth: u8,48 pub presenter_queue_capacity: u8,49 pub transport_delivery_p95_ms: Option<f64>,50 pub decode_p95_ms: Option<f64>,51 pub presentation_p95_ms: Option<f64>,52 pub resync_active: bool,53}545556struct AdaptiveBitrateDecisionPayload {57 generation: u64,58 decision: AdaptiveBitrateDecisionDiagnostic,59}606162struct AdaptiveBitrateDecisionDiagnostic {63 decision_id: u64,64 report_id: u64,65 target_kbps: u32,66 floor_kbps: u32,67 ceiling_kbps: u32,68 state: &'static str,69 reasons: Vec<&'static str>,70 applied: bool,71}7273pub(crate) async fn open_negotiated_feedback_stream(74 endpoint: &Endpoint,75 address: &iroh::EndpointAddr,76 nonce: [u8; 16],77 media_session_id: u64,78) -> Result<79 Option<(80 iroh::endpoint::Connection,81 iroh::endpoint::SendStream,82 iroh::endpoint::RecvStream,83 )>,84 String,85> {86 let connection = match endpoint87 .connect(address.clone(), MEDIA_FEEDBACK_ALPN_V1)88 .await89 {90 Ok(connection) => connection,91 Err(error) if connect_error_is_unsupported_alpn(&error) => return Ok(None),92 Err(error) => {93 return Err(format!(94 "Failed to connect adaptive feedback stream: {error}"95 ));96 }97 };98 let (mut send, mut recv) = connection99 .open_bi()100 .await101 .map_err(|error| format!("Failed to open adaptive feedback handshake: {error}"))?;102 let negotiation = negotiate_v1(103 &mut send,104 &mut recv,105 nonce,106 vec![Capability::VideoH264],107 Some(Capability::VideoH264),108 "adaptive feedback",109 None,110 )111 .await?;112 if negotiation.session_id != media_session_id {113 return Err("Host returned mismatched media and adaptive feedback sessions".to_string());114 }115 Ok(Some((connection, send, recv)))116}117118fn adaptive_bitrate_state_name(state: AdaptiveBitrateStateV1) -> &'static str {119 match state {120 AdaptiveBitrateStateV1::Hold => "hold",121 AdaptiveBitrateStateV1::Decrease => "decrease",122 AdaptiveBitrateStateV1::Increase => "increase",123 }124}125126fn adaptive_bitrate_reason_names(reasons: AdaptiveBitrateReasonFlagsV1) -> Vec<&'static str> {127 [128 (AdaptiveBitrateReasonFlagsV1::RTT_INFLATION, "rtt-inflation"),129 (130 AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,131 "loss-or-cancellation",132 ),133 (134 AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,135 "sender-backpressure",136 ),137 (138 AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,139 "receiver-queue",140 ),141 (142 AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,143 "decode-backlog",144 ),145 (146 AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,147 "delivery-latency",148 ),149 (150 AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,151 "clean-recovery",152 ),153 (154 AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE,155 "feedback-stale",156 ),157 ]158 .into_iter()159 .filter_map(|(flag, name)| reasons.contains(flag).then_some(name))160 .collect()161}162163fn adaptive_bitrate_decision_diagnostic(164 decision: AdaptiveBitrateDecisionV1,165) -> AdaptiveBitrateDecisionDiagnostic {166 AdaptiveBitrateDecisionDiagnostic {167 decision_id: decision.decision_id,168 report_id: decision.report_id,169 target_kbps: decision.target_kbps,170 floor_kbps: decision.floor_kbps,171 ceiling_kbps: decision.ceiling_kbps,172 state: adaptive_bitrate_state_name(decision.state),173 reasons: adaptive_bitrate_reason_names(decision.reasons),174 applied: decision.applied,175 }176}177178179struct AdaptiveDecisionSequence {180 last_decision_id: Option<u64>,181 last_report_id: Option<u64>,182}183184impl AdaptiveDecisionSequence {185 fn accept(&mut self, decision: &AdaptiveBitrateDecisionV1) -> Result<(), String> {186 if self187 .last_decision_id188 .is_some_and(|previous| decision.decision_id <= previous)189 {190 return Err(format!(191 "adaptive decision ID did not increase: previous={:?}, current={}",192 self.last_decision_id, decision.decision_id193 ));194 }195 if self196 .last_report_id197 .is_some_and(|previous| decision.report_id <= previous)198 {199 return Err(format!(200 "adaptive decision report ID did not increase: previous={:?}, current={}",201 self.last_report_id, decision.report_id202 ));203 }204 self.last_decision_id = Some(decision.decision_id);205 self.last_report_id = Some(decision.report_id);206 Ok(())207 }208}209210async fn emit_paced_adaptive_decisions<F>(211 generation: u64,212 mut decisions: tokio::sync::watch::Receiver<Option<AdaptiveBitrateDecisionV1>>,213 delivery_interval: Duration,214 mut emit: F,215) -> Result<(), String>216where217 F: FnMut(AdaptiveBitrateDecisionPayload) -> Result<(), String>,218{219 let mut delivery = tokio::time::interval(delivery_interval);220 delivery.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);221 loop {222 decisions223 .changed()224 .await225 .map_err(|_| "adaptive decision receiver closed".to_string())?;226 // The first tick is immediate. Later ticks pace the webview boundary;227 // any host burst replaces the watch value while this task waits.228 delivery.tick().await;229 let Some(decision) = *decisions.borrow_and_update() else {230 continue;231 };232 emit(AdaptiveBitrateDecisionPayload {233 generation,234 decision: adaptive_bitrate_decision_diagnostic(decision),235 })?;236 }237}238239pub(crate) async fn run_media_feedback_session(240 app: AppHandle,241 generation: u64,242 mut send: iroh::endpoint::SendStream,243 mut recv: iroh::endpoint::RecvStream,244 mut reports: tokio::sync::watch::Receiver<Option<AccumulatedMediaFeedback>>,245) {246 let (decision_tx, decision_rx) = tokio::sync::watch::channel(None);247 let writer = async {248 let mut last_written = None;249 loop {250 reports251 .changed()252 .await253 .map_err(|_| "adaptive feedback sender closed".to_string())?;254 let accumulated = *reports.borrow_and_update();255 let Some(accumulated) = accumulated else {256 continue;257 };258 let report = accumulated.report_since(last_written);259 tokio::time::timeout(260 CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT,261 write_media_feedback_report_v1(&mut send, &report),262 )263 .await264 .map_err(|_| "adaptive feedback write timed out".to_string())?265 .map_err(|error| format!("adaptive feedback write failed: {error}"))?;266 last_written = Some(accumulated);267 }268 269 Ok::<(), String>(())270 };271 let reader = async {272 let mut sequence = AdaptiveDecisionSequence::default();273 loop {274 let decision = read_adaptive_bitrate_decision_v1(&mut recv)275 .await276 .map_err(|error| format!("adaptive decision read failed: {error}"))?277 .ok_or_else(|| "adaptive decision stream closed".to_string())?;278 sequence.accept(&decision)?;279 decision_tx.send_replace(Some(decision));280 }281 282 Ok::<(), String>(())283 };284 let decision_app = app.clone();285 let decision_emitter = emit_paced_adaptive_decisions(286 generation,287 decision_rx,288 CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL,289 move |payload| {290 decision_app291 .emit("adaptive-bitrate-decision", payload)292 .map_err(|error| format!("adaptive decision event delivery failed: {error}"))293 },294 );295 let terminal = tokio::select! {296 terminal = writer => terminal,297 terminal = reader => terminal,298 terminal = decision_emitter => terminal,299 };300 if let Err(error) = terminal {301 eprintln!("[client] {error}");302 if let Err(emit_error) = app.emit(303 "adaptive-feedback-state",304 serde_json::json!({305 "generation": generation,306 "available": false,307 "error": error,308 }),309 ) {310 eprintln!("[client] adaptive feedback terminal event delivery failed: {emit_error}");311 }312 }313 retire_media_feedback_generation(&app, generation).await;314}315316fn feedback_latency_ms(value: Option<f64>, name: &str) -> Result<Option<u16>, String> {317 value318 .map(|value| {319 if !value.is_finite() || !(0.0..=60_000.0).contains(&value) {320 return Err(format!("{name} must be finite and between 0 and 60000 ms"));321 }322 Ok(value.round() as u16)323 })324 .transpose()325}326327fn next_feedback_report_id(counter: &AtomicU64) -> Result<u64, String> {328 counter329 .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {330 current.checked_add(1)331 })332 .map(|previous| previous + 1)333 .map_err(|_| "Adaptive feedback report ID overflowed".to_string())334}335336fn media_feedback_report(337 report_id: u64,338 input: ClientMediaFeedbackReport,339) -> Result<MediaFeedbackReportV1, String> {340 let flags = if input.resync_active {341 MediaFeedbackFlags::RESYNC_ACTIVE342 } else {343 MediaFeedbackFlags::NONE344 };345 let report = MediaFeedbackReportV1 {346 report_id,347 interval_ms: input.interval_ms,348 flags,349 last_sequence: input.last_sequence,350 transport_dropped_delta: input.transport_dropped_delta,351 frontend_dropped_delta: input.frontend_dropped_delta,352 decoder_dropped_delta: input.decoder_dropped_delta,353 presenter_dropped_delta: input.presenter_dropped_delta,354 frontend_queue_depth: input.frontend_queue_depth,355 frontend_queue_capacity: input.frontend_queue_capacity,356 decode_queue_depth: input.decode_queue_depth,357 decode_queue_capacity: input.decode_queue_capacity,358 presenter_queue_depth: input.presenter_queue_depth,359 presenter_queue_capacity: input.presenter_queue_capacity,360 transport_delivery_p95_ms: feedback_latency_ms(361 input.transport_delivery_p95_ms,362 "delivery p95",363 )?,364 decode_p95_ms: feedback_latency_ms(input.decode_p95_ms, "decode p95")?,365 presentation_p95_ms: feedback_latency_ms(input.presentation_p95_ms, "presentation p95")?,366 };367 report368 .validate()369 .map_err(|error| format!("Invalid adaptive feedback report: {error}"))?;370 Ok(report)371}372373pub(crate) async fn send_media_feedback(374 state: &AppState,375 generation: u64,376 report: ClientMediaFeedbackReport,377) -> Result<bool, String> {378 let feedback = state.media_feedback.lock().await;379 let Some((current_generation, _connection, sender)) = feedback.as_ref() else {380 return Ok(false);381 };382 if *current_generation != generation {383 return Ok(false);384 }385 if sender.receiver_count() == 0 {386 return Err("Adaptive feedback channel closed".to_string());387 }388 let report_id = next_feedback_report_id(&state.media_feedback_report_id)?;389 let report = media_feedback_report(report_id, report)?;390 sender.send_modify(|pending| match pending {391 Some(pending) => pending.merge(report),392 None => *pending = Some(AccumulatedMediaFeedback::new(report)),393 });394 Ok(true)395}396397398mod tests {399 use super::*;400401 fn client_feedback() -> ClientMediaFeedbackReport {402 ClientMediaFeedbackReport {403 interval_ms: 1_000,404 last_sequence: Some(99),405 transport_dropped_delta: 1,406 frontend_dropped_delta: 2,407 decoder_dropped_delta: 3,408 presenter_dropped_delta: 4,409 frontend_queue_depth: 1,410 frontend_queue_capacity: 4,411 decode_queue_depth: 1,412 decode_queue_capacity: 2,413 presenter_queue_depth: 0,414 presenter_queue_capacity: 2,415 transport_delivery_p95_ms: Some(17.4),416 decode_p95_ms: Some(4.6),417 presentation_p95_ms: None,418 resync_active: true,419 }420 }421422 423 fn adaptive_feedback_conversion_is_bounded_and_protocol_valid() {424 let report = media_feedback_report(7, client_feedback()).unwrap();425 assert_eq!(report.report_id, 7);426 assert_eq!(report.interval_ms, 1_000);427 assert_eq!(report.last_sequence, Some(99));428 assert_eq!(report.transport_delivery_p95_ms, Some(17));429 assert_eq!(report.decode_p95_ms, Some(5));430 assert!(report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE));431 report.validate().unwrap();432433 let mut invalid = client_feedback();434 invalid.decode_queue_depth = 3;435 assert!(media_feedback_report(8, invalid).is_err());436 let mut invalid_interval = client_feedback();437 invalid_interval.interval_ms = 249;438 assert!(media_feedback_report(9, invalid_interval).is_err());439 assert!(feedback_latency_ms(Some(f64::INFINITY), "decode").is_err());440 assert!(feedback_latency_ms(Some(60_001.0), "decode").is_err());441 }442443 444 fn feedback_report_ids_are_nonzero_monotonic_and_checked() {445 let counter = AtomicU64::new(0);446 assert_eq!(next_feedback_report_id(&counter).unwrap(), 1);447 assert_eq!(next_feedback_report_id(&counter).unwrap(), 2);448 assert!(next_feedback_report_id(&AtomicU64::new(u64::MAX)).is_err());449 }450451 452 async fn adaptive_feedback_watch_coalesces_latest_state_without_losing_pressure() {453 let (sender, mut receiver) = tokio::sync::watch::channel(None);454 let first = media_feedback_report(1, client_feedback()).unwrap();455 sender.send_replace(Some(AccumulatedMediaFeedback::new(first)));456 receiver.changed().await.unwrap();457 let stalled_write = receiver.borrow_and_update().unwrap();458459 let mut latest = client_feedback();460 latest.interval_ms = 1_250;461 latest.transport_dropped_delta = 9;462 let second = media_feedback_report(2, latest).unwrap();463 sender.send_modify(|pending| pending.as_mut().unwrap().merge(second));464 let mut newest = client_feedback();465 newest.interval_ms = 1_500;466 newest.transport_dropped_delta = 11;467 newest.frontend_queue_depth = 3;468 let third = media_feedback_report(3, newest).unwrap();469 sender.send_modify(|pending| pending.as_mut().unwrap().merge(third));470471 assert_eq!(stalled_write.report_since(None), first);472 receiver.changed().await.unwrap();473 let coalesced = receiver474 .borrow_and_update()475 .unwrap()476 .report_since(Some(stalled_write));477 assert_eq!(coalesced.report_id, 3);478 assert_eq!(coalesced.interval_ms, 2_750);479 assert_eq!(coalesced.transport_dropped_delta, 20);480 assert_eq!(coalesced.frontend_queue_depth, 3);481 }482483 484 fn adaptive_decision_diagnostics_are_explicitly_shadow_state() {485 let diagnostic = adaptive_bitrate_decision_diagnostic(adaptive_decision(2, 1));486 assert_eq!(diagnostic.state, "decrease");487 assert_eq!(diagnostic.reasons, vec!["receiver-queue", "decode-backlog"]);488 assert!(!diagnostic.applied);489 }490491 fn adaptive_decision(decision_id: u64, report_id: u64) -> AdaptiveBitrateDecisionV1 {492 AdaptiveBitrateDecisionV1 {493 decision_id,494 report_id,495 target_kbps: 8_000,496 floor_kbps: 4_000,497 ceiling_kbps: 20_000,498 state: AdaptiveBitrateStateV1::Decrease,499 reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE500 .union(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG),501 applied: false,502 }503 }504505 506 fn adaptive_decision_sequence_rejects_duplicate_or_regressing_ids() {507 let mut sequence = AdaptiveDecisionSequence::default();508 sequence.accept(&adaptive_decision(1, 10)).unwrap();509 sequence.accept(&adaptive_decision(3, 12)).unwrap();510 assert!(sequence.accept(&adaptive_decision(3, 13)).is_err());511512 let mut report_regression = AdaptiveDecisionSequence::default();513 report_regression.accept(&adaptive_decision(1, 10)).unwrap();514 assert!(report_regression.accept(&adaptive_decision(2, 10)).is_err());515 assert!(report_regression.accept(&adaptive_decision(3, 9)).is_err());516 }517518 519 async fn adaptive_decision_delivery_is_paced_latest_value_and_emit_failure_is_terminal() {520 let (sender, receiver) = tokio::sync::watch::channel(None);521 let (delivered_tx, mut delivered_rx) = tokio::sync::mpsc::channel(2);522 let delivery = tokio::spawn(emit_paced_adaptive_decisions(523 7,524 receiver,525 Duration::from_millis(20),526 move |payload| {527 delivered_tx528 .try_send(payload)529 .map_err(|error| format!("test delivery failed: {error}"))530 },531 ));532 sender.send_replace(Some(adaptive_decision(1, 1)));533 sender.send_replace(Some(adaptive_decision(2, 2)));534 let first = tokio::time::timeout(Duration::from_millis(100), delivered_rx.recv())535 .await536 .unwrap()537 .unwrap();538 assert_eq!(first.generation, 7);539 assert_eq!(first.decision.decision_id, 2);540541 sender.send_replace(Some(adaptive_decision(3, 3)));542 assert!(543 tokio::time::timeout(Duration::from_millis(5), delivered_rx.recv())544 .await545 .is_err()546 );547 let second = tokio::time::timeout(Duration::from_millis(100), delivered_rx.recv())548 .await549 .unwrap()550 .unwrap();551 assert_eq!(second.decision.decision_id, 3);552 delivery.abort();553554 let (sender, receiver) = tokio::sync::watch::channel(None);555 let terminal = tokio::spawn(emit_paced_adaptive_decisions(556 7,557 receiver,558 Duration::from_millis(1),559 |_payload| Err("webview channel closed".to_string()),560 ));561 sender.send_replace(Some(adaptive_decision(4, 4)));562 let error = tokio::time::timeout(Duration::from_millis(100), terminal)563 .await564 .unwrap()565 .unwrap()566 .unwrap_err();567 assert_eq!(error, "webview channel closed");568 }569}