- commit
- 7770f8042e525c7e55cb05bb3c99cef312250642
- parent
- 77aacb288d396ce1fc4c85e650d20c02cc31bfcd
- author
- Dami <dami.etoile@gmail.com>
| M | src-tauri/ | 578 | +----------------------------- |
| A | src-tauri/ | 569 | ++++++++++++++++++++++++++++++ |
| M | src-tauri/ | 1 | + |
@@ -1,8 +1,13 @@
11 use super::auth::derive_iroh_secret_from_key;22 use super::enrollment::{connection_enrollment, mark_invitation_redeemed};33 use super::state::{4- AccumulatedMediaFeedback, AppState, CLIENT_INPUT_QUEUE_CAPACITY, FRAME_ALPN, INPUT_ALPN,5- MediaFeedbackSender, development_direct_node_available,4+ AppState, CLIENT_INPUT_QUEUE_CAPACITY, FRAME_ALPN, INPUT_ALPN, MediaFeedbackSender,5+ development_direct_node_available,6+};7+pub use crate::media::adaptive_feedback::ClientMediaFeedbackReport;8+use crate::media::adaptive_feedback::{9+ CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT, open_negotiated_feedback_stream,10+ run_media_feedback_session, send_media_feedback,611 };712 use crate::media::audio_delivery::{813 self, AudioStartRequest, lock_audio_deliveries, next_audio_generation, try_start_audio,@@ -11,8 +16,8 @@ pub use crate::media::frame_channel::FramePayload;
1116 use crate::media::frame_channel::{1217 FrameEnvelopeMetadata, byte_to_codec, close_generation_connection, emit_frame_error,1318 encode_frame_envelope, next_media_generation, release_frame_channel_slot,14- release_frame_channel_slot_for_generation, take_generation_owned_triple,15- try_reserve_frame_channel_slot, validate_legacy_media_header,19+ release_frame_channel_slot_for_generation, try_reserve_frame_channel_slot,20+ validate_legacy_media_header,1621 };1722 use crate::media::metrics::{ClientMediaMetrics, lock_client_media_metrics};1823 use crate::media::moq_receiver::{@@ -24,30 +29,29 @@ use crate::media::object_receiver::{
2429 MediaObjectReceiver, MediaObjectSequence, MediaObjectSequenceDecision,2530 };2631 use crate::media::transport::{27- CLIENT_ENDPOINT_CLOSE_TIMEOUT, MediaTransport, NegotiatedV1Stream,28- connect_error_is_unsupported_alpn, negotiate_v1, open_negotiated_media_stream,32+ CLIENT_ENDPOINT_CLOSE_TIMEOUT, MediaTransport, NegotiatedV1Stream, negotiate_v1,33+ open_negotiated_media_stream,2934 };3035 use base64::Engine;3136 use iroh::{Endpoint, SecretKey, endpoint::presets};3237 use openh264::{formats::YUVSource, nal_units};33-use serde::{Deserialize, Serialize};38+use serde::Serialize;3439 use sigil_protocol::{35- AdaptiveBitrateDecisionV1, AdaptiveBitrateReasonFlagsV1, AdaptiveBitrateStateV1, Capability,36- FrameFlags, INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3,37- MAX_MEDIA_GROUP_BYTES_V3, MEDIA_FEEDBACK_ALPN_V1, MediaCodec, MediaControlRequestV3,38- MediaFeedbackFlags, MediaFeedbackReportV1, MediaObjectV3, PointerPosition,40+ Capability, FrameFlags, INPUT_ALPN_V1, InputEvent, InvitationGrants, KeyframeRequestReasonV3,41+ MAX_MEDIA_GROUP_BYTES_V3, MediaCodec, MediaControlRequestV3, MediaObjectV3, PointerPosition,3942 PointerSurfaceDimensions, ProtocolError, RELATIVE_POINTER_DELTA_MAX,40- RELATIVE_POINTER_DELTA_MIN, read_adaptive_bitrate_decision_v1, read_input_ack,41- read_media_frame, read_media_object_v3, write_input_event, write_media_control_request_v3,42- write_media_feedback_report_v1,43+ RELATIVE_POINTER_DELTA_MIN, read_input_ack, read_media_frame, read_media_object_v3,44+ write_input_event, write_media_control_request_v3,4345 };4446 use std::collections::BTreeMap;4547 use std::io::Cursor;46-use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};48+49+use std::sync::atomic::AtomicU64;50+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};4751 use std::sync::{Arc, Mutex as StdMutex, MutexGuard as StdMutexGuard};4852 use std::time::{Duration, Instant};4953 use tauri::{50- AppHandle, Emitter, Manager, State,54+ AppHandle, Emitter, State,5155 ipc::{Channel, Response},5256 };5357 @@ -55,9 +59,6 @@ use tauri::{
5559 5660 const LEGACY_MEDIA_IDLE_TIMEOUT: Duration = Duration::from_secs(10);5761 const CLIENT_FRAME_STATS_INTERVAL: Duration = Duration::from_millis(250);58-const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);59-const CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT: Duration = Duration::from_secs(5);60-const CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL: Duration = Duration::from_secs(1);6162 6263 fn lock_network_diagnostics(6364 diagnostics: &StdMutex<NetworkSessionDiagnostics>,@@ -67,20 +68,6 @@ fn lock_network_diagnostics(
6768 .unwrap_or_else(std::sync::PoisonError::into_inner)6869 }6970 70-async fn retire_media_feedback_generation(app: &AppHandle, generation: u64) -> bool {71- let state = app.state::<AppState>();72- let connection = {73- let _connection_serial = state.client_connection_serial.lock().await;74- let mut slot = state.media_feedback.lock().await;75- take_generation_owned_triple(&mut slot, generation).map(|(connection, _)| connection)76- };77- let Some(connection) = connection else {78- return false;79- };80- connection.close(0_u32.into(), b"adaptive feedback ended");81- true82-}83-8471 struct ClientConnectGuard {8572 active: Arc<AtomicBool>,8673 committed: bool,@@ -133,44 +120,6 @@ pub struct ConnectResult {
133120 pub error: Option<String>,134121 }135122 136-137-pub struct ClientMediaFeedbackReport {138- pub interval_ms: u16,139- pub last_sequence: Option<u64>,140- pub transport_dropped_delta: u32,141- pub frontend_dropped_delta: u32,142- pub decoder_dropped_delta: u32,143- pub presenter_dropped_delta: u32,144- pub frontend_queue_depth: u8,145- pub frontend_queue_capacity: u8,146- pub decode_queue_depth: u8,147- pub decode_queue_capacity: u8,148- pub presenter_queue_depth: u8,149- pub presenter_queue_capacity: u8,150- pub transport_delivery_p95_ms: Option<f64>,151- pub decode_p95_ms: Option<f64>,152- pub presentation_p95_ms: Option<f64>,153- pub resync_active: bool,154-}155-156-157-struct AdaptiveBitrateDecisionPayload {158- generation: u64,159- decision: AdaptiveBitrateDecisionDiagnostic,160-}161-162-163-struct AdaptiveBitrateDecisionDiagnostic {164- decision_id: u64,165- report_id: u64,166- target_kbps: u32,167- floor_kbps: u32,168- ceiling_kbps: u32,169- state: &'static str,170- reasons: Vec<&'static str>,171- applied: bool,172-}173-174123 175124 enum MediaObjectReadOutcomeV3 {176125 Object {@@ -579,249 +528,6 @@ async fn open_negotiated_input_stream(
579528 Ok((connection, send, recv, negotiation))580529 }581530 582-async fn open_negotiated_feedback_stream(583- endpoint: &Endpoint,584- address: &iroh::EndpointAddr,585- nonce: [u8; 16],586- media_session_id: u64,587-) -> Result<588- Option<(589- iroh::endpoint::Connection,590- iroh::endpoint::SendStream,591- iroh::endpoint::RecvStream,592- )>,593- String,594-> {595- let connection = match endpoint596- .connect(address.clone(), MEDIA_FEEDBACK_ALPN_V1)597- .await598- {599- Ok(connection) => connection,600- Err(error) if connect_error_is_unsupported_alpn(&error) => return Ok(None),601- Err(error) => {602- return Err(format!(603- "Failed to connect adaptive feedback stream: {error}"604- ));605- }606- };607- let (mut send, mut recv) = connection608- .open_bi()609- .await610- .map_err(|error| format!("Failed to open adaptive feedback handshake: {error}"))?;611- let negotiation = negotiate_v1(612- &mut send,613- &mut recv,614- nonce,615- vec![Capability::VideoH264],616- Some(Capability::VideoH264),617- "adaptive feedback",618- None,619- )620- .await?;621- if negotiation.session_id != media_session_id {622- return Err("Host returned mismatched media and adaptive feedback sessions".to_string());623- }624- Ok(Some((connection, send, recv)))625-}626-627-fn adaptive_bitrate_state_name(state: AdaptiveBitrateStateV1) -> &'static str {628- match state {629- AdaptiveBitrateStateV1::Hold => "hold",630- AdaptiveBitrateStateV1::Decrease => "decrease",631- AdaptiveBitrateStateV1::Increase => "increase",632- }633-}634-635-fn adaptive_bitrate_reason_names(reasons: AdaptiveBitrateReasonFlagsV1) -> Vec<&'static str> {636- [637- (AdaptiveBitrateReasonFlagsV1::RTT_INFLATION, "rtt-inflation"),638- (639- AdaptiveBitrateReasonFlagsV1::LOSS_OR_CANCELLATION,640- "loss-or-cancellation",641- ),642- (643- AdaptiveBitrateReasonFlagsV1::SENDER_BACKPRESSURE,644- "sender-backpressure",645- ),646- (647- AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE,648- "receiver-queue",649- ),650- (651- AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG,652- "decode-backlog",653- ),654- (655- AdaptiveBitrateReasonFlagsV1::DELIVERY_LATENCY,656- "delivery-latency",657- ),658- (659- AdaptiveBitrateReasonFlagsV1::CLEAN_RECOVERY,660- "clean-recovery",661- ),662- (663- AdaptiveBitrateReasonFlagsV1::FEEDBACK_STALE,664- "feedback-stale",665- ),666- ]667- .into_iter()668- .filter_map(|(flag, name)| reasons.contains(flag).then_some(name))669- .collect()670-}671-672-fn adaptive_bitrate_decision_diagnostic(673- decision: AdaptiveBitrateDecisionV1,674-) -> AdaptiveBitrateDecisionDiagnostic {675- AdaptiveBitrateDecisionDiagnostic {676- decision_id: decision.decision_id,677- report_id: decision.report_id,678- target_kbps: decision.target_kbps,679- floor_kbps: decision.floor_kbps,680- ceiling_kbps: decision.ceiling_kbps,681- state: adaptive_bitrate_state_name(decision.state),682- reasons: adaptive_bitrate_reason_names(decision.reasons),683- applied: decision.applied,684- }685-}686-687-688-struct AdaptiveDecisionSequence {689- last_decision_id: Option<u64>,690- last_report_id: Option<u64>,691-}692-693-impl AdaptiveDecisionSequence {694- fn accept(&mut self, decision: &AdaptiveBitrateDecisionV1) -> Result<(), String> {695- if self696- .last_decision_id697- .is_some_and(|previous| decision.decision_id <= previous)698- {699- return Err(format!(700- "adaptive decision ID did not increase: previous={:?}, current={}",701- self.last_decision_id, decision.decision_id702- ));703- }704- if self705- .last_report_id706- .is_some_and(|previous| decision.report_id <= previous)707- {708- return Err(format!(709- "adaptive decision report ID did not increase: previous={:?}, current={}",710- self.last_report_id, decision.report_id711- ));712- }713- self.last_decision_id = Some(decision.decision_id);714- self.last_report_id = Some(decision.report_id);715- Ok(())716- }717-}718-719-async fn emit_paced_adaptive_decisions<F>(720- generation: u64,721- mut decisions: tokio::sync::watch::Receiver<Option<AdaptiveBitrateDecisionV1>>,722- delivery_interval: Duration,723- mut emit: F,724-) -> Result<(), String>725-where726- F: FnMut(AdaptiveBitrateDecisionPayload) -> Result<(), String>,727-{728- let mut delivery = tokio::time::interval(delivery_interval);729- delivery.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);730- loop {731- decisions732- .changed()733- .await734- .map_err(|_| "adaptive decision receiver closed".to_string())?;735- // The first tick is immediate. Later ticks pace the webview boundary;736- // any host burst replaces the watch value while this task waits.737- delivery.tick().await;738- let Some(decision) = *decisions.borrow_and_update() else {739- continue;740- };741- emit(AdaptiveBitrateDecisionPayload {742- generation,743- decision: adaptive_bitrate_decision_diagnostic(decision),744- })?;745- }746-}747-748-async fn run_media_feedback_session(749- app: AppHandle,750- generation: u64,751- mut send: iroh::endpoint::SendStream,752- mut recv: iroh::endpoint::RecvStream,753- mut reports: tokio::sync::watch::Receiver<Option<AccumulatedMediaFeedback>>,754-) {755- let (decision_tx, decision_rx) = tokio::sync::watch::channel(None);756- let writer = async {757- let mut last_written = None;758- loop {759- reports760- .changed()761- .await762- .map_err(|_| "adaptive feedback sender closed".to_string())?;763- let accumulated = *reports.borrow_and_update();764- let Some(accumulated) = accumulated else {765- continue;766- };767- let report = accumulated.report_since(last_written);768- tokio::time::timeout(769- CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT,770- write_media_feedback_report_v1(&mut send, &report),771- )772- .await773- .map_err(|_| "adaptive feedback write timed out".to_string())?774- .map_err(|error| format!("adaptive feedback write failed: {error}"))?;775- last_written = Some(accumulated);776- }777- 778- Ok::<(), String>(())779- };780- let reader = async {781- let mut sequence = AdaptiveDecisionSequence::default();782- loop {783- let decision = read_adaptive_bitrate_decision_v1(&mut recv)784- .await785- .map_err(|error| format!("adaptive decision read failed: {error}"))?786- .ok_or_else(|| "adaptive decision stream closed".to_string())?;787- sequence.accept(&decision)?;788- decision_tx.send_replace(Some(decision));789- }790- 791- Ok::<(), String>(())792- };793- let decision_app = app.clone();794- let decision_emitter = emit_paced_adaptive_decisions(795- generation,796- decision_rx,797- CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL,798- move |payload| {799- decision_app800- .emit("adaptive-bitrate-decision", payload)801- .map_err(|error| format!("adaptive decision event delivery failed: {error}"))802- },803- );804- let terminal = tokio::select! {805- terminal = writer => terminal,806- terminal = reader => terminal,807- terminal = decision_emitter => terminal,808- };809- if let Err(error) = terminal {810- eprintln!("[client] {error}");811- if let Err(emit_error) = app.emit(812- "adaptive-feedback-state",813- serde_json::json!({814- "generation": generation,815- "available": false,816- "error": error,817- }),818- ) {819- eprintln!("[client] adaptive feedback terminal event delivery failed: {emit_error}");820- }821- }822- retire_media_feedback_generation(&app, generation).await;823-}824-825531 async fn run_media_control_writer_v3(826532 mut stream: iroh::endpoint::SendStream,827533 mut requests: tokio::sync::mpsc::Receiver<(KeyframeRequestReasonV3, Option<u64>)>,@@ -2283,175 +1989,6 @@ mod tests {
22831989 assert!(next_media_generation(&exhausted).is_err());22841990 }22851991 2286- fn client_feedback() -> ClientMediaFeedbackReport {2287- ClientMediaFeedbackReport {2288- interval_ms: 1_000,2289- last_sequence: Some(99),2290- transport_dropped_delta: 1,2291- frontend_dropped_delta: 2,2292- decoder_dropped_delta: 3,2293- presenter_dropped_delta: 4,2294- frontend_queue_depth: 1,2295- frontend_queue_capacity: 4,2296- decode_queue_depth: 1,2297- decode_queue_capacity: 2,2298- presenter_queue_depth: 0,2299- presenter_queue_capacity: 2,2300- transport_delivery_p95_ms: Some(17.4),2301- decode_p95_ms: Some(4.6),2302- presentation_p95_ms: None,2303- resync_active: true,2304- }2305- }2306-2307- 2308- fn adaptive_feedback_conversion_is_bounded_and_protocol_valid() {2309- let report = media_feedback_report(7, client_feedback()).unwrap();2310- assert_eq!(report.report_id, 7);2311- assert_eq!(report.interval_ms, 1_000);2312- assert_eq!(report.last_sequence, Some(99));2313- assert_eq!(report.transport_delivery_p95_ms, Some(17));2314- assert_eq!(report.decode_p95_ms, Some(5));2315- assert!(report.flags.contains(MediaFeedbackFlags::RESYNC_ACTIVE));2316- report.validate().unwrap();2317-2318- let mut invalid = client_feedback();2319- invalid.decode_queue_depth = 3;2320- assert!(media_feedback_report(8, invalid).is_err());2321- let mut invalid_interval = client_feedback();2322- invalid_interval.interval_ms = 249;2323- assert!(media_feedback_report(9, invalid_interval).is_err());2324- assert!(feedback_latency_ms(Some(f64::INFINITY), "decode").is_err());2325- assert!(feedback_latency_ms(Some(60_001.0), "decode").is_err());2326- }2327-2328- 2329- fn feedback_report_ids_are_nonzero_monotonic_and_checked() {2330- let counter = AtomicU64::new(0);2331- assert_eq!(next_feedback_report_id(&counter).unwrap(), 1);2332- assert_eq!(next_feedback_report_id(&counter).unwrap(), 2);2333- assert!(next_feedback_report_id(&AtomicU64::new(u64::MAX)).is_err());2334- }2335-2336- 2337- async fn adaptive_feedback_watch_coalesces_latest_state_without_losing_pressure() {2338- let (sender, mut receiver) = tokio::sync::watch::channel(None);2339- let first = media_feedback_report(1, client_feedback()).unwrap();2340- sender.send_replace(Some(AccumulatedMediaFeedback::new(first)));2341- receiver.changed().await.unwrap();2342- let stalled_write = receiver.borrow_and_update().unwrap();2343-2344- let mut latest = client_feedback();2345- latest.interval_ms = 1_250;2346- latest.transport_dropped_delta = 9;2347- let second = media_feedback_report(2, latest).unwrap();2348- sender.send_modify(|pending| pending.as_mut().unwrap().merge(second));2349- let mut newest = client_feedback();2350- newest.interval_ms = 1_500;2351- newest.transport_dropped_delta = 11;2352- newest.frontend_queue_depth = 3;2353- let third = media_feedback_report(3, newest).unwrap();2354- sender.send_modify(|pending| pending.as_mut().unwrap().merge(third));2355-2356- assert_eq!(stalled_write.report_since(None), first);2357- receiver.changed().await.unwrap();2358- let coalesced = receiver2359- .borrow_and_update()2360- .unwrap()2361- .report_since(Some(stalled_write));2362- assert_eq!(coalesced.report_id, 3);2363- assert_eq!(coalesced.interval_ms, 2_750);2364- assert_eq!(coalesced.transport_dropped_delta, 20);2365- assert_eq!(coalesced.frontend_queue_depth, 3);2366- }2367-2368- 2369- fn adaptive_decision_diagnostics_are_explicitly_shadow_state() {2370- let diagnostic = adaptive_bitrate_decision_diagnostic(adaptive_decision(2, 1));2371- assert_eq!(diagnostic.state, "decrease");2372- assert_eq!(diagnostic.reasons, vec!["receiver-queue", "decode-backlog"]);2373- assert!(!diagnostic.applied);2374- }2375-2376- fn adaptive_decision(decision_id: u64, report_id: u64) -> AdaptiveBitrateDecisionV1 {2377- AdaptiveBitrateDecisionV1 {2378- decision_id,2379- report_id,2380- target_kbps: 8_000,2381- floor_kbps: 4_000,2382- ceiling_kbps: 20_000,2383- state: AdaptiveBitrateStateV1::Decrease,2384- reasons: AdaptiveBitrateReasonFlagsV1::RECEIVER_QUEUE2385- .union(AdaptiveBitrateReasonFlagsV1::DECODE_BACKLOG),2386- applied: false,2387- }2388- }2389-2390- 2391- fn adaptive_decision_sequence_rejects_duplicate_or_regressing_ids() {2392- let mut sequence = AdaptiveDecisionSequence::default();2393- sequence.accept(&adaptive_decision(1, 10)).unwrap();2394- sequence.accept(&adaptive_decision(3, 12)).unwrap();2395- assert!(sequence.accept(&adaptive_decision(3, 13)).is_err());2396-2397- let mut report_regression = AdaptiveDecisionSequence::default();2398- report_regression.accept(&adaptive_decision(1, 10)).unwrap();2399- assert!(report_regression.accept(&adaptive_decision(2, 10)).is_err());2400- assert!(report_regression.accept(&adaptive_decision(3, 9)).is_err());2401- }2402-2403- 2404- async fn adaptive_decision_delivery_is_paced_latest_value_and_emit_failure_is_terminal() {2405- let (sender, receiver) = tokio::sync::watch::channel(None);2406- let (delivered_tx, mut delivered_rx) = tokio::sync::mpsc::channel(2);2407- let delivery = tokio::spawn(emit_paced_adaptive_decisions(2408- 7,2409- receiver,2410- Duration::from_millis(20),2411- move |payload| {2412- delivered_tx2413- .try_send(payload)2414- .map_err(|error| format!("test delivery failed: {error}"))2415- },2416- ));2417- sender.send_replace(Some(adaptive_decision(1, 1)));2418- sender.send_replace(Some(adaptive_decision(2, 2)));2419- let first = tokio::time::timeout(Duration::from_millis(100), delivered_rx.recv())2420- .await2421- .unwrap()2422- .unwrap();2423- assert_eq!(first.generation, 7);2424- assert_eq!(first.decision.decision_id, 2);2425-2426- sender.send_replace(Some(adaptive_decision(3, 3)));2427- assert!(2428- tokio::time::timeout(Duration::from_millis(5), delivered_rx.recv())2429- .await2430- .is_err()2431- );2432- let second = tokio::time::timeout(Duration::from_millis(100), delivered_rx.recv())2433- .await2434- .unwrap()2435- .unwrap();2436- assert_eq!(second.decision.decision_id, 3);2437- delivery.abort();2438-2439- let (sender, receiver) = tokio::sync::watch::channel(None);2440- let terminal = tokio::spawn(emit_paced_adaptive_decisions(2441- 7,2442- receiver,2443- Duration::from_millis(1),2444- |_payload| Err("webview channel closed".to_string()),2445- ));2446- sender.send_replace(Some(adaptive_decision(4, 4)));2447- let error = tokio::time::timeout(Duration::from_millis(100), terminal)2448- .await2449- .unwrap()2450- .unwrap()2451- .unwrap_err();2452- assert_eq!(error, "webview channel closed");2453- }2454-24551992 24561993 fn client_connect_guard_rejects_overlap_and_resets_only_failed_attempts() {24571994 let active = Arc::new(AtomicBool::new(false));@@ -2891,86 +2428,13 @@ pub async fn iroh_client_request_keyframe(
28912428 }28922429 }28932430 2894-fn feedback_latency_ms(value: Option<f64>, name: &str) -> Result<Option<u16>, String> {2895- value2896- .map(|value| {2897- if !value.is_finite() || !(0.0..=60_000.0).contains(&value) {2898- return Err(format!("{name} must be finite and between 0 and 60000 ms"));2899- }2900- Ok(value.round() as u16)2901- })2902- .transpose()2903-}2904-2905-fn next_feedback_report_id(counter: &AtomicU64) -> Result<u64, String> {2906- counter2907- .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {2908- current.checked_add(1)2909- })2910- .map(|previous| previous + 1)2911- .map_err(|_| "Adaptive feedback report ID overflowed".to_string())2912-}2913-2914-fn media_feedback_report(2915- report_id: u64,2916- input: ClientMediaFeedbackReport,2917-) -> Result<MediaFeedbackReportV1, String> {2918- let flags = if input.resync_active {2919- MediaFeedbackFlags::RESYNC_ACTIVE2920- } else {2921- MediaFeedbackFlags::NONE2922- };2923- let report = MediaFeedbackReportV1 {2924- report_id,2925- interval_ms: input.interval_ms,2926- flags,2927- last_sequence: input.last_sequence,2928- transport_dropped_delta: input.transport_dropped_delta,2929- frontend_dropped_delta: input.frontend_dropped_delta,2930- decoder_dropped_delta: input.decoder_dropped_delta,2931- presenter_dropped_delta: input.presenter_dropped_delta,2932- frontend_queue_depth: input.frontend_queue_depth,2933- frontend_queue_capacity: input.frontend_queue_capacity,2934- decode_queue_depth: input.decode_queue_depth,2935- decode_queue_capacity: input.decode_queue_capacity,2936- presenter_queue_depth: input.presenter_queue_depth,2937- presenter_queue_capacity: input.presenter_queue_capacity,2938- transport_delivery_p95_ms: feedback_latency_ms(2939- input.transport_delivery_p95_ms,2940- "delivery p95",2941- )?,2942- decode_p95_ms: feedback_latency_ms(input.decode_p95_ms, "decode p95")?,2943- presentation_p95_ms: feedback_latency_ms(input.presentation_p95_ms, "presentation p95")?,2944- };2945- report2946- .validate()2947- .map_err(|error| format!("Invalid adaptive feedback report: {error}"))?;2948- Ok(report)2949-}2950-29512431 29522432 pub async fn iroh_client_send_media_feedback(29532433 state: State<'_, AppState>,29542434 generation: u64,29552435 report: ClientMediaFeedbackReport,29562436 ) -> Result<bool, String> {2957- let feedback = state.media_feedback.lock().await;2958- let Some((current_generation, _connection, sender)) = feedback.as_ref() else {2959- return Ok(false);2960- };2961- if *current_generation != generation {2962- return Ok(false);2963- }2964- if sender.receiver_count() == 0 {2965- return Err("Adaptive feedback channel closed".to_string());2966- }2967- let report_id = next_feedback_report_id(&state.media_feedback_report_id)?;2968- let report = media_feedback_report(report_id, report)?;2969- sender.send_modify(|pending| match pending {2970- Some(pending) => pending.merge(report),2971- None => *pending = Some(AccumulatedMediaFeedback::new(report)),2972- });2973- Ok(true)2437+ send_media_feedback(&state, generation, report).await29742438 }29752439 29762440 @@ -0,0 +1,569 @@
1+use std::sync::atomic::{AtomicU64, Ordering};2+use std::time::Duration;3+4+use iroh::Endpoint;5+use serde::{Deserialize, Serialize};6+use 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+};11+use tauri::{AppHandle, Emitter, Manager};12+13+use crate::commands::state::{AccumulatedMediaFeedback, AppState};14+use crate::media::frame_channel::take_generation_owned_triple;15+use crate::media::transport::{connect_error_is_unsupported_alpn, negotiate_v1};16+17+pub(crate) const CLIENT_MEDIA_FEEDBACK_CONNECT_TIMEOUT: Duration = Duration::from_secs(3);18+const CLIENT_MEDIA_FEEDBACK_IO_TIMEOUT: Duration = Duration::from_secs(5);19+const CLIENT_ADAPTIVE_DECISION_DELIVERY_INTERVAL: Duration = Duration::from_secs(1);20+21+async 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+}34+35+36+pub 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+}54+55+56+struct AdaptiveBitrateDecisionPayload {57+ generation: u64,58+ decision: AdaptiveBitrateDecisionDiagnostic,59+}60+61+62+struct 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+}72+73+pub(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+}117+118+fn 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+}125+126+fn 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+}162+163+fn 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+}177+178+179+struct AdaptiveDecisionSequence {180+ last_decision_id: Option<u64>,181+ last_report_id: Option<u64>,182+}183+184+impl 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+}209+210+async 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>216+where217+ 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+}238+239+pub(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+}315+316+fn 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+}326+327+fn 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+}335+336+fn 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+}372+373+pub(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+}396+397+398+mod tests {399+ use super::*;400+401+ 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+ }421+422+ 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();432+433+ 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+ }442+443+ 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+ }450+451+ 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();458+459+ 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));470+471+ 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+ }482+483+ 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+ }490+491+ 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+ }504+505+ 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());511+512+ 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+ }517+518+ 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);540+541+ 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();553+554+ 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+}